286 lines
9.9 KiB
Go
286 lines
9.9 KiB
Go
package service
|
|
|
|
import (
|
|
"errors"
|
|
"strings"
|
|
"time"
|
|
|
|
"browser.local/platform/domain"
|
|
"browser.local/platform/repo"
|
|
"browser.local/platform/validator"
|
|
)
|
|
|
|
const defaultLogQueryLimit = 100
|
|
|
|
func (svc *CoreService) IngestLogBatch(batch domain.LogBatchIngest) (domain.LogBatchIngestResult, error) {
|
|
batch = domain.CopyLogBatchIngest(batch)
|
|
if err := validator.ValidateLogBatchIngest(batch); err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
if err := svc.validateRunSession(batch.RunEndpointID, batch.SessionToken); err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
lock := svc.logIngestLock(batch.ServerInstanceID)
|
|
lock.Lock()
|
|
locked := true
|
|
defer func() {
|
|
if locked {
|
|
lock.Unlock()
|
|
}
|
|
}()
|
|
|
|
stamp := svc.now()
|
|
stream, err := svc.store.LogStreams().Get(batch.LogStreamID)
|
|
if errors.Is(err, repo.ErrNotFound) {
|
|
if repairErr := svc.ensureLogStreamForBatch(batch, stamp); repairErr == nil {
|
|
stream, err = svc.store.LogStreams().Get(batch.LogStreamID)
|
|
}
|
|
}
|
|
if err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
if err := validateLogBatchStream(batch, stream); err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
|
|
if batch.LastSeq <= stream.LatestSeq {
|
|
record, exists, err := svc.logStore.GetBatch(batch.LogStreamID, batch.FirstSeq)
|
|
if err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
if exists && record.LastSeq == batch.LastSeq && logBatchRecordMatches(record, batch) {
|
|
locked = false
|
|
lock.Unlock()
|
|
if err := svc.projectPluginLogBatch(stream, storedLogEntries(batch.Entries)); err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
return domain.LogBatchIngestResult{
|
|
Accepted: true,
|
|
LogStreamID: batch.LogStreamID,
|
|
AcceptedFrom: batch.FirstSeq,
|
|
AcceptedTo: batch.LastSeq,
|
|
LatestSeq: stream.LatestSeq,
|
|
Duplicate: true,
|
|
ServerTime: stamp,
|
|
}, nil
|
|
}
|
|
return domain.LogBatchIngestResult{}, validationError("log batch conflicts with acknowledged range")
|
|
}
|
|
if stream.LatestSeq > 0 && batch.FirstSeq != stream.LatestSeq+1 {
|
|
return domain.LogBatchIngestResult{}, validationError("log batch firstSeq must follow latest acknowledged sequence")
|
|
}
|
|
|
|
storedBatch := domain.CopyLogBatchIngest(batch)
|
|
sanitizeLogNetworkFields(&storedBatch)
|
|
record := domain.CopyLogBatchRecord(domain.LogBatchRecord{
|
|
Checksum: batch.Checksum,
|
|
FirstSeq: batch.FirstSeq,
|
|
LastSeq: batch.LastSeq,
|
|
Entries: storedBatch.Entries,
|
|
})
|
|
if err := svc.logStore.AppendBatch(batch.LogStreamID, record); err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
stream.LatestSeq = batch.LastSeq
|
|
stream.UpdatedAt = stamp
|
|
if err := svc.store.LogStreams().Update(stream); err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
locked = false
|
|
lock.Unlock()
|
|
if err := svc.projectPluginLogBatch(stream, storedBatch.Entries); err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
svc.publishLogEvents(stream, storedBatch.Entries)
|
|
return domain.LogBatchIngestResult{
|
|
Accepted: true,
|
|
LogStreamID: batch.LogStreamID,
|
|
AcceptedFrom: batch.FirstSeq,
|
|
AcceptedTo: batch.LastSeq,
|
|
LatestSeq: stream.LatestSeq,
|
|
ServerTime: stamp,
|
|
}, nil
|
|
}
|
|
|
|
func storedLogEntries(entries []domain.LogEntry) []domain.LogEntry {
|
|
stored := domain.CopyLogEntries(entries)
|
|
batch := domain.LogBatchIngest{Entries: stored}
|
|
sanitizeLogNetworkFields(&batch)
|
|
return batch.Entries
|
|
}
|
|
|
|
func (svc *CoreService) ensureJobLogStreamForBatch(batch domain.LogBatchIngest, stamp time.Time) error {
|
|
jobID, ok := jobIDFromLogBatch(batch)
|
|
if !ok {
|
|
return repo.ErrNotFound
|
|
}
|
|
job, err := svc.store.Jobs().Get(jobID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if job.ServerInstanceID != batch.ServerInstanceID || job.RunEndpointID != batch.RunEndpointID {
|
|
return validationError("log batch job scope does not match stream")
|
|
}
|
|
return svc.ensureJobLogStreamsUnlocked(job, stamp)
|
|
}
|
|
|
|
func (svc *CoreService) ensureLogStreamForBatch(batch domain.LogBatchIngest, stamp time.Time) error {
|
|
jobID, hasJobID := jobIDFromLogBatch(batch)
|
|
if hasJobID {
|
|
job, err := svc.store.Jobs().Get(jobID)
|
|
if err == nil {
|
|
if job.ServerInstanceID != batch.ServerInstanceID || job.RunEndpointID != batch.RunEndpointID {
|
|
return validationError("log batch job scope does not match stream")
|
|
}
|
|
return svc.ensureJobLogStreamsUnlocked(job, stamp)
|
|
}
|
|
if !errors.Is(err, repo.ErrNotFound) || !strings.HasPrefix(jobID, "autonomous-") {
|
|
return err
|
|
}
|
|
}
|
|
return svc.ensureRunLogStreamForBatch(batch, stamp)
|
|
}
|
|
|
|
func (svc *CoreService) ensureRunLogStreamForBatch(batch domain.LogBatchIngest, stamp time.Time) error {
|
|
if batch.Source != domain.LogStreamSourceProcess && batch.Source != domain.LogStreamSourceFile && batch.Source != domain.LogStreamSourceManagementProgram {
|
|
return repo.ErrNotFound
|
|
}
|
|
expectedStreamID := runLogStreamID(batch.RunEndpointID, batch.ServerInstanceID, batch.StreamKey)
|
|
if batch.LogSessionID != "" {
|
|
expectedStreamID = runSessionLogStreamID(batch.RunEndpointID, batch.ServerInstanceID, batch.LogSessionID, batch.StreamKey)
|
|
}
|
|
if batch.LogStreamID != expectedStreamID {
|
|
if batch.LogSessionID != "" || !legacyAutonomousLogStream(batch) {
|
|
return repo.ErrNotFound
|
|
}
|
|
}
|
|
instance, err := svc.store.ServerInstances().Get(batch.ServerInstanceID)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if instance.RunEndpointID != batch.RunEndpointID {
|
|
return validationError("server instance run endpoint must match log batch endpoint")
|
|
}
|
|
_, err = svc.createLogStream(domain.LogStream{ID: batch.LogStreamID, ServerInstanceID: batch.ServerInstanceID, Source: batch.Source, StreamKey: batch.StreamKey, LogSessionID: batch.LogSessionID, SessionStartedAt: batch.SessionStartedAt, StorageBackend: domain.LogStorageBackendLocalSegments, RetentionPolicy: "default", CreatedAt: stamp, UpdatedAt: stamp})
|
|
if errors.Is(err, repo.ErrDuplicate) {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
|
|
func legacyAutonomousLogStream(batch domain.LogBatchIngest) bool {
|
|
jobID, ok := jobIDFromLogBatch(batch)
|
|
return ok && strings.HasPrefix(jobID, "autonomous-")
|
|
}
|
|
|
|
func jobIDFromLogBatch(batch domain.LogBatchIngest) (string, bool) {
|
|
streamKey := strings.TrimSpace(batch.StreamKey)
|
|
if streamKey == "" || !strings.HasPrefix(batch.LogStreamID, "job.") {
|
|
return "", false
|
|
}
|
|
suffix := "." + streamKey
|
|
body := strings.TrimPrefix(batch.LogStreamID, "job.")
|
|
if !strings.HasSuffix(body, suffix) {
|
|
return "", false
|
|
}
|
|
jobID := strings.TrimSuffix(body, suffix)
|
|
return jobID, strings.TrimSpace(jobID) != ""
|
|
}
|
|
|
|
func logBatchRecordMatches(record domain.LogBatchRecord, batch domain.LogBatchIngest) bool {
|
|
if record.Checksum == batch.Checksum {
|
|
return true
|
|
}
|
|
if len(record.Entries) == 1 && len(batch.Entries) == 1 {
|
|
return batch.Checksum == validator.LogLineChecksum(record.Entries[0].Line)
|
|
}
|
|
return false
|
|
}
|
|
|
|
// sanitizeLogNetworkFields removes raw network material before the durable log body is written.
|
|
func sanitizeLogNetworkFields(batch *domain.LogBatchIngest) {
|
|
for index := range batch.Entries {
|
|
fields := batch.Entries[index].Fields
|
|
if fields == nil {
|
|
continue
|
|
}
|
|
delete(fields, "networkFingerprint")
|
|
delete(fields, "ip")
|
|
delete(fields, "ipAddress")
|
|
}
|
|
}
|
|
|
|
func (svc *CoreService) QueryLogStream(query domain.LogStreamCursorQuery) (domain.LogStreamCursorResult, error) {
|
|
if err := validator.ValidateLogStreamCursorQuery(query); err != nil {
|
|
return domain.LogStreamCursorResult{}, err
|
|
}
|
|
stream, err := svc.store.LogStreams().Get(query.LogStreamID)
|
|
if err != nil {
|
|
return domain.LogStreamCursorResult{}, err
|
|
}
|
|
limit := query.Limit
|
|
if limit == 0 {
|
|
limit = defaultLogQueryLimit
|
|
}
|
|
|
|
selected, nextSeq, err := svc.logStore.Query(query.LogStreamID, query.AfterSeq, limit)
|
|
if err != nil {
|
|
return domain.LogStreamCursorResult{}, err
|
|
}
|
|
return domain.CopyLogStreamCursorResult(domain.LogStreamCursorResult{
|
|
LogStreamID: query.LogStreamID,
|
|
Entries: selected,
|
|
NextSeq: nextSeq,
|
|
LatestSeq: stream.LatestSeq,
|
|
}), nil
|
|
}
|
|
|
|
func (svc *CoreService) GetRunLogStreamProgress(request domain.RunLogStreamProgress) (domain.RunLogStreamProgressResult, error) {
|
|
if err := validator.ValidateRunLogStreamProgress(request); err != nil {
|
|
return domain.RunLogStreamProgressResult{}, err
|
|
}
|
|
if _, err := svc.validatedRunSession(request.RunEndpointID, request.SessionToken); err != nil {
|
|
return domain.RunLogStreamProgressResult{}, err
|
|
}
|
|
instance, err := svc.store.ServerInstances().Get(request.ServerInstanceID)
|
|
if err != nil {
|
|
return domain.RunLogStreamProgressResult{}, err
|
|
}
|
|
if instance.RunEndpointID != request.RunEndpointID {
|
|
return domain.RunLogStreamProgressResult{}, validationError("runEndpointId must match server instance")
|
|
}
|
|
if !strings.HasPrefix(request.LogStreamID, "run."+request.RunEndpointID+"."+request.ServerInstanceID+".") {
|
|
return domain.RunLogStreamProgressResult{}, validationError("logStreamId is not a bound run stream")
|
|
}
|
|
latest := uint64(0)
|
|
stream, err := svc.store.LogStreams().Get(request.LogStreamID)
|
|
if err == nil {
|
|
if stream.ServerInstanceID != instance.ID {
|
|
return domain.RunLogStreamProgressResult{}, validationError("logStreamId does not belong to server instance")
|
|
}
|
|
latest = stream.LatestSeq
|
|
} else if !errors.Is(err, repo.ErrNotFound) {
|
|
return domain.RunLogStreamProgressResult{}, err
|
|
}
|
|
return domain.RunLogStreamProgressResult{Accepted: true, RunEndpointID: request.RunEndpointID, ServerInstanceID: instance.ID, LogStreamID: request.LogStreamID, LatestSeq: latest, ServerTime: svc.now()}, nil
|
|
}
|
|
|
|
func validateLogBatchStream(batch domain.LogBatchIngest, stream domain.LogStream) error {
|
|
if stream.ID != batch.LogStreamID {
|
|
return validationError("logStreamId must match stream")
|
|
}
|
|
if stream.ServerInstanceID != batch.ServerInstanceID {
|
|
return validationError("serverInstanceId must match stream")
|
|
}
|
|
if stream.StreamKey != batch.StreamKey {
|
|
return validationError("streamKey must match stream")
|
|
}
|
|
if stream.Source != batch.Source {
|
|
return validationError("source must match stream")
|
|
}
|
|
if stream.LogSessionID != batch.LogSessionID || !stream.SessionStartedAt.Equal(batch.SessionStartedAt) {
|
|
return validationError("log session metadata must match stream")
|
|
}
|
|
return nil
|
|
}
|