admin343/run
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add runtime SQLite query targets

Browse Source
This commit is contained in:
npc0-hue
2026-09-01 17:15:09 +08:00
parent 65da73eff0
commit 163fbda394
17 changed files with 708 additions and 114 deletions
Download Patch File Download Diff File Expand all files Collapse all files
runtime/worker.go
+15 -7
View File
@@ -735,6 +735,18 @@ func (worker *Worker) executeAssignment(ctx context.Context, assignment protocol
}
return worker.executor.sqliteSchemaProbe.Execute(ctx, assignment)
}
if assignment.Capability == protocol.RunCapabilityRemoteRunDBSQLiteQuery {
log.Printf("RUN phase=job.dispatch status=selected job=%s executor=sqlite_query", assignment.JobID)
if worker.executor.sqliteQuery == nil {
return lifecycleFailure("sqlite_query_unavailable", "SQLite query executor is unavailable")
}
targetKey, err := worker.materializeSQLiteQueryDataTarget(ctx, assignment)
if err != nil {
return sqliteQueryFailureForDataTarget(err)
}
assignment.TargetKey = targetKey
return worker.executor.sqliteQuery.Execute(ctx, assignment)
}
if isSupportedLifecycleCapability(assignment.Capability) {
log.Printf("RUN phase=job.dispatch status=selected job=%s executor=lifecycle", assignment.JobID)
return worker.executor.ExecuteContext(ctx, assignment)
@@ -1484,8 +1496,6 @@ func (sink *LiveLogSink) append(assignment protocol.RunJobAssignment, stream str
if sink == nil {
return nil
}
redactedLine := RedactText(line)
redacted := line != redactedLine || strings.HasPrefix(redactedLine, "[redacted protected ")
streamKey := declaredProcessStreamKey(assignment, stream)
logStreamID := logStreamIDForAssignment(assignment, streamKey)
sink.mu.Lock()
@@ -1514,7 +1524,7 @@ func (sink *LiveLogSink) append(assignment protocol.RunJobAssignment, stream str
FirstSeq: sequence,
LastSeq: sequence,
Compression: "none",
Entries: []protocol.LogEntry{{Seq: sequence, Timestamp: time.Now().UTC(), Level: "info", Line: redactedLine, Redacted: redacted}},
Entries: []protocol.LogEntry{{Seq: sequence, Timestamp: time.Now().UTC(), Level: "info", Line: line, Redacted: false}},
}
batch.Checksum, _ = checksumForLogEntries(batch.Entries)
select {
@@ -1564,11 +1574,9 @@ func (sink *SpoolLogSink) AppendWithCursor(ctx context.Context, assignment proto
func (sink *SpoolLogSink) append(ctx context.Context, assignment protocol.RunJobAssignment, stream string, line string, cursor *spool.LogSourceCursor) error {
sink.mu.Lock()
defer sink.mu.Unlock()
redactedLine := RedactText(line)
redacted := line != redactedLine || strings.HasPrefix(redactedLine, "[redacted protected ")
streamKey := declaredProcessStreamKey(assignment, stream)
logStreamID := logStreamIDForAssignment(assignment, streamKey)
entry := protocol.LogEntry{Timestamp: time.Now().UTC(), Level: "info", Line: redactedLine, Redacted: redacted}
entry := protocol.LogEntry{Timestamp: time.Now().UTC(), Level: "info", Line: line, Redacted: false}
source := "process"
if strings.HasPrefix(stream, "management-program.") {
source = "management-program"
@@ -1640,7 +1648,7 @@ type QueueArtifactHook struct {
func (hook QueueArtifactHook) QueueLifecycleResult(_ context.Context, assignment protocol.RunJobAssignment, result ProcessResult) (string, error) {
ref := fmt.Sprintf("artifact://jobs/%s/lifecycle-result", assignment.JobID)
payload := []byte(RedactText(result.Stdout + result.Stderr))
payload := []byte(result.Stdout + result.Stderr)
if len(payload) == 0 {
payload = []byte("lifecycle result metadata")
}