eric/distributor
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 7 Wiki Activity

Implement structured run reporting

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-03 11:43:50 +00:00
parent 44df38e555
commit f236a8086a
5 changed files with 263 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

148
internal/app/run_output.go
View File

@@ -10,61 +10,125 @@ import (
"gitea.maximumdirect.net/eric/distributor/internal/storage"
)
func writePlanLine(w io.Writer, backend string, plan publish.Plan, planErr error) {
if w == nil {
return
func WriteRunReport(w io.Writer, format OutputFormat, report RunReport) error {
if IsJSONOutput(format) {
return WriteJSONEnvelope(w, "run", len(report.OutputErrors) == 0, report.Warnings, report, report.OutputErrors)
}
if planErr != nil {
destinationID := plan.DestinationID
return writeRunReportText(w, report)
}
func writeRunReportText(w io.Writer, report RunReport) error {
if w == nil {
return nil
}
if err := writeWarnings(w, report.PreambleWarnings); err != nil {
return err
}
if _, err := fmt.Fprintf(w, "Configured pipelines: %d\n", len(report.Pipelines)); err != nil {
return err
}
for _, pipeline := range report.Pipelines {
if err := writeWarnings(w, pipeline.Warnings); err != nil {
return err
}
if _, err := fmt.Fprintf(w, "- pipeline=%s source=%s bundles=%d destinations=%s\n", pipeline.ID, pipeline.SourceBackend, pipeline.BundleCount, destinationIDSummary(pipeline.Destinations)); err != nil {
return err
}
for _, event := range pipeline.events {
if event.warning != nil {
if err := writeWarnings(w, []OutputWarning{*event.warning}); err != nil {
return err
}
continue
}
if event.actionIndex < 0 || event.actionIndex >= len(report.Actions) {
continue
}
writeRunActionLine(w, report.Actions[event.actionIndex])
}
}
_, err := fmt.Fprintln(w, report.Summary.Line())
return err
}
func writeRunActionLine(w io.Writer, action RunActionRecord) {
if action.Action == "error" {
destinationID := action.DestinationID
if destinationID == "" {
destinationID = "unknown"
}
fmt.Fprintf(w, " - bundle=%s destination=%s backend=%s%s action=error reason=%q\n", storage.DisplayPath(plan.BundlePath), destinationID, backend, pathMappingSummary(plan), planErr.Error())
fmt.Fprintf(w, " - bundle=%s destination=%s backend=%s%s action=error reason=%q\n", action.BundlePath, destinationID, action.Backend, pathMappingRecordSummary(action), action.Reason)
return
}
fmt.Fprintf(w, " - bundle=%s destination=%s backend=%s%s action=%s outputs=%s reason=%q\n", storage.DisplayPath(plan.BundlePath), plan.DestinationID, backend, pathMappingSummary(plan), plan.Action, outputSummary(plan.Outputs), plan.Reason)
fmt.Fprintf(w, " - bundle=%s destination=%s backend=%s%s action=%s outputs=%s reason=%q\n", action.BundlePath, action.DestinationID, action.Backend, pathMappingRecordSummary(action), action.Action, outputRecordSummary(action.Outputs), action.Reason)
}
func pathMappingSummary(plan publish.Plan) string {
if plan.PathMapping != config.PathMappingFixed {
func pathMappingRecordSummary(action RunActionRecord) string {
if action.PathMapping != config.PathMappingFixed {
return ""
}
return fmt.Sprintf(" path_mapping=fixed target=%s", storage.DisplayPath(plan.DestinationBundlePath))
return fmt.Sprintf(" path_mapping=fixed target=%s", action.DestinationPath)
}
func writeErrorLine(w io.Writer, bundlePath, destinationID, backend string, err error) {
if w == nil {
return
}
fmt.Fprintf(w, " - bundle=%s destination=%s backend=%s action=error reason=%q\n", storage.DisplayPath(bundlePath), destinationID, backend, err.Error())
}
func outputSummary(outputs []publish.Output) string {
func outputRecordSummary(outputs []RunOutputRecord) string {
if len(outputs) == 0 {
return "none"
}
paths := make([]string, 0, len(outputs))
for _, output := range outputs {
paths = append(paths, output.DestinationPath)
paths = append(paths, output.Path)
}
return strings.Join(paths, ",")
}
type runResult struct {
DryRun bool `json:"dry_run"`
Pipelines []runPipelineResult `json:"pipelines"`
Actions []runActionResult `json:"actions"`
Summary runSummaryResult `json:"summary"`
func destinationIDSummary(ids []string) string {
if len(ids) == 0 {
return "none"
}
return strings.Join(ids, ",")
}
type runPipelineResult struct {
ID string `json:"id"`
SourceBackend string `json:"source_backend"`
BundleCount int `json:"bundle_count"`
Destinations []string `json:"destinations"`
type RunReport struct {
DryRun bool `json:"dry_run"`
Pipelines []RunPipelineSummary `json:"pipelines"`
Actions []RunActionRecord `json:"actions"`
Summary RunSummaryCounters `json:"summary"`
Warnings []OutputWarning `json:"-"`
OutputErrors []OutputError `json:"-"`
PreambleWarnings []OutputWarning `json:"-"`
}
type runActionResult struct {
func (r *RunReport) addWarning(warning OutputWarning) {
r.Warnings = append(r.Warnings, warning)
}
func (r *RunReport) addWarnings(warnings []OutputWarning) {
r.Warnings = append(r.Warnings, warnings...)
}
type RunPipelineSummary struct {
ID string `json:"id"`
SourceBackend string `json:"source_backend"`
BundleCount int `json:"bundle_count"`
Destinations []string `json:"destinations"`
Warnings []OutputWarning `json:"-"`
events []runPipelineEvent
}
type runPipelineEvent struct {
warning *OutputWarning
actionIndex int
}
func warningEvent(warning OutputWarning) runPipelineEvent {
return runPipelineEvent{warning: &warning, actionIndex: -1}
}
func actionEvent(actionIndex int) runPipelineEvent {
return runPipelineEvent{actionIndex: actionIndex}
}
type RunActionRecord struct {
PipelineID string `json:"pipeline_id,omitempty"`
DestinationID string `json:"destination_id"`
Backend string `json:"backend"`
@@ -75,10 +139,10 @@ type runActionResult struct {
Action string `json:"action"`
PrimaryURL string `json:"primary_url,omitempty"`
Reason string `json:"reason,omitempty"`
Outputs []runOutputResult `json:"outputs"`
Outputs []RunOutputRecord `json:"outputs"`
}
type runOutputResult struct {
type RunOutputRecord struct {
Path string `json:"path"`
Kind string `json:"kind"`
SourcePath string `json:"source_path,omitempty"`
@@ -88,13 +152,13 @@ type runOutputResult struct {
Size int64 `json:"size"`
}
func runActionFromPlan(backend string, plan publish.Plan, planErr error) runActionResult {
func runActionFromPlan(backend string, plan publish.Plan, planErr error) RunActionRecord {
if planErr != nil {
destinationID := plan.DestinationID
if destinationID == "" {
destinationID = "unknown"
}
return runActionResult{
return RunActionRecord{
PipelineID: plan.PipelineID,
DestinationID: destinationID,
Backend: backend,
@@ -105,10 +169,10 @@ func runActionFromPlan(backend string, plan publish.Plan, planErr error) runActi
Action: "error",
PrimaryURL: plan.PrimaryURL,
Reason: planErr.Error(),
Outputs: []runOutputResult{},
Outputs: []RunOutputRecord{},
}
}
return runActionResult{
return RunActionRecord{
PipelineID: plan.PipelineID,
DestinationID: plan.DestinationID,
Backend: backend,
@@ -123,8 +187,8 @@ func runActionFromPlan(backend string, plan publish.Plan, planErr error) runActi
}
}
func errorAction(pipelineID, destinationID, backend, bundlePath string, err error) runActionResult {
return runActionResult{
func errorAction(pipelineID, destinationID, backend, bundlePath string, err error) RunActionRecord {
return RunActionRecord{
PipelineID: pipelineID,
DestinationID: destinationID,
Backend: backend,
@@ -132,15 +196,15 @@ func errorAction(pipelineID, destinationID, backend, bundlePath string, err erro
DestinationPath: storage.DisplayPath(bundlePath),
Action: "error",
Reason: err.Error(),
Outputs: []runOutputResult{},
Outputs: []RunOutputRecord{},
}
}
func runOutputsFromPlan(outputs []publish.Output) []runOutputResult {
results := make([]runOutputResult, 0, len(outputs))
func runOutputsFromPlan(outputs []publish.Output) []RunOutputRecord {
results := make([]RunOutputRecord, 0, len(outputs))
for _, output := range outputs {
stateOutput := output.StateOutputFile()
results = append(results, runOutputResult{
results = append(results, RunOutputRecord{
Path: stateOutput.Path,
Kind: stateOutput.Kind,
SourcePath: stateOutput.SourcePath,