127 lines
3.9 KiB
Go
127 lines
3.9 KiB
Go
package runtime
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"fmt"
|
|
"io"
|
|
"net/url"
|
|
"os"
|
|
"strings"
|
|
|
|
"browser.local/run/protocol"
|
|
)
|
|
|
|
type LogSourceCheckpoint struct {
|
|
SourceKey string
|
|
Offset int64
|
|
Sequence uint64
|
|
CursorRef string
|
|
}
|
|
|
|
type LogCheckpointStore interface {
|
|
GetLogCheckpoint(sourceKey string) LogSourceCheckpoint
|
|
PutLogCheckpoint(checkpoint LogSourceCheckpoint)
|
|
}
|
|
|
|
type MemoryLogCheckpointStore struct {
|
|
checkpoints map[string]LogSourceCheckpoint
|
|
}
|
|
|
|
func NewMemoryLogCheckpointStore() *MemoryLogCheckpointStore {
|
|
return &MemoryLogCheckpointStore{checkpoints: map[string]LogSourceCheckpoint{}}
|
|
}
|
|
|
|
func (store *MemoryLogCheckpointStore) GetLogCheckpoint(sourceKey string) LogSourceCheckpoint {
|
|
if store == nil || store.checkpoints == nil {
|
|
return LogSourceCheckpoint{SourceKey: sourceKey}
|
|
}
|
|
return store.checkpoints[sourceKey]
|
|
}
|
|
|
|
func (store *MemoryLogCheckpointStore) PutLogCheckpoint(checkpoint LogSourceCheckpoint) {
|
|
if store == nil {
|
|
return
|
|
}
|
|
if store.checkpoints == nil {
|
|
store.checkpoints = map[string]LogSourceCheckpoint{}
|
|
}
|
|
store.checkpoints[checkpoint.SourceKey] = checkpoint
|
|
}
|
|
|
|
func TailDeclaredFileLogSource(ctx context.Context, workspaceRoot string, assignment protocol.RunJobAssignment, source RuntimeLogSource, sink ProcessLogSink, store LogCheckpointStore) LifecycleExecutionResult {
|
|
if source.Kind != "file.tail" {
|
|
return lifecycleFailure("unsupported_log_source", "only file.tail sources are supported by the local tailer")
|
|
}
|
|
if !protocol.ValidLogicalFileKey(source.Key) || !protocol.ValidLogicalFileKey(source.TargetKey) || !protocol.ValidLogicalFileKey(source.StreamKey) {
|
|
return lifecycleFailure("unsafe_log_source", "log source is unsafe")
|
|
}
|
|
if sink == nil {
|
|
sink = NoopProcessLogSink{}
|
|
}
|
|
if store == nil {
|
|
store = NewMemoryLogCheckpointStore()
|
|
}
|
|
serverRoot, err := scopedServerWorkspace(workspaceRoot, assignment.ServerInstanceID)
|
|
if err != nil {
|
|
return lifecycleFailure("unsafe_log_workspace", err.Error())
|
|
}
|
|
path, err := scopedPath(serverRoot, source.TargetKey)
|
|
if err != nil {
|
|
return lifecycleFailure("unsafe_log_source", err.Error())
|
|
}
|
|
file, err := os.Open(path)
|
|
if err != nil {
|
|
return lifecycleFailure("log_source_open_failed", err.Error())
|
|
}
|
|
defer file.Close()
|
|
|
|
checkpoint := store.GetLogCheckpoint(source.Key)
|
|
if checkpoint.Offset > 0 {
|
|
if _, err := file.Seek(checkpoint.Offset, 0); err != nil {
|
|
return lifecycleFailure("log_source_seek_failed", err.Error())
|
|
}
|
|
}
|
|
reader := bufio.NewReader(file)
|
|
for {
|
|
// A newline only frames an entry. Every other byte, including CR, blank
|
|
// lines, and arbitrarily long output, remains untouched.
|
|
line, readErr := reader.ReadString('\n')
|
|
if len(line) > 0 {
|
|
checkpoint.Sequence++
|
|
if err := sink.Append(ctx, assignment, source.StreamKey, strings.TrimSuffix(line, "\n")); err != nil {
|
|
return lifecycleFailure("log_source_sink_failed", err.Error())
|
|
}
|
|
checkpoint.SourceKey = source.Key
|
|
checkpoint.Offset += int64(len(line))
|
|
checkpoint.CursorRef = fmt.Sprintf("artifact://jobs/%s/live-log-checkpoint", url.PathEscape(assignment.JobID))
|
|
store.PutLogCheckpoint(checkpoint)
|
|
}
|
|
if readErr == nil {
|
|
continue
|
|
}
|
|
if readErr == io.EOF {
|
|
break
|
|
}
|
|
return lifecycleFailure("log_source_read_failed", readErr.Error())
|
|
}
|
|
if checkpoint.CursorRef == "" {
|
|
checkpoint.CursorRef = fmt.Sprintf("artifact://jobs/%s/live-log-checkpoint", url.PathEscape(assignment.JobID))
|
|
}
|
|
return LifecycleExecutionResult{
|
|
State: lifecycleResultStateSucceeded,
|
|
Progress: protocol.RunJobProgressReport{Percent: 100, Message: "live log checkpoint updated"},
|
|
ResultRef: checkpoint.CursorRef,
|
|
Message: "live log source tailed with durable offset checkpoint",
|
|
}
|
|
}
|
|
|
|
func RedactedLogCheckpointSummary(checkpoint LogSourceCheckpoint) string {
|
|
return strings.Join([]string{
|
|
"source=" + checkpoint.SourceKey,
|
|
fmt.Sprintf("offset=%d", checkpoint.Offset),
|
|
fmt.Sprintf("sequence=%d", checkpoint.Sequence),
|
|
"cursorRef=" + checkpoint.CursorRef,
|
|
}, " ")
|
|
}
|