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

Expose takeover actions in run output

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-18 15:35:06 +00:00
parent 11d1eabe2a
commit eba65018be
18 changed files with 141 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
internal/app/run_output.go
View File

@@ -60,7 +60,7 @@ func writeRunActionLine(w io.Writer, action RunActionRecord) {
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", action.BundlePath, action.DestinationID, action.Backend, pathMappingRecordSummary(action), action.Action, outputRecordSummary(action.Outputs), action.Reason)
fmt.Fprintf(w, " - bundle=%s destination=%s backend=%s%s action=%s%s outputs=%s reason=%q\n", action.BundlePath, action.DestinationID, action.Backend, pathMappingRecordSummary(action), action.Action, takeoverModeRecordSummary(action), outputRecordSummary(action.Outputs), action.Reason)
}
func pathMappingRecordSummary(action RunActionRecord) string {
@@ -70,6 +70,13 @@ func pathMappingRecordSummary(action RunActionRecord) string {
return fmt.Sprintf(" path_mapping=fixed target=%s", action.DestinationPath)
}
func takeoverModeRecordSummary(action RunActionRecord) string {
if action.TakeoverMode == "" {
return ""
}
return fmt.Sprintf(" takeover_mode=%s", action.TakeoverMode)
}
func outputRecordSummary(outputs []RunOutputRecord) string {
if len(outputs) == 0 {
return "none"
@@ -137,6 +144,7 @@ type RunActionRecord struct {
DestinationPath string `json:"destination_path"`
PathMapping string `json:"path_mapping,omitempty"`
Action string `json:"action"`
TakeoverMode string `json:"takeover_mode,omitempty"`
PrimaryURL string `json:"primary_url,omitempty"`
Reason string `json:"reason,omitempty"`
Outputs []RunOutputRecord `json:"outputs"`
@@ -181,12 +189,20 @@ func runActionFromPlan(backend string, plan publish.Plan, planErr error) RunActi
DestinationPath: storage.DisplayPath(plan.DestinationBundlePath),
PathMapping: plan.PathMapping,
Action: string(plan.Action),
TakeoverMode: takeoverModeForAction(plan),
PrimaryURL: plan.PrimaryURL,
Reason: plan.Reason,
Outputs: runOutputsFromPlan(plan.Outputs),
}
}
func takeoverModeForAction(plan publish.Plan) string {
if plan.Action != publish.ActionReplaceTakeover {
return ""
}
return plan.TakeoverMode
}
func errorAction(pipelineID, destinationID, backend, bundlePath string, err error) RunActionRecord {
return RunActionRecord{
PipelineID: pipelineID,

3
internal/app/run_selection.go
View File

@@ -67,6 +67,9 @@ func isDestructiveFixedPathAction(action publish.Action) bool {
}
func fixedPathReplacementWarning(plan publish.Plan) OutputWarning {
if plan.Action == publish.ActionReplaceTakeover {
return OutputWarning{Message: fmt.Sprintf("pipeline=%s destination=%s path_mapping=fixed action=%s takeover_mode=%s replaces destination root for selected_bundle=%s reason=%q", plan.PipelineID, plan.DestinationID, plan.Action, plan.TakeoverMode, storage.DisplayPath(plan.BundlePath), plan.Reason)}
}
return OutputWarning{Message: fmt.Sprintf("pipeline=%s destination=%s path_mapping=fixed action=%s replaces destination root for selected_bundle=%s", plan.PipelineID, plan.DestinationID, plan.Action, storage.DisplayPath(plan.BundlePath))}
}

59
internal/app/run_summary.go
View File

@@ -7,14 +7,15 @@ import (
)
type runSummary struct {
dryRun bool
planned int
publishNew int
replaceOlder int
forceReplace int
skipped int
failures int
fixedPath int
dryRun bool
planned int
publishNew int
replaceOlder int
replaceTakeover int
forceReplace int
skipped int
failures int
fixedPath int
}
func (s *runSummary) recordPlan(action publish.Action) {
@@ -24,6 +25,8 @@ func (s *runSummary) recordPlan(action publish.Action) {
s.publishNew++
case publish.ActionReplaceOlder:
s.replaceOlder++
case publish.ActionReplaceTakeover:
s.replaceTakeover++
case publish.ActionForceReplace:
s.forceReplace++
case publish.ActionSkipSame, publish.ActionSkipDestinationNewer:
@@ -40,19 +43,20 @@ func (s *runSummary) recordFixedPath() {
}
type RunSummaryCounters struct {
Status string `json:"status"`
Planned int `json:"planned"`
PublishNew int `json:"publish_new"`
ReplaceOlder int `json:"replace_older"`
ForceReplace int `json:"force_replace"`
Skipped int `json:"skipped"`
Failed int `json:"failed"`
DryRun bool `json:"dry_run"`
FixedPath int `json:"fixed_path"`
Status string `json:"status"`
Planned int `json:"planned"`
PublishNew int `json:"publish_new"`
ReplaceOlder int `json:"replace_older"`
ReplaceTakeover int `json:"replace_takeover"`
ForceReplace int `json:"force_replace"`
Skipped int `json:"skipped"`
Failed int `json:"failed"`
DryRun bool `json:"dry_run"`
FixedPath int `json:"fixed_path"`
}
func (s RunSummaryCounters) Line() string {
return fmt.Sprintf("Final status: %s planned=%d publish_new=%d replace_older=%d force_replace=%d skipped=%d failed=%d dry_run=%t fixed_path=%d", s.Status, s.Planned, s.PublishNew, s.ReplaceOlder, s.ForceReplace, s.Skipped, s.Failed, s.DryRun, s.FixedPath)
return fmt.Sprintf("Final status: %s planned=%d publish_new=%d replace_older=%d replace_takeover=%d force_replace=%d skipped=%d failed=%d dry_run=%t fixed_path=%d", s.Status, s.Planned, s.PublishNew, s.ReplaceOlder, s.ReplaceTakeover, s.ForceReplace, s.Skipped, s.Failed, s.DryRun, s.FixedPath)
}
func (s runSummary) Result() RunSummaryCounters {
@@ -61,14 +65,15 @@ func (s runSummary) Result() RunSummaryCounters {
status = "failed"
}
return RunSummaryCounters{
Status: status,
Planned: s.planned,
PublishNew: s.publishNew,
ReplaceOlder: s.replaceOlder,
ForceReplace: s.forceReplace,
Skipped: s.skipped,
Failed: s.failures,
DryRun: s.dryRun,
FixedPath: s.fixedPath,
Status: status,
Planned: s.planned,
PublishNew: s.publishNew,
ReplaceOlder: s.replaceOlder,
ReplaceTakeover: s.replaceTakeover,
ForceReplace: s.forceReplace,
Skipped: s.skipped,
Failed: s.failures,
DryRun: s.dryRun,
FixedPath: s.fixedPath,
}
}

66
internal/app/run_test.go
View File

@@ -43,7 +43,7 @@ func TestRunDryRunPrintsConfigSummary(t *testing.T) {
"Configured pipelines: 1",
"- pipeline=reports source=local bundles=1 destinations=archive",
"bundle=. destination=archive backend=local action=publish_new outputs=report.md,summary.txt",
"Final status: ok planned=1 publish_new=1 replace_older=0 force_replace=0 skipped=0 failed=0 dry_run=true",
"Final status: ok planned=1 publish_new=1 replace_older=0 replace_takeover=0 force_replace=0 skipped=0 failed=0 dry_run=true",
} {
if !strings.Contains(output, want) {
t.Fatalf("Run() output = %q, want substring %q", output, want)
@@ -592,8 +592,9 @@ func TestRunFixedPathDryRunWarnsForReplacement(t *testing.T) {
}
output := stdout.String()
for _, want := range []string{
"Warning: pipeline=reports destination=archive path_mapping=fixed action=replace_takeover replaces destination root for selected_bundle=new",
"bundle=new destination=archive backend=local path_mapping=fixed target=. action=replace_takeover",
"Warning: pipeline=reports destination=archive path_mapping=fixed action=replace_takeover takeover_mode=same_pipeline replaces destination root for selected_bundle=new reason=\"destination source id differs from source\"",
"bundle=new destination=archive backend=local path_mapping=fixed target=. action=replace_takeover takeover_mode=same_pipeline outputs=report.md,summary.txt reason=\"destination source id differs from source\"",
"replace_takeover=1",
} {
if !strings.Contains(output, want) {
t.Fatalf("stdout = %q, want substring %q", output, want)
@@ -602,6 +603,61 @@ func TestRunFixedPathDryRunWarnsForReplacement(t *testing.T) {
testutil.AssertFile(t, filepath.Join(destinationRoot, "report.md"), "# Report\nOld.\n")
}
func TestRunJSONIncludesTakeoverActionAndSummary(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
writeSourceBundle(t, sourceRoot, "old", testBundleOptions{
ID: "reports.old",
Created: testutil.DefaultCreated,
Files: []testFile{
{Path: "report.md", Data: "# Report\nOld.\n"},
{Path: "summary.txt", Data: "Old summary\n"},
},
})
configPath := testutil.WriteLocalConfigWithPathMapping(t, sourceRoot, destinationRoot, config.PathMappingFixed)
if err := Run(context.Background(), RunOptions{ConfigPath: configPath}); err != nil {
t.Fatalf("first Run() error = %v", err)
}
writeSourceBundle(t, sourceRoot, "new", testBundleOptions{
ID: "reports.new",
Created: testutil.DefaultCreated.Add(time.Hour),
Files: []testFile{
{Path: "report.md", Data: "# Report\nNew.\n"},
{Path: "summary.txt", Data: "New summary\n"},
},
})
var stdout bytes.Buffer
err := Run(context.Background(), RunOptions{
ConfigPath: configPath,
DryRun: true,
Stdout: &stdout,
OutputFormat: OutputFormatJSON,
})
if err != nil {
t.Fatalf("Run() error = %v", err)
}
result := decodeAppResult(t, stdout.String())
actions, ok := result["actions"].([]any)
if !ok || len(actions) != 1 {
t.Fatalf("actions = %#v, want one action", result["actions"])
}
action, ok := actions[0].(map[string]any)
if !ok {
t.Fatalf("action = %#v, want object", actions[0])
}
if action["action"] != "replace_takeover" || action["takeover_mode"] != "same_pipeline" || action["reason"] != "destination source id differs from source" {
t.Fatalf("action = %#v, want takeover action metadata", action)
}
summary, ok := result["summary"].(map[string]any)
if !ok {
t.Fatalf("summary = %#v, want object", result["summary"])
}
if summary["replace_takeover"] != float64(1) || summary["replace_older"] != float64(0) || summary["force_replace"] != float64(0) {
t.Fatalf("summary = %#v, want takeover counter only", summary)
}
}
func TestRunFixedPathReplacesOlderManagedState(t *testing.T) {
sourceRoot := t.TempDir()
destinationRoot := t.TempDir()
@@ -1221,7 +1277,7 @@ func TestRunContinuesAfterDestinationFailure(t *testing.T) {
for _, want := range []string{
"destination=archive-one backend=local action=error",
"destination=archive-two backend=local action=publish_new",
"Final status: failed planned=1 publish_new=1 replace_older=0 force_replace=0 skipped=0 failed=1 dry_run=false",
"Final status: failed planned=1 publish_new=1 replace_older=0 replace_takeover=0 force_replace=0 skipped=0 failed=1 dry_run=false",
} {
if !strings.Contains(output, want) {
t.Fatalf("stdout = %q, want substring %q", output, want)
@@ -1662,7 +1718,7 @@ func TestRunExercisesRemoteBackendShapesThroughCommonPath(t *testing.T) {
"pipeline=local-to-ssh source=local",
"destination=ssh-archive backend=ssh action=publish_new",
"pipeline=ssh-to-local source=ssh",
"Final status: ok planned=4 publish_new=4 replace_older=0 force_replace=0 skipped=0 failed=0 dry_run=true",
"Final status: ok planned=4 publish_new=4 replace_older=0 replace_takeover=0 force_replace=0 skipped=0 failed=0 dry_run=true",
} {
if !strings.Contains(dryRunOutput.String(), want) {
t.Fatalf("dry-run output = %q, want substring %q", dryRunOutput.String(), want)

2
internal/cli/root_test.go
View File

@@ -635,7 +635,7 @@ func TestExecuteRunDryRun(t *testing.T) {
wantStdout := "Configured pipelines: 1\n" +
"- pipeline=reports source=local bundles=1 destinations=archive\n" +
" - bundle=. destination=archive backend=local action=publish_new outputs=report.md,summary.txt reason=\"destination state is absent\"\n" +
"Final status: ok planned=1 publish_new=1 replace_older=0 force_replace=0 skipped=0 failed=0 dry_run=true fixed_path=0\n"
"Final status: ok planned=1 publish_new=1 replace_older=0 replace_takeover=0 force_replace=0 skipped=0 failed=0 dry_run=true fixed_path=0\n"
if got := stdout.String(); got != wantStdout {
t.Fatalf("stdout = %q, want %q", got, wantStdout)
}

2
internal/publish/plan.go
View File

@@ -62,6 +62,7 @@ type Plan struct {
StateMode string
OwnerScope state.OwnerScope
Reconciliation config.ReconciliationPolicy
TakeoverMode string
Outputs []Output
ExistingState *state.DistributorState
ExistingSharedRoot *state.SharedRootState
@@ -121,6 +122,7 @@ func Build(ctx context.Context, req Request) (Plan, error) {
StateMode: stateMode,
OwnerScope: state.CurrentOwnerScope(req.PipelineID, req.DestinationID),
Reconciliation: reconciliation,
TakeoverMode: normalizeTakeover(req.Takeover).Mode,
Outputs: outputs,
ExistingState: status.State,
ExistingSharedRoot: status.SharedRoot,