335 lines
11 KiB
Go
335 lines
11 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"sort"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/distributor/internal/config"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/state"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/storage"
|
|
)
|
|
|
|
const reconcileStateWalkLimit = 10000
|
|
|
|
type ReconcileStateOptions struct {
|
|
ConfigPath string
|
|
PipelineID string
|
|
DestinationID string
|
|
AllOwners bool
|
|
DryRun bool
|
|
Stdout io.Writer
|
|
OutputFormat OutputFormat
|
|
}
|
|
|
|
type ReconcileStateReport struct {
|
|
PipelineID string `json:"pipeline_id"`
|
|
DestinationID string `json:"destination_id"`
|
|
Backend string `json:"backend"`
|
|
RootPath string `json:"root_path"`
|
|
StateSchema int `json:"state_schema"`
|
|
OwnerScope *ReconcileStateOwnerScope `json:"owner_scope,omitempty"`
|
|
CheckedCount int `json:"checked_count"`
|
|
MissingManagedOutputs []ReconcileStatePath `json:"missing_managed_outputs"`
|
|
UnmanagedEntries []ReconcileStateEntry `json:"unmanaged_entries"`
|
|
Changed bool `json:"changed"`
|
|
WouldChange bool `json:"would_change"`
|
|
DryRun bool `json:"dry_run"`
|
|
}
|
|
|
|
type ReconcileStateOwnerScope struct {
|
|
PipelineID string `json:"pipeline_id"`
|
|
DestinationID string `json:"destination_id"`
|
|
AllOwners bool `json:"all_owners,omitempty"`
|
|
}
|
|
|
|
type ReconcileStatePath struct {
|
|
Path string `json:"path"`
|
|
OwnerScope *ReconcileStateOwnerScope `json:"owner_scope,omitempty"`
|
|
StorageStatus string `json:"storage_status"`
|
|
}
|
|
|
|
type ReconcileStateEntry struct {
|
|
Path string `json:"path"`
|
|
Type string `json:"type"`
|
|
Size int64 `json:"size,omitempty"`
|
|
}
|
|
|
|
func ReconcileState(ctx context.Context, options ReconcileStateOptions) (ReconcileStateReport, error) {
|
|
if err := ValidateOutputFormat(options.OutputFormat); err != nil {
|
|
return ReconcileStateReport{}, err
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return ReconcileStateReport{}, err
|
|
}
|
|
|
|
setup, err := loadRuntimeSetup(options.ConfigPath)
|
|
if err != nil {
|
|
return ReconcileStateReport{}, err
|
|
}
|
|
return reconcileStateSetup(ctx, setup, options)
|
|
}
|
|
|
|
func reconcileStateConfigWithBackendFactory(ctx context.Context, cfg config.Config, options ReconcileStateOptions, provider backendFactoryProvider) (ReconcileStateReport, error) {
|
|
setup, err := runtimeSetupFromConfig("", cfg)
|
|
if err != nil {
|
|
return ReconcileStateReport{}, err
|
|
}
|
|
return reconcileStateSetupWithBackendFactory(ctx, setup, options, provider)
|
|
}
|
|
|
|
func reconcileStateSetup(ctx context.Context, setup runtimeSetup, options ReconcileStateOptions) (ReconcileStateReport, error) {
|
|
return reconcileStateSetupWithBackendFactory(ctx, setup, options, newBackendFactoryWithEnvironment)
|
|
}
|
|
|
|
func reconcileStateSetupWithBackendFactory(ctx context.Context, setup runtimeSetup, options ReconcileStateOptions, provider backendFactoryProvider) (ReconcileStateReport, error) {
|
|
if err := requireReconcileStateScope(options); err != nil {
|
|
return ReconcileStateReport{}, err
|
|
}
|
|
pipeline, ok := findPipeline(setup.Config, options.PipelineID)
|
|
if !ok {
|
|
return ReconcileStateReport{}, PipelineNotFoundError{ID: options.PipelineID}
|
|
}
|
|
destination, ok := findDestination(pipeline, options.DestinationID)
|
|
if !ok {
|
|
return ReconcileStateReport{}, 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 ReconcileStateReport{}, err
|
|
}
|
|
defer closeBackend(destinationBackend)
|
|
|
|
report, err := buildReconcileStateReport(ctx, destinationBackend, pipeline, destination, options)
|
|
if err != nil {
|
|
return ReconcileStateReport{}, err
|
|
}
|
|
if err := WriteReconcileStateReport(options.Stdout, options.OutputFormat, report); err != nil {
|
|
return ReconcileStateReport{}, err
|
|
}
|
|
return report, nil
|
|
}
|
|
|
|
func requireReconcileStateScope(options ReconcileStateOptions) error {
|
|
if options.PipelineID == "" {
|
|
return fmt.Errorf("pipeline id is required")
|
|
}
|
|
if options.DestinationID == "" {
|
|
return fmt.Errorf("destination id is required")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func findDestination(pipeline config.Pipeline, id string) (config.Destination, bool) {
|
|
for _, destination := range pipeline.Destinations {
|
|
if destination.ID == id {
|
|
return destination, true
|
|
}
|
|
}
|
|
return config.Destination{}, false
|
|
}
|
|
|
|
func buildReconcileStateReport(ctx context.Context, backend storage.Backend, pipeline config.Pipeline, destination config.Destination, options ReconcileStateOptions) (ReconcileStateReport, error) {
|
|
statePath, err := storage.StatePath("")
|
|
if err != nil {
|
|
return ReconcileStateReport{}, err
|
|
}
|
|
data, err := backend.ReadFile(ctx, statePath)
|
|
if err != nil {
|
|
return ReconcileStateReport{}, err
|
|
}
|
|
document, err := state.ParseDocument(data)
|
|
if err != nil {
|
|
return ReconcileStateReport{}, err
|
|
}
|
|
|
|
report := ReconcileStateReport{
|
|
PipelineID: pipeline.ID,
|
|
DestinationID: destination.ID,
|
|
Backend: destination.Backend,
|
|
RootPath: destinationRootPath(destination),
|
|
MissingManagedOutputs: []ReconcileStatePath{},
|
|
UnmanagedEntries: []ReconcileStateEntry{},
|
|
DryRun: options.DryRun,
|
|
}
|
|
scope := state.CurrentOwnerScope(pipeline.ID, destination.ID)
|
|
if document.Catalog != nil {
|
|
return reconcileCatalogState(ctx, backend, statePath, *document.Catalog, scope, report, options)
|
|
}
|
|
return ReconcileStateReport{}, unsupportedStateDocumentError(document)
|
|
}
|
|
|
|
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 := state.CatalogManagedOutputPaths(catalog)
|
|
outputs := catalog.Outputs
|
|
if !options.AllOwners {
|
|
outputs = state.CatalogOutputsForOwner(catalog.Outputs, scope)
|
|
}
|
|
missing, err := missingCatalogOutputs(ctx, backend, outputs)
|
|
if err != nil {
|
|
return ReconcileStateReport{}, err
|
|
}
|
|
report.CheckedCount = len(outputs)
|
|
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)
|
|
var next state.CatalogState
|
|
var changed bool
|
|
if options.AllOwners {
|
|
next, changed = state.RemoveMissingCatalogOutputs(catalog, missingPaths)
|
|
} else {
|
|
next, changed = state.RemoveMissingCatalogOwnerOutputs(catalog, scope, missingPaths)
|
|
}
|
|
report.Changed = changed
|
|
if changed {
|
|
next.UpdatedAt = time.Now().UTC()
|
|
if err := state.ValidateCatalog(next); err != nil {
|
|
return ReconcileStateReport{}, err
|
|
}
|
|
if err := writeRepairedState(ctx, backend, statePath, next); err != nil {
|
|
return ReconcileStateReport{}, err
|
|
}
|
|
}
|
|
}
|
|
return report, nil
|
|
}
|
|
|
|
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 {
|
|
if storage.IsNotFound(err) {
|
|
missing = append(missing, ReconcileStatePath{
|
|
Path: output.Path,
|
|
OwnerScope: &ReconcileStateOwnerScope{
|
|
PipelineID: output.PipelineID,
|
|
DestinationID: output.DestinationID,
|
|
},
|
|
StorageStatus: "missing",
|
|
})
|
|
continue
|
|
}
|
|
return nil, err
|
|
}
|
|
}
|
|
return missing, nil
|
|
}
|
|
|
|
func checkManagedOutput(ctx context.Context, backend storage.Backend, path string) error {
|
|
_, err := backend.Stat(ctx, path)
|
|
return err
|
|
}
|
|
|
|
func unmanagedEntries(ctx context.Context, backend storage.Backend, managedPaths []string) ([]ReconcileStateEntry, error) {
|
|
managed := make(map[string]struct{}, len(managedPaths)+1)
|
|
for _, path := range managedPaths {
|
|
managed[path] = struct{}{}
|
|
}
|
|
managed[storage.StateFileName] = struct{}{}
|
|
|
|
entries := make([]ReconcileStateEntry, 0)
|
|
err := backend.Walk(ctx, "", storage.WalkOptions{Recursive: true, Limit: reconcileStateWalkLimit}, func(entry storage.Entry) error {
|
|
if entry.Type == storage.EntryTypeDirectory {
|
|
return nil
|
|
}
|
|
if _, ok := managed[entry.Path]; ok {
|
|
return nil
|
|
}
|
|
entries = append(entries, ReconcileStateEntry{
|
|
Path: entry.Path,
|
|
Type: string(entry.Type),
|
|
Size: entry.Size,
|
|
})
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
sort.Slice(entries, func(i, j int) bool {
|
|
return entries[i].Path < entries[j].Path
|
|
})
|
|
return entries, nil
|
|
}
|
|
|
|
func missingReportPaths(missing []ReconcileStatePath) []string {
|
|
paths := make([]string, 0, len(missing))
|
|
for _, item := range missing {
|
|
paths = append(paths, item.Path)
|
|
}
|
|
return paths
|
|
}
|
|
|
|
func writeRepairedState(ctx context.Context, backend storage.Backend, path string, value any) error {
|
|
data, err := json.MarshalIndent(value, "", " ")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
data = append(data, '\n')
|
|
_, err = backend.WriteFile(ctx, path, data, storage.WriteOptions{Overwrite: true, PreferAtomic: true})
|
|
return err
|
|
}
|
|
|
|
func destinationRootPath(destination config.Destination) string {
|
|
switch destination.Backend {
|
|
case config.BackendS3:
|
|
if destination.Prefix == "" {
|
|
return "."
|
|
}
|
|
return destination.Prefix
|
|
default:
|
|
if destination.Path == "" {
|
|
return "."
|
|
}
|
|
return destination.Path
|
|
}
|
|
}
|
|
|
|
func WriteReconcileStateReport(w io.Writer, format OutputFormat, report ReconcileStateReport) error {
|
|
if IsJSONOutput(format) {
|
|
return WriteJSONEnvelope(w, "reconcile-state", true, nil, report, nil)
|
|
}
|
|
return writeReconcileStateReportText(w, report)
|
|
}
|
|
|
|
func writeReconcileStateReportText(w io.Writer, report ReconcileStateReport) error {
|
|
if w == nil {
|
|
return nil
|
|
}
|
|
status := "unchanged"
|
|
if report.Changed {
|
|
status = "changed"
|
|
} else if report.WouldChange {
|
|
status = "would_change"
|
|
}
|
|
_, err := fmt.Fprintf(w, "Reconcile state: pipeline=%s destination=%s backend=%s root=%s status=%s checked=%d missing=%d unmanaged=%d dry_run=%t\n",
|
|
report.PipelineID,
|
|
report.DestinationID,
|
|
report.Backend,
|
|
report.RootPath,
|
|
status,
|
|
report.CheckedCount,
|
|
len(report.MissingManagedOutputs),
|
|
len(report.UnmanagedEntries),
|
|
report.DryRun,
|
|
)
|
|
return err
|
|
}
|