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

Add managed output pruning

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-08 19:32:57 +00:00
parent c67ecf86a9
commit 6daddad543
5 changed files with 392 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

188
internal/app/prune.go
View File

@@ -1,13 +1,23 @@
package app
import (
"context"
"fmt"
"time"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/state"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
)
type PruneOptions struct {
ConfigPath string
PipelineID string
DestinationID string
DryRun bool
Now time.Time
}
type PrunePlanOptions struct {
PipelineID string
DestinationID string
@@ -24,6 +34,23 @@ type PrunePlanReport struct {
PreservedOutputs []PruneOutputRecord `json:"preserved_outputs"`
}
type PruneReport struct {
PipelineID string `json:"pipeline_id"`
DestinationID string `json:"destination_id"`
Backend string `json:"backend"`
RootPath string `json:"root_path"`
OwnerScope PruneOwnerScope `json:"owner_scope"`
Enabled bool `json:"enabled"`
CheckedCount int `json:"checked_count"`
PlannedOutputs []PruneOutputRecord `json:"planned_outputs"`
DeletedOutputs []PruneOutputRecord `json:"deleted_outputs"`
PreservedOutputs []PruneOutputRecord `json:"preserved_outputs"`
FailedOutput *PruneOutputRecord `json:"failed_output,omitempty"`
StateChanged bool `json:"state_changed"`
WouldChange bool `json:"would_change"`
DryRun bool `json:"dry_run"`
}
type PruneOwnerScope struct {
PipelineID string `json:"pipeline_id"`
DestinationID string `json:"destination_id"`
@@ -35,6 +62,167 @@ type PruneOutputRecord struct {
Owner *PruneOwnerScope `json:"owner,omitempty"`
}
func Prune(ctx context.Context, options PruneOptions) (PruneReport, error) {
if err := ctx.Err(); err != nil {
return PruneReport{}, err
}
setup, err := loadRuntimeSetup(options.ConfigPath)
if err != nil {
return PruneReport{}, err
}
return pruneSetup(ctx, setup, options)
}
func pruneConfigWithBackendFactory(ctx context.Context, cfg config.Config, options PruneOptions, provider backendFactoryProvider) (PruneReport, error) {
setup, err := runtimeSetupFromConfig("", cfg)
if err != nil {
return PruneReport{}, err
}
return pruneSetupWithBackendFactory(ctx, setup, options, provider)
}
func pruneSetup(ctx context.Context, setup runtimeSetup, options PruneOptions) (PruneReport, error) {
return pruneSetupWithBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
}
func pruneSetupWithBackendFactory(ctx context.Context, setup runtimeSetup, options PruneOptions, provider backendFactoryProvider) (PruneReport, error) {
if err := requirePruneScope(options); err != nil {
return PruneReport{}, err
}
pipeline, ok := findPipeline(setup.Config, options.PipelineID)
if !ok {
return PruneReport{}, PipelineNotFoundError{ID: options.PipelineID}
}
destination, ok := findDestination(pipeline, options.DestinationID)
if !ok {
return PruneReport{}, fmt.Errorf("pipeline %s destination %s not found", options.PipelineID, options.DestinationID)
}
backends := provider(setup.Environment)
destinationBackend, err := backends.openDestination(ctx, destination)
if err != nil {
return PruneReport{}, err
}
defer closeBackend(destinationBackend)
return executePrune(ctx, destinationBackend, pipeline, destination, options)
}
func requirePruneScope(options PruneOptions) error {
if options.PipelineID == "" {
return fmt.Errorf("pipeline id is required")
}
if options.DestinationID == "" {
return fmt.Errorf("destination id is required")
}
return nil
}
func executePrune(ctx context.Context, backend storage.Backend, pipeline config.Pipeline, destination config.Destination, options PruneOptions) (PruneReport, error) {
now := options.Now
if now.IsZero() {
now = time.Now().UTC()
} else {
now = now.UTC()
}
statePath, err := storage.StatePath("")
if err != nil {
return PruneReport{}, err
}
data, err := backend.ReadFile(ctx, statePath)
if err != nil {
return PruneReport{}, err
}
document, err := state.ParseDocument(data)
if err != nil {
return PruneReport{}, err
}
plan, err := PlanPrune(document, destination.Retention.Prune, PrunePlanOptions{
PipelineID: pipeline.ID,
DestinationID: destination.ID,
Now: now,
})
if err != nil {
return PruneReport{}, err
}
report := PruneReport{
PipelineID: pipeline.ID,
DestinationID: destination.ID,
Backend: destination.Backend,
RootPath: destinationRootPath(destination),
OwnerScope: plan.OwnerScope,
Enabled: plan.Enabled,
CheckedCount: plan.CheckedCount,
PlannedOutputs: plan.PrunedOutputs,
DeletedOutputs: []PruneOutputRecord{},
PreservedOutputs: plan.PreservedOutputs,
DryRun: options.DryRun,
}
report.WouldChange = options.DryRun && len(report.PlannedOutputs) > 0
if options.DryRun || len(report.PlannedOutputs) == 0 {
return report, nil
}
deletedPaths := make([]string, 0, len(report.PlannedOutputs))
for _, output := range report.PlannedOutputs {
err := backend.DeleteManagedOutputs(ctx, "", []string{output.Path}, storage.DeleteOptions{
IgnoreMissing: true,
PruneEmptyDirs: true,
})
if err != nil {
failed := output
report.FailedOutput = &failed
if len(deletedPaths) > 0 {
changed, writeErr := removePrunedStateRecords(ctx, backend, statePath, document, state.CurrentOwnerScope(pipeline.ID, destination.ID), deletedPaths, now)
report.StateChanged = changed
report.DeletedOutputs = report.PlannedOutputs[:len(deletedPaths)]
if writeErr != nil {
return report, writeErr
}
}
return report, err
}
deletedPaths = append(deletedPaths, output.Path)
}
changed, err := removePrunedStateRecords(ctx, backend, statePath, document, state.CurrentOwnerScope(pipeline.ID, destination.ID), deletedPaths, now)
report.StateChanged = changed
report.DeletedOutputs = report.PlannedOutputs
return report, err
}
func removePrunedStateRecords(ctx context.Context, backend storage.Backend, statePath string, document state.StateDocument, scope state.OwnerScope, paths []string, now time.Time) (bool, error) {
if len(paths) == 0 {
return false, nil
}
if document.SingleOwner != nil {
next, changed := state.RemoveMissingOutputs(*document.SingleOwner, paths)
if !changed {
return false, nil
}
next.UpdatedAt = now
if err := state.Validate(next); err != nil {
return false, err
}
return true, writeRepairedState(ctx, backend, statePath, next)
}
if document.SharedRoot != nil {
next, changed := state.RemoveMissingSharedRootOwnerOutputs(*document.SharedRoot, scope, paths)
if !changed {
return false, nil
}
next.UpdatedAt = now
if err := state.ValidateSharedRoot(next); err != nil {
return false, err
}
return true, writeRepairedState(ctx, backend, statePath, next)
}
return false, fmt.Errorf("destination state document is empty")
}
func PlanPrune(document state.StateDocument, policy config.PrunePolicy, options PrunePlanOptions) (PrunePlanReport, error) {
scope := state.CurrentOwnerScope(options.PipelineID, options.DestinationID)
report := PrunePlanReport{

191
internal/app/prune_test.go
View File

@@ -1,12 +1,16 @@
package app
import (
"context"
"encoding/json"
"strings"
"testing"
"time"
"gitea.maximumdirect.net/eric/distributor/internal/config"
"gitea.maximumdirect.net/eric/distributor/internal/state"
"gitea.maximumdirect.net/eric/distributor/internal/storage"
"gitea.maximumdirect.net/eric/distributor/internal/storage/fake"
"gitea.maximumdirect.net/eric/distributor/internal/testutil"
)
@@ -72,6 +76,134 @@ func TestPlanPruneSharedRootCurrentOwnerOnly(t *testing.T) {
}
}
func TestPruneDryRunReportsPlannedDeletesWithoutDeletingOrRewritingState(t *testing.T) {
now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)
backend := fake.New()
cfg := pruneS3Config(t, pruneOlderThanPolicy(48*time.Hour))
original := pruneSingleOwnerState(now)
writeFakeSingleOwnerStateForPrune(t, backend, original)
testutil.WriteFakeFile(t, backend, "unmanaged.txt", "keep")
report, err := pruneConfigWithBackendFactory(context.Background(), cfg, PruneOptions{
PipelineID: "reports",
DestinationID: "archive",
DryRun: true,
Now: now,
}, fakeBackendFactoryProvider(t, map[string]storage.Backend{"s3:reports": backend}))
if err != nil {
t.Fatalf("pruneConfigWithBackendFactory() error = %v", err)
}
if !report.WouldChange || report.StateChanged || len(report.DeletedOutputs) != 0 {
t.Fatalf("report would_change=%t state_changed=%t deleted=%d, want dry-run only", report.WouldChange, report.StateChanged, len(report.DeletedOutputs))
}
if got, want := pruneRecordPaths(report.PlannedOutputs), "old.txt"; got != want {
t.Fatalf("planned outputs = %q, want %q", got, want)
}
testutil.AssertFakeFile(t, backend, "old.txt", "managed")
testutil.AssertFakeFile(t, backend, "fresh.txt", "managed")
testutil.AssertFakeFile(t, backend, "unmanaged.txt", "keep")
destinationState := readFakeSingleOwnerState(t, backend)
if got := strings.Join(state.ManagedOutputPaths(destinationState), ","); got != "old.txt,fresh.txt" {
t.Fatalf("state outputs = %q, want original outputs", got)
}
if !destinationState.UpdatedAt.Equal(original.UpdatedAt) {
t.Fatalf("state updated_at = %s, want original %s", destinationState.UpdatedAt, original.UpdatedAt)
}
}
func TestPruneApplyDeletesOnlyManagedOutputsAndUpdatesState(t *testing.T) {
now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)
backend := fake.New()
cfg := pruneS3Config(t, pruneOlderThanPolicy(48*time.Hour))
writeFakeSingleOwnerStateForPrune(t, backend, pruneSingleOwnerState(now))
testutil.WriteFakeFile(t, backend, "unmanaged.txt", "keep")
report, err := pruneConfigWithBackendFactory(context.Background(), cfg, PruneOptions{
PipelineID: "reports",
DestinationID: "archive",
Now: now,
}, fakeBackendFactoryProvider(t, map[string]storage.Backend{"s3:reports": backend}))
if err != nil {
t.Fatalf("pruneConfigWithBackendFactory() error = %v", err)
}
if !report.StateChanged || report.WouldChange {
t.Fatalf("report state_changed=%t would_change=%t, want applied change", report.StateChanged, report.WouldChange)
}
if got, want := pruneRecordPaths(report.DeletedOutputs), "old.txt"; got != want {
t.Fatalf("deleted outputs = %q, want %q", got, want)
}
testutil.AssertFakeMissing(t, backend, "old.txt")
testutil.AssertFakeFile(t, backend, "fresh.txt", "managed")
testutil.AssertFakeFile(t, backend, "unmanaged.txt", "keep")
assertFakeStateExists(t, backend)
destinationState := readFakeSingleOwnerState(t, backend)
if got := strings.Join(state.ManagedOutputPaths(destinationState), ","); got != "fresh.txt" {
t.Fatalf("state outputs = %q, want fresh.txt", got)
}
if !destinationState.UpdatedAt.Equal(now) {
t.Fatalf("state updated_at = %s, want %s", destinationState.UpdatedAt, now)
}
}
func TestPruneApplyPreservesStateForFailedDeletes(t *testing.T) {
now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)
backend := fake.New()
keepLatest := 0
cfg := pruneS3Config(t, config.PrunePolicy{Enabled: true, KeepLatest: &keepLatest})
writeFakeSingleOwnerStateForPrune(t, backend, pruneSingleOwnerState(now))
failingBackend := failingDeleteBackend{Backend: backend, failPath: "fresh.txt"}
report, err := pruneConfigWithBackendFactory(context.Background(), cfg, PruneOptions{
PipelineID: "reports",
DestinationID: "archive",
Now: now,
}, fakeBackendFactoryProvider(t, map[string]storage.Backend{"s3:reports": failingBackend}))
if err == nil {
t.Fatal("pruneConfigWithBackendFactory() error = nil, want delete failure")
}
if report.FailedOutput == nil || report.FailedOutput.Path != "fresh.txt" {
t.Fatalf("failed output = %#v, want fresh.txt", report.FailedOutput)
}
if got, want := pruneRecordPaths(report.DeletedOutputs), "old.txt"; got != want {
t.Fatalf("deleted outputs = %q, want %q", got, want)
}
testutil.AssertFakeMissing(t, backend, "old.txt")
testutil.AssertFakeFile(t, backend, "fresh.txt", "managed")
assertFakeStateExists(t, backend)
destinationState := readFakeSingleOwnerState(t, backend)
if got := strings.Join(state.ManagedOutputPaths(destinationState), ","); got != "fresh.txt" {
t.Fatalf("state outputs = %q, want only failed output preserved", got)
}
}
func TestPruneSharedRootPreservesOtherOwnersWhenScopedToCurrentOwner(t *testing.T) {
now := time.Date(2026, 6, 8, 12, 0, 0, 0, time.UTC)
backend := fake.New()
keepLatest := 0
cfg := pruneS3Config(t, config.PrunePolicy{Enabled: true, KeepLatest: &keepLatest})
writeFakeSharedRootStateForApp(t, backend, pruneSharedRootState(now))
testutil.WriteFakeFile(t, backend, "unmanaged.txt", "keep")
report, err := pruneConfigWithBackendFactory(context.Background(), cfg, PruneOptions{
PipelineID: "reports",
DestinationID: "archive",
Now: now,
}, fakeBackendFactoryProvider(t, map[string]storage.Backend{"s3:reports": backend}))
if err != nil {
t.Fatalf("pruneConfigWithBackendFactory() error = %v", err)
}
if got, want := pruneRecordPaths(report.DeletedOutputs), "archive.txt"; got != want {
t.Fatalf("deleted outputs = %q, want %q", got, want)
}
testutil.AssertFakeMissing(t, backend, "archive.txt")
testutil.AssertFakeFile(t, backend, "html.txt", "old")
testutil.AssertFakeFile(t, backend, "unmanaged.txt", "keep")
sharedRoot := readFakeSharedRootStateForApp(t, backend)
if got := strings.Join(sharedRoot.AllManagedOutputPaths(), ","); got != "html.txt" {
t.Fatalf("shared-root outputs = %q, want other owner output preserved", got)
}
}
func pruneSingleOwnerState(now time.Time) state.DistributorState {
manifest := testutil.ValidManifest(testutil.BundleOptions{})
publishedAt := now.Add(-96 * time.Hour)
@@ -160,3 +292,62 @@ func pruneRecordPaths(records []PruneOutputRecord) string {
}
return strings.Join(paths, ",")
}
func pruneS3Config(t *testing.T, policy config.PrunePolicy) config.Config {
t.Helper()
cfg := config.Config{Pipelines: []config.Pipeline{{
ID: "reports",
Source: config.Backend{Backend: config.BackendLocal, Path: t.TempDir()},
Destinations: []config.Destination{{
ID: "archive",
Backend: config.BackendS3,
Bucket: "reports",
Retention: config.RetentionPolicy{
Prune: policy,
},
}},
}}}
config.ApplyDefaults(&cfg)
return cfg
}
func pruneOlderThanPolicy(duration time.Duration) config.PrunePolicy {
value := config.Duration(duration)
return config.PrunePolicy{
Enabled: true,
OlderThan: &value,
}
}
func writeFakeSingleOwnerStateForPrune(t *testing.T, backend *fake.Backend, destinationState state.DistributorState) {
t.Helper()
data, err := json.MarshalIndent(destinationState, "", " ")
if err != nil {
t.Fatalf("marshal single-owner state: %v", err)
}
testutil.WriteFakeFile(t, backend, storage.StateFileName, string(append(data, '\n')))
for _, output := range destinationState.Outputs {
testutil.WriteFakeFile(t, backend, output.Path, "managed")
}
}
func assertFakeStateExists(t *testing.T, backend *fake.Backend) {
t.Helper()
if _, err := backend.Stat(context.Background(), storage.StateFileName); err != nil {
t.Fatalf("state file stat error = %v", err)
}
}
type failingDeleteBackend struct {
storage.Backend
failPath string
}
func (b failingDeleteBackend) DeleteManagedOutputs(ctx context.Context, bundlePath string, managedOutputPaths []string, opts storage.DeleteOptions) error {
for _, path := range managedOutputPaths {
if path == b.failPath {
return storage.NewError(storage.OpDeleteManagedOutputs, "fake", path, storage.ErrPermission, nil)
}
}
return b.Backend.DeleteManagedOutputs(ctx, bundlePath, managedOutputPaths, opts)
}