diff --git a/pkg/executors/push.go b/pkg/executors/push.go index 7b2d96c..63cbfdd 100644 --- a/pkg/executors/push.go +++ b/pkg/executors/push.go @@ -5,43 +5,75 @@ import ( "errors" "fmt" "strings" + "sync" "time" "github.com/google/go-containerregistry/pkg/authn" "github.com/google/go-containerregistry/pkg/crane" "github.com/google/go-containerregistry/pkg/name" + v1 "github.com/google/go-containerregistry/pkg/v1" "github.com/lyzrai/flow/pkg/engine" "github.com/lyzrai/flow/pkg/models" ) -// PushExecutor copies an OCI image from one registry to another. Source -// defaults to the local registry image produced by an upstream Build node -// (read from inputs[0][0].__build.image) but can be overridden by the -// srcImage parameter. Destination is constructed from targetRegistry + -// targetImage + tag. Optional username/password authenticate the push. +// PushExecutor copies an OCI image from one registry to one or more +// destinations in parallel. Each target is an independent network op with +// its own registry/image/tag/auth — so mirroring to N clouds takes +// max(targetN), not sum(targetN). // -// Implementation uses go-containerregistry's crane.Copy — same primitive -// the `crane` CLI / ko / skopeo-Go-callers use. No docker daemon required; -// works against any OCI v2 registry. The local registry (registry:5000) is -// reached via plain HTTP because we mark it insecure when constructing the -// reference. +// Source defaults to the upstream Build node's `__build.image` (read off +// inputs[0][0].__build.image) and can be overridden by srcImage. The local +// registry is HTTP, so srcInsecure defaults to true. // -// Parameters: -// - srcImage string override source image ref (optional) -// - targetRegistry string e.g. "ghcr.io" -// - targetImage string e.g. "org/agent" -// - tag string defaults to upstream __build.commit then "latest" -// - username string optional -// - password string optional -// - srcInsecure bool treat src registry as plain HTTP (default true; the local one is) -// - dstInsecure bool treat dst registry as plain HTTP (default false) +// Targets shape (parameters.targets is a JSON array): +// +// [ +// { +// "name": "ghcr", +// "registry": "ghcr.io", +// "image": "org/agent", +// "tag": "v1.2.3", # optional; defaults to upstream commit SHA, then "latest" +// "username": "owner", +// "password": "ghp_…", +// "insecure": false +// }, +// { "name": "mirror", "registry": "registry.local:5000", "image": "org/agent", "insecure": true } +// ] +// +// Backward compat: if `targets` is empty/missing, the legacy single-target +// fields (targetRegistry / targetImage / tag / username / password / +// dstInsecure) are read. +// +// Implementation uses go-containerregistry — same primitive `crane` CLI / +// ko / skopeo-Go-callers use. No docker daemon required. type PushExecutor struct{} +// pushTarget is the parsed shape of one entry in parameters.targets[]. +type pushTarget struct { + Name string + Registry string + Image string + Tag string + Username string + Password string + Insecure bool +} + +// pushCopy is what we record per target for the run's output items. +type pushCopy struct { + Name string `json:"name,omitempty"` + Registry string `json:"registry"` + ImageRef string `json:"imageRef"` + Digest string `json:"digest,omitempty"` + DurationMS int64 `json:"durationMs"` + Error string `json:"error,omitempty"` +} + func (e *PushExecutor) Execute(ctx context.Context, node models.NodeDef, inputs [][]models.Item, _ *engine.ExecutionContext) (map[int][]models.Item, error) { logger := engine.NodeLoggerFromContext(ctx) - // Source: explicit param wins, otherwise upstream __build.image. + // --- source --- srcImage := strParam(node.Parameters, "srcImage", "") if srcImage == "" { srcImage = imageFromBuildOutput(inputs) @@ -49,106 +81,256 @@ func (e *PushExecutor) Execute(ctx context.Context, node models.NodeDef, inputs if srcImage == "" { return nil, errors.New("push: no source image (set srcImage or wire a Build node upstream)") } - - // Destination. - dstRegistry := strings.TrimRight(strings.TrimSpace(strParam(node.Parameters, "targetRegistry", "")), "/") - dstImage := strings.TrimSpace(strParam(node.Parameters, "targetImage", "")) - if dstRegistry == "" || dstImage == "" { - return nil, errors.New("push: targetRegistry and targetImage are required") - } - tag := strings.TrimSpace(strParam(node.Parameters, "tag", "")) - if tag == "" { - tag = strFirst(commitFromBuildOutput(inputs), "latest") - } - dstRef := fmt.Sprintf("%s/%s:%s", dstRegistry, dstImage, tag) - - username := strParam(node.Parameters, "username", "") - password := strParam(node.Parameters, "password", "") - - srcInsecure := boolParam(node.Parameters, "srcInsecure", true) - dstInsecure := boolParam(node.Parameters, "dstInsecure", false) - - logger.Log(fmt.Sprintf("copying %s -> %s", srcImage, dstRef)) - - // Build the crane options. We layer: - // - context (for cancellation) - // - per-side insecure flag (lets us read from local registry:5000 over HTTP) - // - keychain for auth (only used by the destination side; explicit - // username/password takes precedence via WithAuth) - srcOpts := []crane.Option{crane.WithContext(ctx)} - dstOpts := []crane.Option{crane.WithContext(ctx)} - if srcInsecure { - srcOpts = append(srcOpts, crane.Insecure) - } - if dstInsecure { - dstOpts = append(dstOpts, crane.Insecure) - } - if username != "" || password != "" { - dstOpts = append(dstOpts, crane.WithAuth(&authn.Basic{ - Username: username, - Password: password, - })) - } else { - // Falls back to docker config / GHCR anonymous / etc. - dstOpts = append(dstOpts, crane.WithAuthFromKeychain(authn.DefaultKeychain)) - } - - // Validate refs early so we surface a clear error. if _, err := name.ParseReference(srcImage); err != nil { return nil, fmt.Errorf("push: invalid src %q: %w", srcImage, err) } - if _, err := name.ParseReference(dstRef); err != nil { - return nil, fmt.Errorf("push: invalid dst %q: %w", dstRef, err) + srcInsecure := boolParam(node.Parameters, "srcInsecure", true) + + // --- targets --- + targets := parseTargets(node.Parameters, inputs) + if len(targets) == 0 { + return nil, errors.New("push: no targets configured (set targets[] or targetRegistry/targetImage)") } - // crane.Copy doesn't accept per-side options — we emulate by pulling - // the image with src options and pushing it with dst options. + logger.Log(fmt.Sprintf("[push] source %s, %d target(s)", srcImage, len(targets))) + + // --- pull source once --- + srcOpts := []crane.Option{crane.WithContext(ctx)} + if srcInsecure { + srcOpts = append(srcOpts, crane.Insecure) + } logger.Log("pulling source manifest…") img, err := crane.Pull(srcImage, srcOpts...) if err != nil { return nil, fmt.Errorf("push: pull %s: %w", srcImage, err) } - logger.Log("pushing to destination…") - started := time.Now() - if err := crane.Push(img, dstRef, dstOpts...); err != nil { - return nil, fmt.Errorf("push: push %s: %w", dstRef, err) + // --- fan out pushes --- + results := make([]pushCopy, len(targets)) + var wg sync.WaitGroup + wg.Add(len(targets)) + for i, t := range targets { + go func(i int, t pushTarget) { + defer wg.Done() + results[i] = pushOne(ctx, img, t, logger) + }(i, t) } - logger.Log(fmt.Sprintf("done in %s", time.Since(started).Round(100*time.Millisecond))) + wg.Wait() - digest, derr := img.Digest() - digestStr := "" - if derr == nil { - digestStr = digest.String() + // --- summarize --- + succeeded := 0 + for _, r := range results { + if r.Error == "" { + succeeded++ + } + } + logger.Log(fmt.Sprintf("[push] done: %d/%d target(s) succeeded", succeeded, len(results))) + + // Convert to JSON-friendly maps for the output items. + copiesAny := make([]map[string]any, 0, len(results)) + for _, r := range results { + copiesAny = append(copiesAny, map[string]any{ + "name": r.Name, + "registry": r.Registry, + "imageRef": r.ImageRef, + "digest": r.Digest, + "durationMs": r.DurationMS, + "error": r.Error, + }) + } + summary := map[string]any{ + "src": srcImage, + "copies": copiesAny, + "finished_at": time.Now().UTC(), + } + // Convenience top-level fields when there's only one target — keeps + // the old `__push.dst / digest` shape working for downstream tooling. + if len(results) == 1 { + summary["dst"] = results[0].ImageRef + summary["digest"] = results[0].Digest } - // Pass-through items + a __push summary so downstream nodes can read it. out := make([]models.Item, 0) for _, in := range inputs { for _, item := range in { ci := copyItem(item) - ci["__push"] = map[string]any{ - "src": srcImage, - "dst": dstRef, - "digest": digestStr, - "finished_at": time.Now().UTC(), - } + ci["__push"] = summary out = append(out, ci) } } if len(out) == 0 { - out = append(out, models.Item{ - "__push": map[string]any{ - "src": srcImage, - "dst": dstRef, - "digest": digestStr, - "finished_at": time.Now().UTC(), - }, - }) + out = append(out, models.Item{"__push": summary}) + } + + if succeeded != len(results) { + // Surface the first error so the run shows a meaningful failure + // message; the rest are still in __push.copies. + for _, r := range results { + if r.Error != "" { + return nil, fmt.Errorf("push to %s failed: %s", r.ImageRef, r.Error) + } + } } return map[int][]models.Item{0: out}, nil } +// pushOne pushes a previously-pulled image to one target. +func pushOne(ctx context.Context, img v1.Image, t pushTarget, logger engine.NodeLogger) pushCopy { + started := time.Now() + dst := fmt.Sprintf("%s/%s:%s", strings.TrimRight(t.Registry, "/"), t.Image, t.Tag) + tag := t.tagPrefix() + + logger.Log(fmt.Sprintf("%s copying → %s", tag, dst)) + + if _, err := name.ParseReference(dst); err != nil { + logger.Log(fmt.Sprintf("%s ✗ invalid ref: %v", tag, err)) + return pushCopy{ + Name: t.Name, Registry: t.Registry, ImageRef: dst, + DurationMS: time.Since(started).Milliseconds(), + Error: fmt.Sprintf("invalid ref: %v", err), + } + } + + dstOpts := []crane.Option{crane.WithContext(ctx)} + if t.Insecure { + dstOpts = append(dstOpts, crane.Insecure) + } + if t.Username != "" || t.Password != "" { + dstOpts = append(dstOpts, crane.WithAuth(&authn.Basic{ + Username: t.Username, + Password: t.Password, + })) + } else { + dstOpts = append(dstOpts, crane.WithAuthFromKeychain(authn.DefaultKeychain)) + } + + if err := crane.Push(img, dst, dstOpts...); err != nil { + logger.Log(fmt.Sprintf("%s ✗ %v", tag, err)) + return pushCopy{ + Name: t.Name, Registry: t.Registry, ImageRef: dst, + DurationMS: time.Since(started).Milliseconds(), + Error: err.Error(), + } + } + digestStr := "" + if d, derr := img.Digest(); derr == nil { + digestStr = d.String() + } + dur := time.Since(started) + logger.Log(fmt.Sprintf("%s ✓ %s (%s)", tag, dst, dur.Round(100*time.Millisecond))) + return pushCopy{ + Name: t.Name, Registry: t.Registry, ImageRef: dst, + Digest: digestStr, DurationMS: dur.Milliseconds(), + } +} + +// tagPrefix is the per-target log line prefix. Mirrors the Node version's +// "[push:name]" so a multi-target log is greppable. +func (t pushTarget) tagPrefix() string { + if t.Name != "" { + return "[push:" + t.Name + "]" + } + return "[push:" + t.Registry + "]" +} + +// parseTargets unifies the new `targets[]` shape with the old single-target +// fields. Returns nothing if neither is set (caller treats as error). +// +// Tag default order: target.tag → upstream __build.commit → "latest". +func parseTargets(p map[string]any, inputs [][]models.Item) []pushTarget { + commit := commitFromBuildOutput(inputs) + + defaultTag := func(explicit string) string { + t := strings.TrimSpace(explicit) + if t != "" { + return t + } + if commit != "" { + return commit + } + return "latest" + } + + if raw, ok := p["targets"]; ok { + if arr, ok := raw.([]any); ok && len(arr) > 0 { + out := make([]pushTarget, 0, len(arr)) + for i, e := range arr { + m, ok := e.(map[string]any) + if !ok { + continue + } + reg := strings.TrimSpace(strFromAny(m["registry"])) + img := strings.TrimSpace(strFromAny(m["image"])) + if reg == "" || img == "" { + continue + } + out = append(out, pushTarget{ + Name: defaultName(strFromAny(m["name"]), reg, i), + Registry: reg, + Image: img, + Tag: defaultTag(strFromAny(m["tag"])), + Username: strFromAny(m["username"]), + Password: strFromAny(m["password"]), + Insecure: anyToBool(m["insecure"], false), + }) + } + if len(out) > 0 { + return out + } + } + } + + // Legacy single-target fields. + reg := strings.TrimSpace(strParam(p, "targetRegistry", "")) + img := strings.TrimSpace(strParam(p, "targetImage", "")) + if reg == "" || img == "" { + return nil + } + return []pushTarget{{ + Name: defaultName("", reg, 0), + Registry: reg, + Image: img, + Tag: defaultTag(strParam(p, "tag", "")), + Username: strParam(p, "username", ""), + Password: strParam(p, "password", ""), + Insecure: boolParam(p, "dstInsecure", false), + }} +} + +func defaultName(explicit, registry string, idx int) string { + s := strings.TrimSpace(explicit) + if s != "" { + return s + } + // First label of the registry hostname is usually a clear ID. + host := registry + if i := strings.Index(host, "."); i > 0 { + host = host[:i] + } + if host == "" { + return fmt.Sprintf("target-%d", idx+1) + } + return host +} + +func anyToBool(v any, def bool) bool { + switch x := v.(type) { + case bool: + return x + case string: + s := strings.ToLower(strings.TrimSpace(x)) + if s == "true" || s == "1" || s == "yes" { + return true + } + if s == "false" || s == "0" || s == "no" { + return false + } + case float64: + return x != 0 + } + return def +} + // imageFromBuildOutput walks input items for an upstream Build node's // `__build.image` field. Returns "" if not present. func imageFromBuildOutput(inputs [][]models.Item) string { @@ -185,19 +367,6 @@ func boolParam(p map[string]any, key string, def bool) bool { if p == nil { return def } - switch v := p[key].(type) { - case bool: - return v - case string: - s := strings.ToLower(strings.TrimSpace(v)) - if s == "true" || s == "1" || s == "yes" { - return true - } - if s == "false" || s == "0" || s == "no" { - return false - } - case float64: - return v != 0 - } - return def + return anyToBool(p[key], def) } + diff --git a/web/components/canvas/node-form.tsx b/web/components/canvas/node-form.tsx index 92cff32..36aace9 100644 --- a/web/components/canvas/node-form.tsx +++ b/web/components/canvas/node-form.tsx @@ -483,15 +483,38 @@ function ApprovalForm({ node, onChange }: NodeFormProps) { ); } +type PushTarget = { + name?: string; + registry?: string; + image?: string; + tag?: string; + username?: string; + password?: string; + insecure?: boolean; +}; + function PushForm({ node, onChange }: NodeFormProps) { const srcImage = getString(node, "srcImage", ""); - const targetRegistry = getString(node, "targetRegistry", "ghcr.io"); - const targetImage = getString(node, "targetImage", ""); - const tag = getString(node, "tag", ""); - const username = getString(node, "username", ""); - const password = getString(node, "password", ""); const srcInsecure = getBool(node, "srcInsecure", true); - const dstInsecure = getBool(node, "dstInsecure", false); + const rawTargets = (node.parameters?.targets as unknown) as PushTarget[] | undefined; + const targets: PushTarget[] = Array.isArray(rawTargets) ? rawTargets : []; + + function setTargets(next: PushTarget[]) { + onChange(setParam(node, "targets", next as unknown as Record[])); + } + function patch(idx: number, patch: Partial) { + setTargets(targets.map((t, i) => (i === idx ? { ...t, ...patch } : t))); + } + function addTarget() { + setTargets([ + ...targets, + { name: `target-${targets.length + 1}`, registry: "", image: "" }, + ]); + } + function removeTarget(idx: number) { + setTargets(targets.filter((_, i) => i !== idx)); + } + return (
@@ -500,7 +523,7 @@ function PushForm({ node, onChange }: NodeFormProps) { id="push-src" value={srcImage} onChange={(e) => onChange(setParam(node, "srcImage", e.target.value))} - placeholder="registry:5000/owner/agent:sha (defaults to upstream Build output)" + placeholder="registry:5000/owner/agent:sha (defaults to upstream Build)" className="font-mono text-xs" />

@@ -509,115 +532,179 @@ function PushForm({ node, onChange }: NodeFormProps) {

-
-
- - - onChange(setParam(node, "targetRegistry", e.target.value)) - } - placeholder="ghcr.io" - className="font-mono text-xs" - /> -
-
- - onChange(setParam(node, "tag", e.target.value))} - placeholder="defaults to commit SHA, then 'latest'" - className="font-mono text-xs" - /> -
-
- -
- - onChange(setParam(node, "targetImage", e.target.value))} - placeholder="org/agent" - className="font-mono text-xs" +
+ onChange(setParam(node, "srcInsecure", e.target.checked))} + className="size-3.5" /> -

- Final ref: {targetRegistry || ""}/{targetImage || ""}:{tag || ""} -

+
-
-
- - onChange(setParam(node, "username", e.target.value))} - placeholder="(empty = anonymous / docker config)" - className="font-mono text-xs" - autoComplete="off" - /> +
+
+ +
-
- - onChange(setParam(node, "password", e.target.value))} - placeholder="ghp_… or registry password" - className="font-mono text-xs" - autoComplete="new-password" - /> -
-
-
-
- Insecure transport -
-
- - onChange(setParam(node, "srcInsecure", e.target.checked)) - } - className="size-3.5" + {targets.length === 0 && ( +

+ No targets configured — Push will fail at run time. Click “+ add + target”. +

+ )} + + {targets.map((t, idx) => ( + patch(idx, p)} + onRemove={() => removeTarget(idx)} /> - -
-
- - onChange(setParam(node, "dstInsecure", e.target.checked)) - } - className="size-3.5" - /> - -
+ ))}

- Image is pulled from the source over OCI v2 and pushed to the target; - no docker daemon needed. For GHCR, the password is a PAT with{" "} - write:packages. + Each target runs in parallel. Image is pulled from the source once and + pushed concurrently — no docker daemon needed. For GHCR, password is a + PAT with write:packages.

); } +function TargetRow({ + idx, + target, + onPatch, + onRemove, +}: { + idx: number; + target: PushTarget; + onPatch: (p: Partial) => void; + onRemove: () => void; +}) { + const ref = `${target.registry || ""}/${target.image || ""}:${target.tag || ""}`; + return ( +
+
+ onPatch({ name: e.target.value })} + placeholder={`target-${idx + 1}`} + className="h-7 max-w-[140px] font-mono text-[11px]" + /> + + {ref} + + +
+ +
+
+ + onPatch({ registry: e.target.value })} + placeholder="ghcr.io" + className="h-7 font-mono text-[11px]" + /> +
+
+ + onPatch({ tag: e.target.value })} + placeholder="(commit sha → latest)" + className="h-7 font-mono text-[11px]" + /> +
+
+ +
+ + onPatch({ image: e.target.value })} + placeholder="org/agent" + className="h-7 font-mono text-[11px]" + /> +
+ +
+
+ + onPatch({ username: e.target.value })} + placeholder="(anonymous)" + autoComplete="off" + className="h-7 font-mono text-[11px]" + /> +
+
+ + onPatch({ password: e.target.value })} + placeholder="ghp_…" + autoComplete="new-password" + className="h-7 font-mono text-[11px]" + /> +
+
+ +
+ onPatch({ insecure: e.target.checked })} + className="size-3.5" + /> + +
+
+ ); +} + function DeployForm({ node, onChange }: NodeFormProps) { const runtime = getString(node, "runtime", "kubernetes"); const env = getString(node, "env", "dev"); diff --git a/web/lib/node-catalog.ts b/web/lib/node-catalog.ts index 5e9efa7..595776e 100644 --- a/web/lib/node-catalog.ts +++ b/web/lib/node-catalog.ts @@ -137,19 +137,24 @@ export const CATALOG: CatalogEntry[] = [ type: "flow-nodes-base.push", label: "Push", description: - "Mirror the locally-built image to an external registry (GHCR, Docker Hub, ECR, etc.).", + "Mirror the locally-built image to one or more external registries (GHCR, Docker Hub, ECR, etc.) in parallel.", icon: UploadCloud, color: "bg-cyan-500", outputs: 1, defaults: { - // srcImage left empty — defaults to upstream __build.image - targetRegistry: "ghcr.io", - targetImage: "", - tag: "", - username: "", - password: "", + // srcImage empty → defaults to upstream __build.image srcInsecure: true, - dstInsecure: false, + targets: [ + { + name: "ghcr", + registry: "ghcr.io", + image: "", + tag: "", + username: "", + password: "", + insecure: false, + }, + ], }, group: "deploy", },