563 lines
17 KiB
Go
563 lines
17 KiB
Go
package app
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"strings"
|
|
|
|
"gitea.maximumdirect.net/eric/distributor/internal/bundle"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/config"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/notify"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/publish"
|
|
"gitea.maximumdirect.net/eric/distributor/internal/storage"
|
|
)
|
|
|
|
type RunOptions struct {
|
|
ConfigPath string
|
|
DryRun bool
|
|
Force bool
|
|
Stdout io.Writer
|
|
OutputFormat OutputFormat
|
|
Notifier notify.Notifier
|
|
}
|
|
|
|
func Run(ctx context.Context, options RunOptions) error {
|
|
if err := ValidateOutputFormat(options.OutputFormat); err != nil {
|
|
return err
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
|
|
configPath := options.ConfigPath
|
|
if configPath == "" {
|
|
configPath = config.DefaultConfigPath
|
|
}
|
|
cfg, err := config.LoadFile(configPath)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return runConfig(ctx, cfg, options)
|
|
}
|
|
|
|
func runConfig(ctx context.Context, cfg config.Config, options RunOptions) error {
|
|
return runConfigWithBackendFactory(ctx, cfg, options, newBackendFactoryWithEnvironment)
|
|
}
|
|
|
|
type backendFactoryProvider func(config.Environment) *backendFactory
|
|
|
|
func runConfigWithBackendFactory(ctx context.Context, cfg config.Config, options RunOptions, provider backendFactoryProvider) error {
|
|
notifier := options.Notifier
|
|
if notifier == nil {
|
|
notifier = notify.Noop{}
|
|
}
|
|
jsonOutput := IsJSONOutput(options.OutputFormat)
|
|
summary := runSummary{dryRun: options.DryRun}
|
|
result := runResult{
|
|
DryRun: options.DryRun,
|
|
Pipelines: []runPipelineResult{},
|
|
Actions: []runActionResult{},
|
|
}
|
|
var warnings []OutputWarning
|
|
var failures runFailures
|
|
secretLoad, err := config.LoadSecretEnvironment(cfg.Secrets.Directory, nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
secretWarnings := secretConflictWarnings(secretLoad.Conflicts)
|
|
if jsonOutput {
|
|
warnings = append(warnings, secretWarnings...)
|
|
} else if options.Stdout != nil {
|
|
if err := writeWarnings(options.Stdout, secretWarnings); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
backends := provider(secretLoad.Environment)
|
|
backends.readOnlyKnownHosts = options.DryRun
|
|
transforms := newTransformRegistry()
|
|
if options.Stdout != nil && !jsonOutput {
|
|
if _, err := fmt.Fprintf(options.Stdout, "Configured pipelines: %d\n", len(cfg.Pipelines)); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
for _, pipeline := range cfg.Pipelines {
|
|
pipelineWarnings := sshWarnings(pipeline)
|
|
if jsonOutput {
|
|
warnings = append(warnings, pipelineWarnings...)
|
|
} else if options.Stdout != nil {
|
|
if err := writeWarnings(options.Stdout, pipelineWarnings); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
sourceBackend, err := backends.openSource(ctx, pipeline.Source)
|
|
if err != nil {
|
|
return fmt.Errorf("pipeline %s source backend %s: %w", pipeline.ID, pipeline.Source.Backend, err)
|
|
}
|
|
bundles, err := bundle.Discover(ctx, sourceBackend, "")
|
|
if err != nil {
|
|
closeBackend(sourceBackend)
|
|
return fmt.Errorf("pipeline %s source backend %s discover source bundles: %w", pipeline.ID, pipeline.Source.Backend, err)
|
|
}
|
|
result.Pipelines = append(result.Pipelines, runPipelineResult{
|
|
ID: pipeline.ID,
|
|
SourceBackend: pipeline.Source.Backend,
|
|
BundleCount: len(bundles),
|
|
Destinations: destinationIDs(pipeline.Destinations),
|
|
})
|
|
if options.Stdout != nil && !jsonOutput {
|
|
if _, err := fmt.Fprintf(options.Stdout, "- pipeline=%s source=%s bundles=%d destinations=%s\n", pipeline.ID, pipeline.Source.Backend, len(bundles), destinationSummary(pipeline.Destinations)); err != nil {
|
|
closeBackend(sourceBackend)
|
|
return err
|
|
}
|
|
}
|
|
for _, sourceBundle := range bundles {
|
|
for _, destination := range pipeline.Destinations {
|
|
destinationBackend, err := backends.openDestination(ctx, destination)
|
|
if err != nil {
|
|
failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err)
|
|
summary.recordFailure()
|
|
if jsonOutput {
|
|
result.Actions = append(result.Actions, errorAction(pipeline.ID, destination.ID, destination.Backend, sourceBundle.RootRelativePath, err))
|
|
} else if options.Stdout != nil {
|
|
writeErrorLine(options.Stdout, sourceBundle.RootRelativePath, destination.ID, destination.Backend, err)
|
|
}
|
|
continue
|
|
}
|
|
closeDestination := true
|
|
deferCloseDestination := func() {
|
|
if closeDestination {
|
|
closeBackend(destinationBackend)
|
|
closeDestination = false
|
|
}
|
|
}
|
|
req := publish.Request{
|
|
PipelineID: pipeline.ID,
|
|
DestinationID: destination.ID,
|
|
SourceBundle: sourceBundle,
|
|
SourceBackend: sourceBackend,
|
|
DestinationBackend: destinationBackend,
|
|
DestinationBundlePath: sourceBundle.RootRelativePath,
|
|
Publish: *destination.Publish,
|
|
Transform: destination.Transform,
|
|
Transformers: transforms,
|
|
Transfer: destination.Transfer,
|
|
DistributorVersion: Version,
|
|
Force: options.Force,
|
|
}
|
|
plan, err := publish.Build(ctx, req)
|
|
if err != nil {
|
|
if plan.PipelineID == "" {
|
|
plan.PipelineID = pipeline.ID
|
|
}
|
|
if plan.DestinationID == "" {
|
|
plan.DestinationID = destination.ID
|
|
}
|
|
if plan.BundleID == "" {
|
|
plan.BundleID = sourceBundle.Manifest.ID
|
|
}
|
|
if plan.BundlePath == "" {
|
|
plan.BundlePath = sourceBundle.RootRelativePath
|
|
}
|
|
}
|
|
if jsonOutput {
|
|
result.Actions = append(result.Actions, runActionFromPlan(destination.Backend, plan, err))
|
|
} else if options.Stdout != nil {
|
|
writePlanLine(options.Stdout, destination.Backend, plan, err)
|
|
}
|
|
if err != nil {
|
|
deferCloseDestination()
|
|
failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err)
|
|
summary.recordFailure()
|
|
continue
|
|
}
|
|
summary.recordPlan(plan.Action)
|
|
if !options.DryRun {
|
|
if err := publish.Execute(ctx, req, plan); err != nil {
|
|
deferCloseDestination()
|
|
failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err)
|
|
summary.recordFailure()
|
|
continue
|
|
}
|
|
if shouldNotify(plan.Action) {
|
|
if err := notifier.Notify(ctx, notifyEvent(plan)); err != nil {
|
|
deferCloseDestination()
|
|
failures.add(pipeline.ID, destination.ID, destination.Backend, storage.DisplayPath(sourceBundle.RootRelativePath), err)
|
|
summary.recordFailure()
|
|
continue
|
|
}
|
|
}
|
|
}
|
|
deferCloseDestination()
|
|
}
|
|
}
|
|
closeBackend(sourceBackend)
|
|
}
|
|
result.Summary = summary.Result()
|
|
if jsonOutput {
|
|
if err := WriteJSONEnvelope(options.Stdout, "run", len(failures.items) == 0, warnings, result, failures.outputErrors()); err != nil {
|
|
return err
|
|
}
|
|
} else if options.Stdout != nil {
|
|
if _, err := fmt.Fprintln(options.Stdout, summary.Line()); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if len(failures.items) > 0 {
|
|
return failures
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type closeableBackend interface {
|
|
Close() error
|
|
}
|
|
|
|
func closeBackend(backend storage.Backend) {
|
|
closeable, ok := backend.(closeableBackend)
|
|
if !ok {
|
|
return
|
|
}
|
|
_ = closeable.Close()
|
|
}
|
|
|
|
func writePlanLine(w io.Writer, backend string, plan publish.Plan, planErr error) {
|
|
if w == nil {
|
|
return
|
|
}
|
|
if planErr != nil {
|
|
destinationID := plan.DestinationID
|
|
if destinationID == "" {
|
|
destinationID = "unknown"
|
|
}
|
|
fmt.Fprintf(w, " - bundle=%s destination=%s backend=%s action=error reason=%q\n", storage.DisplayPath(plan.BundlePath), destinationID, backend, planErr.Error())
|
|
return
|
|
}
|
|
fmt.Fprintf(w, " - bundle=%s destination=%s backend=%s action=%s outputs=%s reason=%q\n", storage.DisplayPath(plan.BundlePath), plan.DestinationID, backend, plan.Action, outputSummary(plan.Outputs), plan.Reason)
|
|
}
|
|
|
|
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 {
|
|
if len(outputs) == 0 {
|
|
return "none"
|
|
}
|
|
paths := make([]string, 0, len(outputs))
|
|
for _, output := range outputs {
|
|
paths = append(paths, output.DestinationPath)
|
|
}
|
|
return strings.Join(paths, ",")
|
|
}
|
|
|
|
func destinationIDs(destinations []config.Destination) []string {
|
|
ids := make([]string, 0, len(destinations))
|
|
for _, destination := range destinations {
|
|
ids = append(ids, destination.ID)
|
|
}
|
|
return ids
|
|
}
|
|
|
|
func destinationSummary(destinations []config.Destination) string {
|
|
if len(destinations) == 0 {
|
|
return "none"
|
|
}
|
|
ids := make([]string, 0, len(destinations))
|
|
for _, destination := range destinations {
|
|
ids = append(ids, destination.ID)
|
|
}
|
|
return strings.Join(ids, ",")
|
|
}
|
|
|
|
func writeSecretConflictWarnings(w io.Writer, conflicts []config.SecretConflict) error {
|
|
return writeWarnings(w, secretConflictWarnings(conflicts))
|
|
}
|
|
|
|
func secretConflictWarnings(conflicts []config.SecretConflict) []OutputWarning {
|
|
warnings := make([]OutputWarning, 0, len(conflicts))
|
|
for _, conflict := range conflicts {
|
|
warnings = append(warnings, OutputWarning{
|
|
Message: fmt.Sprintf("secret %s ignored because the real environment already has that variable", conflict.Name),
|
|
})
|
|
}
|
|
return warnings
|
|
}
|
|
|
|
func writeSSHWarnings(w io.Writer, pipeline config.Pipeline) error {
|
|
return writeWarnings(w, sshWarnings(pipeline))
|
|
}
|
|
|
|
func sshWarnings(pipeline config.Pipeline) []OutputWarning {
|
|
var warnings []OutputWarning
|
|
if pipeline.Source.Backend == config.BackendSSH && pipeline.Source.SSH.HostKeyPolicy == config.HostKeyPolicyOff {
|
|
warnings = append(warnings, OutputWarning{
|
|
Message: fmt.Sprintf("pipeline=%s source host_key_policy=off disables SSH host key checking", pipeline.ID),
|
|
})
|
|
}
|
|
for _, destination := range pipeline.Destinations {
|
|
if destination.Backend == config.BackendSSH && destination.SSH.HostKeyPolicy == config.HostKeyPolicyOff {
|
|
warnings = append(warnings, OutputWarning{
|
|
Message: fmt.Sprintf("pipeline=%s destination=%s host_key_policy=off disables SSH host key checking", pipeline.ID, destination.ID),
|
|
})
|
|
}
|
|
}
|
|
return warnings
|
|
}
|
|
|
|
func writeWarnings(w io.Writer, warnings []OutputWarning) error {
|
|
for _, warning := range warnings {
|
|
if _, err := fmt.Fprintf(w, "Warning: %s\n", warning.Message); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func shouldNotify(action publish.Action) bool {
|
|
return action == publish.ActionPublishNew || action == publish.ActionReplaceOlder || action == publish.ActionForceReplace
|
|
}
|
|
|
|
func notifyEvent(plan publish.Plan) notify.Event {
|
|
outputs := make([]notify.Output, 0, len(plan.Outputs))
|
|
for _, output := range plan.Outputs {
|
|
outputs = append(outputs, notify.Output{
|
|
Path: output.DestinationPath,
|
|
Kind: output.Kind,
|
|
SourcePath: output.SourcePath,
|
|
Transform: output.Transform,
|
|
SHA256: output.SHA256,
|
|
Size: output.Size,
|
|
})
|
|
}
|
|
return notify.Event{
|
|
PipelineID: plan.PipelineID,
|
|
DestinationID: plan.DestinationID,
|
|
BundleID: plan.BundleID,
|
|
BundlePath: plan.BundlePath,
|
|
Action: string(plan.Action),
|
|
Outputs: outputs,
|
|
}
|
|
}
|
|
|
|
type runResult struct {
|
|
DryRun bool `json:"dry_run"`
|
|
Pipelines []runPipelineResult `json:"pipelines"`
|
|
Actions []runActionResult `json:"actions"`
|
|
Summary runSummaryResult `json:"summary"`
|
|
}
|
|
|
|
type runPipelineResult struct {
|
|
ID string `json:"id"`
|
|
SourceBackend string `json:"source_backend"`
|
|
BundleCount int `json:"bundle_count"`
|
|
Destinations []string `json:"destinations"`
|
|
}
|
|
|
|
type runActionResult struct {
|
|
PipelineID string `json:"pipeline_id,omitempty"`
|
|
DestinationID string `json:"destination_id"`
|
|
Backend string `json:"backend"`
|
|
BundleID string `json:"bundle_id,omitempty"`
|
|
BundlePath string `json:"bundle_path"`
|
|
Action string `json:"action"`
|
|
Reason string `json:"reason,omitempty"`
|
|
Outputs []runOutputResult `json:"outputs"`
|
|
}
|
|
|
|
type runOutputResult struct {
|
|
Path string `json:"path"`
|
|
Kind string `json:"kind"`
|
|
SourcePath string `json:"source_path,omitempty"`
|
|
Transform string `json:"transform,omitempty"`
|
|
SHA256 string `json:"sha256"`
|
|
Size int64 `json:"size"`
|
|
}
|
|
|
|
func runActionFromPlan(backend string, plan publish.Plan, planErr error) runActionResult {
|
|
if planErr != nil {
|
|
destinationID := plan.DestinationID
|
|
if destinationID == "" {
|
|
destinationID = "unknown"
|
|
}
|
|
return runActionResult{
|
|
PipelineID: plan.PipelineID,
|
|
DestinationID: destinationID,
|
|
Backend: backend,
|
|
BundleID: plan.BundleID,
|
|
BundlePath: storage.DisplayPath(plan.BundlePath),
|
|
Action: "error",
|
|
Reason: planErr.Error(),
|
|
Outputs: []runOutputResult{},
|
|
}
|
|
}
|
|
return runActionResult{
|
|
PipelineID: plan.PipelineID,
|
|
DestinationID: plan.DestinationID,
|
|
Backend: backend,
|
|
BundleID: plan.BundleID,
|
|
BundlePath: storage.DisplayPath(plan.BundlePath),
|
|
Action: string(plan.Action),
|
|
Reason: plan.Reason,
|
|
Outputs: runOutputsFromPlan(plan.Outputs),
|
|
}
|
|
}
|
|
|
|
func errorAction(pipelineID, destinationID, backend, bundlePath string, err error) runActionResult {
|
|
return runActionResult{
|
|
PipelineID: pipelineID,
|
|
DestinationID: destinationID,
|
|
Backend: backend,
|
|
BundlePath: storage.DisplayPath(bundlePath),
|
|
Action: "error",
|
|
Reason: err.Error(),
|
|
Outputs: []runOutputResult{},
|
|
}
|
|
}
|
|
|
|
func runOutputsFromPlan(outputs []publish.Output) []runOutputResult {
|
|
results := make([]runOutputResult, 0, len(outputs))
|
|
for _, output := range outputs {
|
|
results = append(results, runOutputResult{
|
|
Path: output.DestinationPath,
|
|
Kind: output.Kind,
|
|
SourcePath: output.SourcePath,
|
|
Transform: output.Transform,
|
|
SHA256: output.SHA256,
|
|
Size: output.Size,
|
|
})
|
|
}
|
|
return results
|
|
}
|
|
|
|
type runSummary struct {
|
|
dryRun bool
|
|
planned int
|
|
publishNew int
|
|
replaceOlder int
|
|
forceReplace int
|
|
skipped int
|
|
failures int
|
|
}
|
|
|
|
func (s *runSummary) recordPlan(action publish.Action) {
|
|
s.planned++
|
|
switch action {
|
|
case publish.ActionPublishNew:
|
|
s.publishNew++
|
|
case publish.ActionReplaceOlder:
|
|
s.replaceOlder++
|
|
case publish.ActionForceReplace:
|
|
s.forceReplace++
|
|
case publish.ActionSkipSame, publish.ActionSkipDestinationNewer:
|
|
s.skipped++
|
|
}
|
|
}
|
|
|
|
func (s *runSummary) recordFailure() {
|
|
s.failures++
|
|
}
|
|
|
|
func (s runSummary) Line() string {
|
|
status := "ok"
|
|
if s.failures > 0 {
|
|
status = "failed"
|
|
}
|
|
return fmt.Sprintf("Final status: %s planned=%d publish_new=%d replace_older=%d force_replace=%d skipped=%d failed=%d dry_run=%t", status, s.planned, s.publishNew, s.replaceOlder, s.forceReplace, s.skipped, s.failures, s.dryRun)
|
|
}
|
|
|
|
type runSummaryResult 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"`
|
|
}
|
|
|
|
func (s runSummary) Result() runSummaryResult {
|
|
status := "ok"
|
|
if s.failures > 0 {
|
|
status = "failed"
|
|
}
|
|
return runSummaryResult{
|
|
Status: status,
|
|
Planned: s.planned,
|
|
PublishNew: s.publishNew,
|
|
ReplaceOlder: s.replaceOlder,
|
|
ForceReplace: s.forceReplace,
|
|
Skipped: s.skipped,
|
|
Failed: s.failures,
|
|
DryRun: s.dryRun,
|
|
}
|
|
}
|
|
|
|
type runFailure struct {
|
|
pipelineID string
|
|
destinationID string
|
|
backend string
|
|
bundlePath string
|
|
err error
|
|
}
|
|
|
|
type runFailures struct {
|
|
items []runFailure
|
|
}
|
|
|
|
func (f *runFailures) add(pipelineID, destinationID, backend, bundlePath string, err error) {
|
|
f.items = append(f.items, runFailure{
|
|
pipelineID: pipelineID,
|
|
destinationID: destinationID,
|
|
backend: backend,
|
|
bundlePath: bundlePath,
|
|
err: err,
|
|
})
|
|
}
|
|
|
|
func (f runFailures) Error() string {
|
|
if len(f.items) == 0 {
|
|
return ""
|
|
}
|
|
parts := make([]string, 0, len(f.items))
|
|
for _, item := range f.items {
|
|
parts = append(parts, fmt.Sprintf("pipeline %s destination %s backend %s bundle %s: %v", item.pipelineID, item.destinationID, item.backend, item.bundlePath, item.err))
|
|
}
|
|
return "run failed: " + strings.Join(parts, "; ")
|
|
}
|
|
|
|
func (f runFailures) outputErrors() []OutputError {
|
|
if len(f.items) == 0 {
|
|
return nil
|
|
}
|
|
errors := make([]OutputError, 0, len(f.items))
|
|
for _, item := range f.items {
|
|
errors = append(errors, OutputError{
|
|
PipelineID: item.pipelineID,
|
|
DestinationID: item.destinationID,
|
|
Backend: item.backend,
|
|
BundlePath: item.bundlePath,
|
|
Message: item.err.Error(),
|
|
})
|
|
}
|
|
return errors
|
|
}
|
|
|
|
func IsPartialResultError(err error) bool {
|
|
var failures runFailures
|
|
return errors.As(err, &failures)
|
|
}
|
|
|
|
func (f runFailures) Unwrap() error {
|
|
errs := make([]error, 0, len(f.items))
|
|
for _, item := range f.items {
|
|
errs = append(errs, item.err)
|
|
}
|
|
return errors.Join(errs...)
|
|
}
|