Add internal debug bundle collaborators
This commit is contained in:
@@ -222,7 +222,11 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
|
||||
if *resume && !workspaceSettings.ResumeEnabled {
|
||||
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("--resume requires workspace.resume.enabled: true"))
|
||||
}
|
||||
debugRecorder, err := frameworkdebug.NewWorkspaceRecorder(workspaceSettings, runID)
|
||||
debugRoot, err := workspaceSettings.DebugRunDirectory(runID)
|
||||
if err != nil {
|
||||
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("resolve debug root: %w", err))
|
||||
}
|
||||
debugRecorder, err := frameworkdebug.NewFilesystemRecorder(debugRoot)
|
||||
if err != nil {
|
||||
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("create debug recorder: %w", err))
|
||||
}
|
||||
|
||||
@@ -6,11 +6,11 @@ func (c Config) Redacted() Config {
|
||||
return cloneConfig(c)
|
||||
}
|
||||
|
||||
func (c Config) RedactedDiagnosticsPayload() any {
|
||||
func (c Config) RedactedSummaryPayload() any {
|
||||
return c.Redacted()
|
||||
}
|
||||
|
||||
func (e EffectiveConfig) RedactedDiagnosticsPayload() any {
|
||||
func (e EffectiveConfig) RedactedSummaryPayload() any {
|
||||
return EffectiveConfig{
|
||||
Config: e.Config.Redacted(),
|
||||
PipelineID: e.PipelineID,
|
||||
|
||||
@@ -36,20 +36,20 @@ func TestRedactedConfigCopiesScriptoriumConfig(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestConfigRedactedDiagnosticsPayloadCopiesConfig(t *testing.T) {
|
||||
func TestConfigRedactedSummaryPayloadCopiesConfig(t *testing.T) {
|
||||
cfg := Default()
|
||||
cfg.Scriptorium.ProfileFile = "./profiles.yml"
|
||||
|
||||
payload, ok := cfg.RedactedDiagnosticsPayload().(Config)
|
||||
payload, ok := cfg.RedactedSummaryPayload().(Config)
|
||||
if !ok {
|
||||
t.Fatalf("expected Config payload, got %T", cfg.RedactedDiagnosticsPayload())
|
||||
t.Fatalf("expected Config payload, got %T", cfg.RedactedSummaryPayload())
|
||||
}
|
||||
if payload.Scriptorium.ProfileFile != "./profiles.yml" {
|
||||
t.Fatalf("expected Scriptorium profile file preserved, got %+v", payload.Scriptorium)
|
||||
}
|
||||
}
|
||||
|
||||
func TestEffectiveConfigRedactedDiagnosticsPayloadCopies(t *testing.T) {
|
||||
func TestEffectiveConfigRedactedSummaryPayloadCopies(t *testing.T) {
|
||||
cfg := validConfig()
|
||||
cfg.Concurrency.TotalLLM = 4
|
||||
cfg.Concurrency.StageWorkers["extract"] = 2
|
||||
@@ -101,9 +101,9 @@ func TestEffectiveConfigRedactedDiagnosticsPayloadCopies(t *testing.T) {
|
||||
t.Fatalf("Resolve: %v", err)
|
||||
}
|
||||
|
||||
payload, ok := effective.RedactedDiagnosticsPayload().(EffectiveConfig)
|
||||
payload, ok := effective.RedactedSummaryPayload().(EffectiveConfig)
|
||||
if !ok {
|
||||
t.Fatalf("expected EffectiveConfig payload, got %T", effective.RedactedDiagnosticsPayload())
|
||||
t.Fatalf("expected EffectiveConfig payload, got %T", effective.RedactedSummaryPayload())
|
||||
}
|
||||
if payload.PipelineID != effective.PipelineID || payload.ResolvedPipeline.Digest != effective.ResolvedPipeline.Digest {
|
||||
t.Fatalf("expected pipeline metadata preserved, got %+v", payload)
|
||||
@@ -151,9 +151,9 @@ func TestEffectiveConfigRedactedDiagnosticsPayloadCopies(t *testing.T) {
|
||||
},
|
||||
},
|
||||
}
|
||||
payload, ok = effective.RedactedDiagnosticsPayload().(EffectiveConfig)
|
||||
payload, ok = effective.RedactedSummaryPayload().(EffectiveConfig)
|
||||
if !ok {
|
||||
t.Fatalf("expected EffectiveConfig payload, got %T", effective.RedactedDiagnosticsPayload())
|
||||
t.Fatalf("expected EffectiveConfig payload, got %T", effective.RedactedSummaryPayload())
|
||||
}
|
||||
payload.ResolvedPipeline.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0].Content[0] = 'X'
|
||||
got := effective.ResolvedPipeline.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0].Content
|
||||
|
||||
89
internal/core/debugbundle/bundle.go
Normal file
89
internal/core/debugbundle/bundle.go
Normal file
@@ -0,0 +1,89 @@
|
||||
// Package debugbundle owns explicitly requested per-run debug bundles.
|
||||
package debugbundle
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
const maxCreateAttempts = 16
|
||||
|
||||
var utcNow = func() time.Time { return time.Now().UTC() }
|
||||
|
||||
type Bundle struct {
|
||||
path, summaryRoot, traceRoot string
|
||||
createdAt time.Time
|
||||
}
|
||||
|
||||
func Allocate(parent string) (*Bundle, error) {
|
||||
parent = strings.TrimSpace(parent)
|
||||
if parent == "" {
|
||||
return nil, fmt.Errorf("debug parent must not be empty")
|
||||
}
|
||||
if err := os.MkdirAll(parent, 0o700); err != nil {
|
||||
return nil, fmt.Errorf("create debug parent %q: %w", parent, err)
|
||||
}
|
||||
var last string
|
||||
for attempt := 0; attempt < maxCreateAttempts; attempt++ {
|
||||
createdAt := utcNow()
|
||||
runID := fmt.Sprintf("run-%d", createdAt.UnixNano())
|
||||
path := filepath.Join(parent, runID)
|
||||
last = path
|
||||
if err := os.Mkdir(path, 0o700); err != nil {
|
||||
if os.IsExist(err) {
|
||||
continue
|
||||
}
|
||||
return nil, fmt.Errorf("create debug bundle %q: %w", path, err)
|
||||
}
|
||||
summary, trace := filepath.Join(path, "summary"), filepath.Join(path, "trace")
|
||||
if err := os.Mkdir(summary, 0o700); err != nil {
|
||||
_ = os.Remove(path)
|
||||
return nil, fmt.Errorf("create debug summary %q: %w", summary, err)
|
||||
}
|
||||
if err := os.Mkdir(trace, 0o700); err != nil {
|
||||
_ = os.RemoveAll(path)
|
||||
return nil, fmt.Errorf("create debug trace %q: %w", trace, err)
|
||||
}
|
||||
return &Bundle{path: path, summaryRoot: summary, traceRoot: trace, createdAt: createdAt}, nil
|
||||
}
|
||||
return nil, fmt.Errorf("create debug bundle %q: exhausted unique run ID attempts", last)
|
||||
}
|
||||
func (b *Bundle) Path() string {
|
||||
if b == nil {
|
||||
return ""
|
||||
}
|
||||
return b.path
|
||||
}
|
||||
func (b *Bundle) SummaryRoot() string {
|
||||
if b == nil {
|
||||
return ""
|
||||
}
|
||||
return b.summaryRoot
|
||||
}
|
||||
func (b *Bundle) TraceRoot() string {
|
||||
if b == nil {
|
||||
return ""
|
||||
}
|
||||
return b.traceRoot
|
||||
}
|
||||
func (b *Bundle) RunID() string {
|
||||
if b == nil {
|
||||
return ""
|
||||
}
|
||||
return filepath.Base(b.path)
|
||||
}
|
||||
func (b *Bundle) CreatedAt() time.Time {
|
||||
if b == nil {
|
||||
return time.Time{}
|
||||
}
|
||||
return b.createdAt
|
||||
}
|
||||
func (b *Bundle) Summary() *SummaryWriter {
|
||||
if b == nil {
|
||||
return nil
|
||||
}
|
||||
return &SummaryWriter{root: b.summaryRoot, runID: b.RunID(), createdAt: b.createdAt}
|
||||
}
|
||||
75
internal/core/debugbundle/bundle_test.go
Normal file
75
internal/core/debugbundle/bundle_test.go
Normal file
@@ -0,0 +1,75 @@
|
||||
package debugbundle
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestAllocateCreatesRestrictiveSummaryAndTrace(t *testing.T) {
|
||||
parent := t.TempDir()
|
||||
fixed := time.Unix(0, 42).UTC()
|
||||
previous := utcNow
|
||||
utcNow = func() time.Time { return fixed }
|
||||
defer func() { utcNow = previous }()
|
||||
bundle, err := Allocate(parent)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if bundle.RunID() != "run-42" || bundle.SummaryRoot() != filepath.Join(bundle.Path(), "summary") || bundle.TraceRoot() != filepath.Join(bundle.Path(), "trace") {
|
||||
t.Fatalf("bundle=%#v", bundle)
|
||||
}
|
||||
for _, path := range []string{bundle.Path(), bundle.SummaryRoot(), bundle.TraceRoot()} {
|
||||
info, err := os.Stat(path)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if info.Mode().Perm() != 0o700 {
|
||||
t.Fatalf("%s mode=%#o", path, info.Mode().Perm())
|
||||
}
|
||||
}
|
||||
if err := bundle.Summary().WriteError("failed"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
info, err := os.Stat(filepath.Join(bundle.SummaryRoot(), ArtifactErrorLog))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if info.Mode().Perm() != 0o600 {
|
||||
t.Fatalf("file mode=%#o", info.Mode().Perm())
|
||||
}
|
||||
}
|
||||
func TestAllocateRetriesAndDoesNotDeleteBundle(t *testing.T) {
|
||||
parent := t.TempDir()
|
||||
fixed := time.Unix(0, 9).UTC()
|
||||
previous := utcNow
|
||||
defer func() { utcNow = previous }()
|
||||
calls := 0
|
||||
utcNow = func() time.Time { calls++; return fixed.Add(time.Duration(calls-1) * time.Nanosecond) }
|
||||
if err := os.Mkdir(filepath.Join(parent, "run-9"), 0o700); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
bundle, err := Allocate(parent)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if bundle.RunID() != "run-10" {
|
||||
t.Fatalf("run id=%q", bundle.RunID())
|
||||
}
|
||||
if _, err := os.Stat(bundle.Path()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
func TestSummaryWriterConfinesArtifacts(t *testing.T) {
|
||||
bundle, err := Allocate(t.TempDir())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := bundle.Summary().WriteJSON("../outside.json", map[string]any{}); err == nil {
|
||||
t.Fatal("accepted traversal")
|
||||
}
|
||||
if err := bundle.Summary().WriteBytes(`trace\\x`, []byte("x")); err == nil {
|
||||
t.Fatal("accepted backslash")
|
||||
}
|
||||
}
|
||||
115
internal/core/debugbundle/summary.go
Normal file
115
internal/core/debugbundle/summary.go
Normal file
@@ -0,0 +1,115 @@
|
||||
package debugbundle
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/core/fileio"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
|
||||
)
|
||||
|
||||
const (
|
||||
ArtifactInvocationMetadata = "invocation.json"
|
||||
ArtifactEffectiveConfig = "effective-config.json"
|
||||
ArtifactResolvedPipeline = "resolved-pipeline.json"
|
||||
ArtifactResolvedReferences = "resolved-references.json"
|
||||
ArtifactCheckpointEvents = "checkpoint-events.json"
|
||||
ArtifactRunManifest = "run-manifest.json"
|
||||
ArtifactChunkPlan = "chunk-plan.json"
|
||||
ArtifactRunReport = "run-report.json"
|
||||
ArtifactWarnings = "warnings.json"
|
||||
ArtifactErrorLog = "error.log"
|
||||
)
|
||||
|
||||
type RedactedSummaryPayload interface{ RedactedSummaryPayload() any }
|
||||
type Invocation struct {
|
||||
Operation string `json:"operation"`
|
||||
PipelineID string `json:"pipeline_id,omitempty"`
|
||||
PipelineDigest string `json:"pipeline_digest,omitempty"`
|
||||
Resume bool `json:"resume,omitempty"`
|
||||
InputPath string `json:"input_path,omitempty"`
|
||||
ConfigPath string `json:"config_path,omitempty"`
|
||||
ConfigSource string `json:"config_source,omitempty"`
|
||||
OnlyLanes []string `json:"only_lanes,omitempty"`
|
||||
ChunkCacheOverride string `json:"chunk_cache_override,omitempty"`
|
||||
RunID string `json:"run_id"`
|
||||
StartedAt time.Time `json:"started_at"`
|
||||
}
|
||||
type RunReport struct {
|
||||
OutputPath string `json:"output_path,omitempty"`
|
||||
DebugPath string `json:"debug_path,omitempty"`
|
||||
Succeeded bool `json:"succeeded"`
|
||||
WarningCount int `json:"warning_count"`
|
||||
}
|
||||
type SummaryWriter struct {
|
||||
root, runID string
|
||||
createdAt time.Time
|
||||
}
|
||||
|
||||
func (w *SummaryWriter) WriteInvocation(payload Invocation) error {
|
||||
if w == nil {
|
||||
return fmt.Errorf("debug summary writer must not be nil")
|
||||
}
|
||||
if payload.RunID == "" {
|
||||
payload.RunID = w.runID
|
||||
}
|
||||
if payload.StartedAt.IsZero() {
|
||||
payload.StartedAt = w.createdAt
|
||||
}
|
||||
return w.WriteJSON(ArtifactInvocationMetadata, payload)
|
||||
}
|
||||
func (w *SummaryWriter) WriteRedactedEffectiveConfig(payload RedactedSummaryPayload) error {
|
||||
if payload == nil {
|
||||
return fmt.Errorf("redacted summary payload must not be nil")
|
||||
}
|
||||
return w.WriteJSON(ArtifactEffectiveConfig, payload.RedactedSummaryPayload())
|
||||
}
|
||||
func (w *SummaryWriter) WriteResolvedPipeline(v any) error {
|
||||
return w.WriteJSON(ArtifactResolvedPipeline, v)
|
||||
}
|
||||
func (w *SummaryWriter) WriteResolvedReferences(v any) error {
|
||||
return w.WriteJSON(ArtifactResolvedReferences, v)
|
||||
}
|
||||
func (w *SummaryWriter) WriteCheckpointEvents(v any) error {
|
||||
return w.WriteJSON(ArtifactCheckpointEvents, v)
|
||||
}
|
||||
func (w *SummaryWriter) WriteRunManifest(v artifacts.RunManifest) error {
|
||||
return w.WriteJSON(ArtifactRunManifest, v)
|
||||
}
|
||||
func (w *SummaryWriter) WriteChunkPlan(v artifacts.ChunkPlanSummary) error {
|
||||
return w.WriteJSON(ArtifactChunkPlan, v)
|
||||
}
|
||||
func (w *SummaryWriter) WriteRunReport(v RunReport) error { return w.WriteJSON(ArtifactRunReport, v) }
|
||||
func (w *SummaryWriter) WriteWarnings(v []contracts.Warning) error {
|
||||
return w.WriteJSON(ArtifactWarnings, v)
|
||||
}
|
||||
func (w *SummaryWriter) WriteError(message string) error {
|
||||
return w.WriteBytes(ArtifactErrorLog, []byte(message+"\n"))
|
||||
}
|
||||
func (w *SummaryWriter) WriteJSON(name string, v any) error {
|
||||
if w == nil {
|
||||
return fmt.Errorf("debug summary writer must not be nil")
|
||||
}
|
||||
if err := fileio.WriteJSON(w.root, summaryName(name), v, 0o700, 0o600); err != nil {
|
||||
return fmt.Errorf("write debug summary artifact %q: %w", name, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func (w *SummaryWriter) WriteBytes(name string, v []byte) error {
|
||||
if w == nil {
|
||||
return fmt.Errorf("debug summary writer must not be nil")
|
||||
}
|
||||
if err := fileio.WriteBytes(w.root, summaryName(name), v, 0o700, 0o600); err != nil {
|
||||
return fmt.Errorf("write debug summary artifact %q: %w", name, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
func summaryName(name string) string {
|
||||
name = strings.TrimSpace(name)
|
||||
if name == "" || strings.ContainsAny(name, "/\\") {
|
||||
return "../invalid"
|
||||
}
|
||||
return name
|
||||
}
|
||||
@@ -43,7 +43,7 @@ type RetentionDecisionInput struct {
|
||||
}
|
||||
|
||||
type RedactedEffectiveConfigPayload interface {
|
||||
RedactedDiagnosticsPayload() any
|
||||
RedactedSummaryPayload() any
|
||||
}
|
||||
|
||||
// InvocationMetadata captures non-secret invocation details for diagnostics.
|
||||
@@ -144,7 +144,7 @@ func (r *RunDirectory) WriteRedactedEffectiveConfig(payload RedactedEffectiveCon
|
||||
if payload == nil {
|
||||
return fmt.Errorf("redacted effective config payload must not be nil")
|
||||
}
|
||||
return r.WriteJSONArtifact(ArtifactEffectiveConfig, payload.RedactedDiagnosticsPayload())
|
||||
return r.WriteJSONArtifact(ArtifactEffectiveConfig, payload.RedactedSummaryPayload())
|
||||
}
|
||||
|
||||
func (r *RunDirectory) WriteResolvedPipeline(payload any) error {
|
||||
|
||||
@@ -378,6 +378,6 @@ type fakeRedactedEffectiveConfig struct {
|
||||
payload any
|
||||
}
|
||||
|
||||
func (f fakeRedactedEffectiveConfig) RedactedDiagnosticsPayload() any {
|
||||
func (f fakeRedactedEffectiveConfig) RedactedSummaryPayload() any {
|
||||
return f.payload
|
||||
}
|
||||
|
||||
@@ -3,39 +3,36 @@ package debug
|
||||
import (
|
||||
"strings"
|
||||
|
||||
coreworkspace "gitea.maximumdirect.net/eric/notarius/internal/core/workspace"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/core/fileio"
|
||||
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
|
||||
)
|
||||
|
||||
type WorkspaceRecorder struct {
|
||||
type FilesystemRecorder struct {
|
||||
root string
|
||||
}
|
||||
|
||||
func NewWorkspaceRecorder(settings coreworkspace.Settings, runID string) (pipeline.DebugRecorder, error) {
|
||||
root, err := settings.DebugRunDirectory(runID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
func NewFilesystemRecorder(root string) (pipeline.DebugRecorder, error) {
|
||||
root = strings.TrimSpace(root)
|
||||
if strings.TrimSpace(root) == "" {
|
||||
return pipeline.NoopDebugRecorder(), nil
|
||||
}
|
||||
return &WorkspaceRecorder{root: root}, nil
|
||||
return &FilesystemRecorder{root: root}, nil
|
||||
}
|
||||
|
||||
func (r *WorkspaceRecorder) Enabled() bool {
|
||||
func (r *FilesystemRecorder) Enabled() bool {
|
||||
return r != nil && strings.TrimSpace(r.root) != ""
|
||||
}
|
||||
|
||||
func (r *WorkspaceRecorder) WriteJSON(name string, payload any) error {
|
||||
func (r *FilesystemRecorder) WriteJSON(name string, payload any) error {
|
||||
if !r.Enabled() {
|
||||
return nil
|
||||
}
|
||||
return coreworkspace.WriteJSON(r.root, name, payload)
|
||||
return fileio.WriteJSON(r.root, name, payload, 0o700, 0o600)
|
||||
}
|
||||
|
||||
func (r *WorkspaceRecorder) WriteBytes(name string, data []byte) error {
|
||||
func (r *FilesystemRecorder) WriteBytes(name string, data []byte) error {
|
||||
if !r.Enabled() {
|
||||
return nil
|
||||
}
|
||||
return coreworkspace.WriteBytes(r.root, name, data)
|
||||
return fileio.WriteBytes(r.root, name, data, 0o700, 0o600)
|
||||
}
|
||||
|
||||
28
internal/framework/debug/recorder_test.go
Normal file
28
internal/framework/debug/recorder_test.go
Normal file
@@ -0,0 +1,28 @@
|
||||
package debug
|
||||
|
||||
import (
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestFilesystemRecorderWritesWithinTraceRoot(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
recorder, err := NewFilesystemRecorder(root)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := recorder.WriteBytes("attempt/data", []byte("payload")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
info, err := os.Stat(filepath.Join(root, "attempt", "data"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if info.Mode().Perm() != 0o600 {
|
||||
t.Fatalf("mode=%#o", info.Mode().Perm())
|
||||
}
|
||||
if err := recorder.WriteBytes("../outside", nil); err == nil {
|
||||
t.Fatal("accepted traversal")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user