diff --git a/CHANGELOG.md b/CHANGELOG.md index 4e8a58c..6594eaf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,8 @@ Versioning. ### Added +- Added bounded extraction of `nddev_tool_cache_event` records from verified + runner diagnostic artifacts, with malformed job text counted but non-blocking. - Added the portable scheduler-recovery decision core: exact stuck-dispatch detection, startup grace, cooldown, and duplicate-recovery exclusion. - Added checkpoint-first recovery orchestration with durable attempt identity, diff --git a/cmd/gha-diagnostic-exporter/main.go b/cmd/gha-diagnostic-exporter/main.go index e7fbceb..d001fa8 100644 --- a/cmd/gha-diagnostic-exporter/main.go +++ b/cmd/gha-diagnostic-exporter/main.go @@ -73,11 +73,27 @@ func run(args []string, stdout, stderr io.Writer) error { logger.Error("diagnostic export incomplete", "error", err) return err } + for _, observed := range summary.ToolCacheEvents { + logger.Info( + "nddev tool cache event", + "event_type", "nddev_tool_cache_event", + "repository", observed.Repository, + "runner", observed.Runner, + "captured_at", observed.CapturedAt, + "source", observed.Event.Source, + "cache_result", observed.Event.CacheResult, + "sha256", observed.Event.SHA256, + "bytes", observed.Event.Bytes, + "duration_ms", observed.Event.DurationMS, + ) + } logger.Info( "diagnostic export complete", "source_bundles", summary.SourceBundles, "exported_bundles", summary.ExportedBundles, "pending_bundles", summary.PendingBundles, + "tool_cache_events", summary.ToolCacheEventCount, + "rejected_tool_cache_events", summary.RejectedToolCacheEventCount, ) return nil } diff --git a/internal/diagnosticexport/bundle.go b/internal/diagnosticexport/bundle.go index a61a036..f85de48 100644 --- a/internal/diagnosticexport/bundle.go +++ b/internal/diagnosticexport/bundle.go @@ -31,14 +31,16 @@ const ( ) type Bundle struct { - Name string - Content []byte - SHA256 string - Manifest workerdiagnostics.Manifest - CapturedAt time.Time - ObjectKey string - device uint64 - inode uint64 + Name string + Content []byte + SHA256 string + Manifest workerdiagnostics.Manifest + CapturedAt time.Time + ObjectKey string + ToolCacheEvents []ToolCacheEvent + RejectedToolCacheEvents int + device uint64 + inode uint64 } type bundleScope uint8 @@ -223,15 +225,18 @@ func ReadBundle(ctx context.Context, config Config, name string) (Bundle, error) if err != nil { return Bundle{}, fmt.Errorf("derive diagnostic object key: %w", err) } + events, rejectedEvents := ExtractToolCacheEvents(content) return Bundle{ - Name: name, - Content: content, - SHA256: digestHex, - Manifest: manifest, - CapturedAt: capturedAt, - ObjectKey: objectKey, - device: before.Dev, - inode: before.Ino, + Name: name, + Content: content, + SHA256: digestHex, + Manifest: manifest, + CapturedAt: capturedAt, + ObjectKey: objectKey, + ToolCacheEvents: events, + RejectedToolCacheEvents: rejectedEvents, + device: before.Dev, + inode: before.Ino, }, nil } diff --git a/internal/diagnosticexport/exporter.go b/internal/diagnosticexport/exporter.go index 8c0e0e9..b71acdf 100644 --- a/internal/diagnosticexport/exporter.go +++ b/internal/diagnosticexport/exporter.go @@ -26,13 +26,23 @@ type ObjectStore interface { } type Summary struct { - SourceBundles int `json:"source_bundles"` - ExportedBundles int `json:"exported_bundles"` - PendingBundles int `json:"pending_bundles"` - SourceBytes int64 `json:"source_bytes"` - ExportedBytes int64 `json:"exported_bytes"` - ScannedBundles int `json:"scanned_bundles"` - DeletedBundles int `json:"deleted_bundles"` + SourceBundles int `json:"source_bundles"` + ExportedBundles int `json:"exported_bundles"` + PendingBundles int `json:"pending_bundles"` + SourceBytes int64 `json:"source_bytes"` + ExportedBytes int64 `json:"exported_bytes"` + ScannedBundles int `json:"scanned_bundles"` + DeletedBundles int `json:"deleted_bundles"` + ToolCacheEventCount int `json:"tool_cache_event_count"` + RejectedToolCacheEventCount int `json:"rejected_tool_cache_event_count"` + ToolCacheEvents []ObservedToolCacheEvent `json:"-"` +} + +type ObservedToolCacheEvent struct { + Event ToolCacheEvent + Repository string + Runner string + CapturedAt time.Time } type Exporter struct { @@ -123,6 +133,7 @@ func (e Exporter) Run(ctx context.Context) (Summary, error) { } delete(state.Exports, name) summary.DeletedBundles++ + summary.addToolCacheEvents(bundle) continue } remote, err := e.Store.Head(ctx, e.Config.Bucket, bundle.ObjectKey) @@ -184,6 +195,7 @@ func (e Exporter) Run(ctx context.Context) (Summary, error) { summary.DeletedBundles++ summary.ExportedBundles++ summary.ExportedBytes += bundleBytes + summary.addToolCacheEvents(bundle) } _, remainingBundles, remainingBytes, scanErr := ListBundleBatch(ctx, e.Config, 1) if scanErr != nil { @@ -236,6 +248,17 @@ func (e Exporter) Run(ctx context.Context) (Summary, error) { return summary, nil } +func (summary *Summary) addToolCacheEvents(bundle Bundle) { + summary.RejectedToolCacheEventCount += bundle.RejectedToolCacheEvents + for _, event := range bundle.ToolCacheEvents { + summary.ToolCacheEvents = append(summary.ToolCacheEvents, ObservedToolCacheEvent{ + Event: event, Repository: bundle.Manifest.Instance.Repository, + Runner: bundle.Manifest.Instance.Name, CapturedAt: bundle.CapturedAt, + }) + summary.ToolCacheEventCount++ + } +} + func (e Exporter) saveFailure( state State, now time.Time, diff --git a/internal/diagnosticexport/toolcache.go b/internal/diagnosticexport/toolcache.go new file mode 100644 index 0000000..0010f1c --- /dev/null +++ b/internal/diagnosticexport/toolcache.go @@ -0,0 +1,81 @@ +package diagnosticexport + +import ( + "archive/tar" + "bufio" + "bytes" + "compress/gzip" + "encoding/hex" + "encoding/json" + "errors" + "io" + "strings" +) + +const ( + toolCacheMarker = "nddev_tool_cache_event=" + maxToolCacheEvents = 64 + maxToolCacheLine = 16 * 1024 +) + +type ToolCacheEvent struct { + Source string `json:"source"` + CacheResult string `json:"cache_result"` + SHA256 string `json:"sha256"` + Bytes int64 `json:"bytes"` + DurationMS int64 `json:"duration_ms"` +} + +func ExtractToolCacheEvents(content []byte) ([]ToolCacheEvent, int) { + compressed := bytes.NewReader(content) + decompressor, err := gzip.NewReader(compressed) + if err != nil { + return nil, 1 + } + defer decompressor.Close() + archive := tar.NewReader(decompressor) + events := make([]ToolCacheEvent, 0) + rejected := 0 + for len(events) < maxToolCacheEvents { + header, err := archive.Next() + if errors.Is(err, io.EOF) { + break + } + if err != nil { + return events, rejected + 1 + } + if header.Typeflag != tar.TypeReg || !strings.HasPrefix(header.Name, "runner/") { + continue + } + scanner := bufio.NewScanner(io.LimitReader(archive, 2*1024*1024)) + scanner.Buffer(make([]byte, 4096), maxToolCacheLine) + for scanner.Scan() && len(events) < maxToolCacheEvents { + line := scanner.Text() + index := strings.Index(line, toolCacheMarker) + if index < 0 { + continue + } + var event ToolCacheEvent + decoder := json.NewDecoder(strings.NewReader(line[index+len(toolCacheMarker):])) + decoder.DisallowUnknownFields() + if err := decoder.Decode(&event); err != nil || !validToolCacheEvent(event) { + rejected++ + continue + } + events = append(events, event) + } + if scanner.Err() != nil { + rejected++ + } + } + return events, rejected +} + +func validToolCacheEvent(event ToolCacheEvent) bool { + if event.Source == "" || len(event.Source) > 32 || event.CacheResult == "" || len(event.CacheResult) > 96 || + event.Bytes < 0 || event.DurationMS < 0 || len(event.SHA256) != 64 { + return false + } + _, err := hex.DecodeString(event.SHA256) + return err == nil +} diff --git a/internal/diagnosticexport/toolcache_test.go b/internal/diagnosticexport/toolcache_test.go new file mode 100644 index 0000000..6553ca1 --- /dev/null +++ b/internal/diagnosticexport/toolcache_test.go @@ -0,0 +1,67 @@ +package diagnosticexport + +import ( + "archive/tar" + "bytes" + "compress/gzip" + "strings" + "testing" + "time" + + "github.com/NDDev-OpenNetwork/github-actions/internal/workerdiagnostics" +) + +func TestExtractToolCacheEventsFromRunnerArtifacts(t *testing.T) { + t.Parallel() + archive := toolCacheArchive(t, map[string]string{ + "runner/Worker.log": "prefix nddev_tool_cache_event={\"source\":\"upstream\",\"cache_result\":\"miss\",\"sha256\":\"" + strings.Repeat("a", 64) + "\",\"bytes\":42,\"duration_ms\":7}\nnddev_tool_cache_event={bad}\n", + "incus/qemu.log": "nddev_tool_cache_event={\"source\":\"ignored\"}\n", + }) + events, rejected := ExtractToolCacheEvents(archive) + if len(events) != 1 || rejected != 1 { + t.Fatalf("events=%#v rejected=%d", events, rejected) + } + if events[0].Source != "upstream" || events[0].Bytes != 42 || events[0].DurationMS != 7 { + t.Fatalf("event=%#v", events[0]) + } +} + +func TestSummaryBindsToolCacheEventToBundleIdentity(t *testing.T) { + t.Parallel() + var summary Summary + captured := time.Date(2026, 8, 24, 10, 0, 0, 0, time.UTC) + summary.addToolCacheEvents(Bundle{ + Manifest: workerdiagnostics.Manifest{Instance: workerdiagnostics.Instance{Name: "runner-1", Repository: "example/repository"}}, + CapturedAt: captured, RejectedToolCacheEvents: 2, + ToolCacheEvents: []ToolCacheEvent{{Source: "cache", CacheResult: "hit", SHA256: strings.Repeat("a", 64)}}, + }) + if summary.ToolCacheEventCount != 1 || summary.RejectedToolCacheEventCount != 2 || len(summary.ToolCacheEvents) != 1 { + t.Fatalf("summary=%#v", summary) + } + observed := summary.ToolCacheEvents[0] + if observed.Repository != "example/repository" || observed.Runner != "runner-1" || !observed.CapturedAt.Equal(captured) { + t.Fatalf("observed=%#v", observed) + } +} + +func toolCacheArchive(t *testing.T, files map[string]string) []byte { + t.Helper() + var buffer bytes.Buffer + compressed := gzip.NewWriter(&buffer) + archive := tar.NewWriter(compressed) + for name, content := range files { + if err := archive.WriteHeader(&tar.Header{Name: name, Mode: 0o600, Size: int64(len(content))}); err != nil { + t.Fatal(err) + } + if _, err := archive.Write([]byte(content)); err != nil { + t.Fatal(err) + } + } + if err := archive.Close(); err != nil { + t.Fatal(err) + } + if err := compressed.Close(); err != nil { + t.Fatal(err) + } + return buffer.Bytes() +} diff --git a/internal/schedulerrecovery/command_test.go b/internal/schedulerrecovery/command_test.go index c8150c2..8709308 100644 --- a/internal/schedulerrecovery/command_test.go +++ b/internal/schedulerrecovery/command_test.go @@ -47,11 +47,13 @@ func TestCommandExecutorRejectsShellAndTimeouts(t *testing.T) { func writeExecutable(t *testing.T, directory, name, content string) string { t.Helper() path := filepath.Join(directory, name) - file, err := os.OpenFile(path, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o700) + temporary := path + ".tmp" + file, err := os.OpenFile(temporary, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o700) require.NoError(t, err) _, err = file.WriteString(content) require.NoError(t, err) require.NoError(t, file.Sync()) require.NoError(t, file.Close()) + require.NoError(t, os.Rename(temporary, path)) return path }