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

Support catalog maintenance commands

Browse Source
This commit is contained in:
Eric Rakestraw
2026-06-19 15:58:36 +00:00
parent 91b72478d5
commit cf7e62733e
8 changed files with 295 additions and 349 deletions
Download Patch File Download Diff File Expand all files Collapse all files

94
internal/app/reconcile_state.go
View File

@@ -158,65 +158,25 @@ func buildReconcileStateReport(ctx context.Context, backend storage.Backend, pip
DryRun: options.DryRun,
}
scope := state.CurrentOwnerScope(pipeline.ID, destination.ID)
if document.SingleOwner != nil {
return reconcileSingleOwnerState(ctx, backend, statePath, *document.SingleOwner, scope, report, options)
}
if document.SharedRoot != nil {
return reconcileSharedRootState(ctx, backend, statePath, *document.SharedRoot, scope, report, options)
if document.Catalog != nil {
return reconcileCatalogState(ctx, backend, statePath, *document.Catalog, scope, report, options)
}
return ReconcileStateReport{}, unsupportedStateDocumentError(document)
}
func reconcileSingleOwnerState(ctx context.Context, backend storage.Backend, statePath string, destinationState state.DistributorState, scope state.OwnerScope, report ReconcileStateReport, options ReconcileStateOptions) (ReconcileStateReport, error) {
if destinationState.PipelineID != scope.PipelineID || destinationState.DestinationID != scope.DestinationID {
return ReconcileStateReport{}, fmt.Errorf("state owner is %s/%s, not %s/%s", destinationState.PipelineID, destinationState.DestinationID, scope.PipelineID, scope.DestinationID)
}
report.StateSchema = destinationState.SchemaVersion
report.OwnerScope = &ReconcileStateOwnerScope{PipelineID: scope.PipelineID, DestinationID: scope.DestinationID}
managed := state.ManagedOutputPaths(destinationState)
missing, err := missingSingleOwnerOutputs(ctx, backend, destinationState.Outputs)
if err != nil {
return ReconcileStateReport{}, err
}
report.CheckedCount = len(managed)
report.MissingManagedOutputs = missing
unmanaged, err := unmanagedEntries(ctx, backend, managed)
if err != nil {
return ReconcileStateReport{}, err
}
report.UnmanagedEntries = unmanaged
report.WouldChange = options.DryRun && len(missing) > 0
if !options.DryRun && len(missing) > 0 {
missingPaths := missingReportPaths(missing)
next, changed := state.RemoveMissingOutputs(destinationState, missingPaths)
report.Changed = changed
if changed {
next.UpdatedAt = time.Now().UTC()
if err := state.Validate(next); err != nil {
return ReconcileStateReport{}, err
}
if err := writeRepairedState(ctx, backend, statePath, next); err != nil {
return ReconcileStateReport{}, err
}
}
}
return report, nil
}
func reconcileSharedRootState(ctx context.Context, backend storage.Backend, statePath string, sharedRoot state.SharedRootState, scope state.OwnerScope, report ReconcileStateReport, options ReconcileStateOptions) (ReconcileStateReport, error) {
report.StateSchema = sharedRoot.SchemaVersion
func reconcileCatalogState(ctx context.Context, backend storage.Backend, statePath string, catalog state.CatalogState, scope state.OwnerScope, report ReconcileStateReport, options ReconcileStateOptions) (ReconcileStateReport, error) {
report.StateSchema = catalog.SchemaVersion
report.OwnerScope = &ReconcileStateOwnerScope{
PipelineID: scope.PipelineID,
DestinationID: scope.DestinationID,
AllOwners: options.AllOwners,
}
managed := sharedRoot.AllManagedOutputPaths()
outputs := sharedRoot.Outputs
managed := state.CatalogManagedOutputPaths(catalog)
outputs := catalog.Outputs
if !options.AllOwners {
outputs = sharedRootOutputsForOwner(sharedRoot.Outputs, scope)
outputs = state.CatalogOutputsForOwner(catalog.Outputs, scope)
}
missing, err := missingSharedRootOutputs(ctx, backend, outputs)
missing, err := missingCatalogOutputs(ctx, backend, outputs)
if err != nil {
return ReconcileStateReport{}, err
}
@@ -231,17 +191,17 @@ func reconcileSharedRootState(ctx context.Context, backend storage.Backend, stat
if !options.DryRun && len(missing) > 0 {
missingPaths := missingReportPaths(missing)
var next state.SharedRootState
var next state.CatalogState
var changed bool
if options.AllOwners {
next, changed = state.RemoveMissingSharedRootOutputs(sharedRoot, missingPaths)
next, changed = state.RemoveMissingCatalogOutputs(catalog, missingPaths)
} else {
next, changed = state.RemoveMissingSharedRootOwnerOutputs(sharedRoot, scope, missingPaths)
next, changed = state.RemoveMissingCatalogOwnerOutputs(catalog, scope, missingPaths)
}
report.Changed = changed
if changed {
next.UpdatedAt = time.Now().UTC()
if err := state.ValidateSharedRoot(next); err != nil {
if err := state.ValidateCatalog(next); err != nil {
return ReconcileStateReport{}, err
}
if err := writeRepairedState(ctx, backend, statePath, next); err != nil {
@@ -252,21 +212,7 @@ func reconcileSharedRootState(ctx context.Context, backend storage.Backend, stat
return report, nil
}
func missingSingleOwnerOutputs(ctx context.Context, backend storage.Backend, outputs []state.OutputFile) ([]ReconcileStatePath, error) {
missing := make([]ReconcileStatePath, 0)
for _, output := range outputs {
if err := checkManagedOutput(ctx, backend, output.Path); err != nil {
if storage.IsNotFound(err) {
missing = append(missing, ReconcileStatePath{Path: output.Path, StorageStatus: "missing"})
continue
}
return nil, err
}
}
return missing, nil
}
func missingSharedRootOutputs(ctx context.Context, backend storage.Backend, outputs []state.SharedRootOutputFile) ([]ReconcileStatePath, error) {
func missingCatalogOutputs(ctx context.Context, backend storage.Backend, outputs []state.CatalogOutputFile) ([]ReconcileStatePath, error) {
missing := make([]ReconcileStatePath, 0)
for _, output := range outputs {
if err := checkManagedOutput(ctx, backend, output.Path); err != nil {
@@ -274,8 +220,8 @@ func missingSharedRootOutputs(ctx context.Context, backend storage.Backend, outp
missing = append(missing, ReconcileStatePath{
Path: output.Path,
OwnerScope: &ReconcileStateOwnerScope{
PipelineID: output.Owner.PipelineID,
DestinationID: output.Owner.DestinationID,
PipelineID: output.PipelineID,
DestinationID: output.DestinationID,
},
StorageStatus: "missing",
})
@@ -323,16 +269,6 @@ func unmanagedEntries(ctx context.Context, backend storage.Backend, managedPaths
return entries, nil
}
func sharedRootOutputsForOwner(outputs []state.SharedRootOutputFile, scope state.OwnerScope) []state.SharedRootOutputFile {
selected := make([]state.SharedRootOutputFile, 0, len(outputs))
for _, output := range outputs {
if output.Owner == scope {
selected = append(selected, output)
}
}
return selected
}
func missingReportPaths(missing []ReconcileStatePath) []string {
paths := make([]string, 0, len(missing))
for _, item := range missing {