feat: added a resource agent to monitor resources. Fixes #16126#16372
feat: added a resource agent to monitor resources. Fixes #16126#16372isubasinghe wants to merge 12 commits into
Conversation
Signed-off-by: isubasinghe <isitha@pipekit.io>
Signed-off-by: isubasinghe <isitha@pipekit.io>
Signed-off-by: isubasinghe <isitha@pipekit.io>
a79c36b to
0044655
Compare
Signed-off-by: isubasinghe <isitha@pipekit.io>
- UI: render ResourceMonitor nodes by adding the type to the DAG genre allow-list (the graph panel hides any node whose genre is not enabled) - agent pod: mount a writable /tmp so the executor can write manifest temp files under the read-only root filesystem - report a Running phase while the agent watches a resource, and make the taskset reconcile idempotent (skip unchanged non-terminal nodes, only stamp FinishedAt on terminal results) so a persisting Running node does not hot-loop the controller - inject the workflow ownerReference in the agent's create path (withAgentMetadata) so setOwnerReference also covers manifestFrom, not just inline manifests; drop the controller-side inline-only injection Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: isubasinghe <isitha@pipekit.io>
Examples for the agent-based resource template: a pod, workflow-of-workflows (a small one and a 30-child scale variant), and manifestFrom sourced from a native S3 artifact. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Signed-off-by: isubasinghe <isitha@pipekit.io>
Signed-off-by: isubasinghe <isitha@pipekit.io>
Signed-off-by: isubasinghe <isitha@pipekit.io>
isubasinghe
changed the title
Signed-off-by: isubasinghe <isitha@pipekit.io>
📝 WalkthroughWalkthroughAdds opt-in shared resource-agent pods for resource templates. The change updates API schemas and node types, adds controller and executor support for resource creation and monitoring, introduces RBAC and service-account wiring, and provides documentation and workflow examples. ChangesResource-agent execution
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
🧹 Nitpick comments (2)
workflow/controller/taskset.go (1)
189-191: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider initializing
common.LabelKeyCompletedon creation.
common.LabelKeyCompletedis included when patching an existing TaskSet (line 226) but is omitted during its initial creation here. Adding it to the initiallabelsmap ensures consistency and allows components relying on this label for filtering to correctly identify newly created, active TaskSets right from the start.♻️ Proposed refactor
labels := map[string]string{ + common.LabelKeyCompleted: strconv.FormatBool(woc.wf.Status.Fulfilled()), common.LabelKeyWorkflowUID: string(woc.wf.UID), }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@workflow/controller/taskset.go` around lines 189 - 191, Update the initial labels map in the TaskSet creation flow to include common.LabelKeyCompleted with the active/uncompleted value expected by the existing patch logic. Keep the workflow UID label unchanged and align the creation labels with the completed-label handling used when updating an existing TaskSet.manifests/quick-start/base/resource-agent-sa.yaml (1)
9-17: 🔒 Security & Privacy | 🔵 Trivial | ⚡ Quick winAvoid creating long-lived static ServiceAccount tokens.
Starting with Kubernetes 1.24, long-lived static ServiceAccount token secrets are no longer automatically generated and are discouraged for security reasons. Unless this static token is explicitly required by an external system or an older architectural constraint, the resource agent pod will automatically mount a short-lived token using the TokenRequest API (Bound Service Account Token Volumes) via
automountServiceAccountToken.Consider removing this
Secretto align with modern Kubernetes security best practices.♻️ Proposed refactor
---- -apiVersion: v1 -kind: Secret -metadata: - name: default-resource-agent.service-account-token - annotations: - kubernetes.io/service-account.name: default-resource-agent -type: kubernetes.io/service-account-token🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@manifests/quick-start/base/resource-agent-sa.yaml` around lines 9 - 17, Remove the default-resource-agent.service-account-token Secret manifest, including its service-account-token annotations, and rely on the resource agent pod’s short-lived TokenRequest-based token mounting through automountServiceAccountToken.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@workflow/controller/taskset.go`:
- Around line 153-167: Update the skip condition in the task-result
reconciliation loop to deep-compare node.Outputs with taskResult.Outputs and
remove the taskResult.Phase.Completed() exclusion, so unchanged running and
completed nodes are skipped. Preserve updates when Outputs, Phase, or Message
differ, and only assign FinishedAt when transitioning to a changed terminal
result so it is stamped once.
In `@workflow/executor/resource-agent.go`:
- Around line 635-643: Update the AlreadyExists recovery branch in the resource
creation flow to validate the fetched object’s workflow-UID label and node-ID
annotation against this agent’s expected values before returning it. Only accept
and return the existing object when both ownership markers match; otherwise
preserve and return the original AlreadyExists conflict.
- Around line 138-164: The create-to-watch flow in the resource execution path
can miss a deletion between createResource and ensureInformer, leaving the node
Running indefinitely. Update the logic around createResource and ensureInformer
so informer startup and cache synchronization occur before creating the
resource, or provide an equivalent gap-free GET/watch handoff; preserve the
existing success handling for deletes and empty requirements.
- Around line 711-725: Update the non-transient error branch in Agent to
propagate the fatal patch failure instead of populating and requeueing errored
results through rae.requeueResults. Preserve requeue behavior only for transient
or payload-correctable errors, and ensure the propagated error allows the
controller to terminate the monitor nodes.
- Around line 597-600: The manifest validation in the create flow must prevent
restart-unsafe generateName usage or persist and reuse the server-assigned name.
Update the logic around obj.GetGenerateName() and the create/AlreadyExists
recovery path to either reject generateName manifests with a clear bad-request
error or reliably record and reuse the assigned metadata.name across restarts;
preserve direct-name creation behavior.
- Line 765: Start the task consumer workers before the initial cache
synchronization and LIST processing so enqueue can continue draining tasks
beyond the 32-entry channel capacity. Update the setup flow around taskQueue,
WaitForCacheSync, and the worker startup blocks at the referenced locations,
preserving existing task handling while ensuring synchronization cannot be
blocked by a full queue.
- Around line 752-760: Validate taskWorkers immediately after reading it in the
executor setup, treating zero or negative values like the existing requeueTime
validation. Log the invalid value and fall back to the default worker count of
16, ensuring consumers always start while preserving valid configured counts.
- Around line 379-396: Update the informer preflight probe around the existing
lister List calls to also perform a bounded WATCH request, preserving namespace
scoping and returning a descriptive error when watch authorization or setup
fails. Ensure the watch is closed or canceled within the probe timeout, and add
coverage for a service account that permits LIST but denies WATCH, verifying the
probe rejects it.
---
Nitpick comments:
In `@manifests/quick-start/base/resource-agent-sa.yaml`:
- Around line 9-17: Remove the default-resource-agent.service-account-token
Secret manifest, including its service-account-token annotations, and rely on
the resource agent pod’s short-lived TokenRequest-based token mounting through
automountServiceAccountToken.
In `@workflow/controller/taskset.go`:
- Around line 189-191: Update the initial labels map in the TaskSet creation
flow to include common.LabelKeyCompleted with the active/uncompleted value
expected by the existing patch logic. Keep the workflow UID label unchanged and
align the creation labels with the completed-label handling used when updating
an existing TaskSet.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: fc8bfa78-eecf-4d5a-ad48-adefe8eb6674
⛔ Files ignored due to path filters (4)
api/openapi-spec/swagger.jsonis excluded by!**/api/openapi-spec/*.jsonpkg/apis/workflow/v1alpha1/generated.pb.gois excluded by!**/*.pb.go,!**/*.pb.gopkg/apis/workflow/v1alpha1/openapi_generated.gois excluded by!**/*_generated.gosdks/java/client/docs/IoArgoprojWorkflowV1alpha1ResourceTemplate.mdis excluded by!**/sdks/java/client/**
📒 Files selected for processing (49)
.cspell.json.features/pending/resource-agent.md.gitignoreapi/jsonschema/schema.jsoncmd/argo/commands/common/get.gocmd/argoexec/commands/resource_agent.gocmd/argoexec/commands/root.godocs/environment-variables.mddocs/executor_swagger.mddocs/fields.mddocs/resource-template.mdexamples/resource-agent-manifest-from.yamlexamples/resource-agent-pod.yamlexamples/resource-agent-workflow-of-workflows-scale.yamlexamples/resource-agent-workflow-of-workflows.yamlexamples/resource-agent.yamlmanifests/base/crds/full/argoproj.io_clusterworkflowtemplates.yamlmanifests/base/crds/full/argoproj.io_cronworkflows.yamlmanifests/base/crds/full/argoproj.io_workflows.yamlmanifests/base/crds/full/argoproj.io_workflowtasksets.yamlmanifests/base/crds/full/argoproj.io_workflowtemplates.yamlmanifests/base/crds/minimal/argoproj.io_workflowtasksets.yamlmanifests/quick-start-minimal.yamlmanifests/quick-start-mysql.yamlmanifests/quick-start-postgres.yamlmanifests/quick-start-telemetry.yamlmanifests/quick-start/base/kustomization.yamlmanifests/quick-start/base/resource-agent-default-rolebinding.yamlmanifests/quick-start/base/resource-agent-role.yamlmanifests/quick-start/base/resource-agent-sa.yamlpkg/apis/workflow/v1alpha1/generated.protopkg/apis/workflow/v1alpha1/workflow_types.gopkg/plugins/executor/swagger.ymlui/src/workflows/components/workflow-dag/genres.tsworkflow/common/common.goworkflow/controller/agent.goworkflow/controller/operator.goworkflow/controller/resource_template.goworkflow/controller/taskset.goworkflow/controller/taskset_test.goworkflow/executor/agent.goworkflow/executor/agent_test.goworkflow/executor/resource-agent.goworkflow/executor/resource-agent_test.goworkflow/executor/resource.goworkflow/executor/resource_test.goworkflow/progress/updater.goworkflow/util/util.goworkflow/validate/validate.go
Signed-off-by: isubasinghe <isitha@pipekit.io>
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@manifests/quick-start-minimal.yaml`:
- Around line 190911-190927: Expand the resource-agent Role permissions for
configmaps, pods, and workflows to include the patch, update, and delete verbs
alongside the existing create, get, list, and watch verbs, enabling all Resource
template actions.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 92bd515e-ef2b-46a8-891b-8911d1c59639
📒 Files selected for processing (10)
manifests/quick-start-minimal.yamlmanifests/quick-start-mysql.yamlmanifests/quick-start-postgres.yamlmanifests/quick-start-telemetry.yamlmanifests/quick-start/base/resource-agent-role.yamlworkflow/common/common.goworkflow/controller/taskset.goworkflow/controller/taskset_test.goworkflow/executor/resource-agent.goworkflow/executor/resource-agent_test.go
🚧 Files skipped from review as they are similar to previous changes (5)
- manifests/quick-start-telemetry.yaml
- workflow/common/common.go
- workflow/controller/taskset.go
- workflow/executor/resource-agent.go
- manifests/quick-start-postgres.yaml
| resources: | ||
| - configmaps | ||
| - pods | ||
| verbs: | ||
| - create | ||
| - get | ||
| - list | ||
| - watch | ||
| - apiGroups: | ||
| - argoproj.io | ||
| resources: | ||
| - workflows | ||
| verbs: | ||
| - create | ||
| - get | ||
| - list | ||
| - watch |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Add missing verbs to fully support Resource template actions.
The resource-agent Role currently grants create, get, list, and watch permissions for configmaps, pods, and workflows. While this is sufficient for action: create, it will cause Resource templates using action: apply, action: replace, action: patch, or action: delete to fail with RBAC errors.
For a quick-start environment, consider adding patch, update, and delete verbs to support all available resource actions out of the box.
🛠️ Proposed fix to add missing verbs
resources:
- configmaps
- pods
verbs:
- create
+ - delete
- get
- list
+ - patch
+ - update
- watch
- apiGroups:
- argoproj.io
resources:
- workflows
verbs:
- create
+ - delete
- get
- list
+ - patch
+ - update
- watch📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| resources: | |
| - configmaps | |
| - pods | |
| verbs: | |
| - create | |
| - get | |
| - list | |
| - watch | |
| - apiGroups: | |
| - argoproj.io | |
| resources: | |
| - workflows | |
| verbs: | |
| - create | |
| - get | |
| - list | |
| - watch | |
| resources: | |
| - configmaps | |
| - pods | |
| verbs: | |
| - create | |
| - delete | |
| - get | |
| - list | |
| - patch | |
| - update | |
| - watch | |
| - apiGroups: | |
| - argoproj.io | |
| resources: | |
| - workflows | |
| verbs: | |
| - create | |
| - delete | |
| - get | |
| - list | |
| - patch | |
| - update | |
| - watch |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@manifests/quick-start-minimal.yaml` around lines 190911 - 190927, Expand the
resource-agent Role permissions for configmaps, pods, and workflows to include
the patch, update, and delete verbs alongside the existing create, get, list,
and watch verbs, enabling all Resource template actions.
There was a problem hiding this comment.
Code review
Found 19 issues (7 high-confidence, 12 minor). Each is also posted as an inline comment on the relevant file/line. Overall this is a carefully written change — it preserves the prior taskset fixes (#16075, #15220), handles the tricky races (restart safety, create-to-watch gap, terminal-node resurrection), and the client-go usage is idiomatic. The one real bug found is #1.
|
|
||
| // A delete leaves no live object to watch, and zero requirements (empty or whitespace-only | ||
| // conditions) mean there is nothing to wait for: succeed immediately, as WaitResource does. | ||
| if tmpl.Resource.Action == "delete" || (len(rc.successReqs) == 0 && len(rc.failReqs) == 0) { |
There was a problem hiding this comment.
[bug] A whitespace-only success/failure condition (e.g. produced by parameter substitution) parses to zero label requirements, so this branch reports immediate Succeeded without ever reading the resource back.
The per-pod path was deliberately hardened against exactly this case by this same PR — WaitResource in workflow/executor/resource.go (lines 219‑223) only short-circuits when both conditions are the empty string, with a comment explaining that a whitespace-only condition “must still poll at least once … instead of being reported a spurious success”. The agent path diverges from that guard: consider checking the raw strings here (as WaitResource does) rather than the parsed requirement counts.
| // matching the agent's own single-doc guard. setOwnerReference is injected by the agent's | ||
| // create path (withAgentMetadata) so it also covers manifestFrom, whose content we never see. | ||
| if common.ManifestDocCount([]byte(tmpl.Resource.Manifest)) > 1 { | ||
| return woc.markNodeError(ctx, nodeName, fmt.Errorf("agent-based resource templates support only a single manifest document")) |
There was a problem hiding this comment.
[validation gap] The single-manifest-document constraint on resource.agent: true is only enforced here at runtime, so a workflow with a static multi-document manifest passes lint/submit and fails mid-run. For static manifests this is checkable at submit time in workflow/validate (manifests coming from parameters/artifacts can still be left to this runtime check).
Same feedback pattern as the previous agent-routed field: on #15440 a maintainer asked for “some code in workflow/validate to validate this field”.
| } | ||
|
|
||
| // runKubectl mutates global os.Args and kubectl's fatal handler; serialize across workers. | ||
| rae.kubeCTLMutex.Lock() |
There was a problem hiding this comment.
[scalability] Because in-process kubectl mutates global state (os.Args, the fatal handler), this global mutex serializes resource creation across all task workers — partly defeating the worker-pool concurrency and the PR's goal of reducing overhead for workflows that create many resources. The per-pod executor got away with global-state kubectl because each pod ran exactly one invocation; the shared agent amplifies it.
The executor already holds a DynamicClient (used for get/list/watch/informers); create could go through it, avoiding both the global-state hazard and the serialization. Reusing kubectl buys behavior parity (resource.Flags passthrough, merge strategies, server-side apply), so if the mutex stays it's worth a comment acknowledging the throughput ceiling as a conscious trade-off.
| rae.eventQueue.ShutDown() | ||
| }() | ||
|
|
||
| <-ctx.Done() |
There was a problem hiding this comment.
[reliability] The resource agent only exits on ctx.Done(). The sibling AgentExecutor (workflow/executor/agent.go, ~lines 101‑113) additionally self-terminates when the taskset is deleted (watch.Deleted) and when the workflow-completed label is set (IsWorkflowCompleted). Here the taskset informer's DeleteFunc only logs, and LabelKeyCompleted is never checked — so any path where the controller fails to delete the pod leaves it running indefinitely. Consider mirroring AgentExecutor's belt-and-suspenders self-exit.
| executeTemplate = ae.executeHTTPTemplate | ||
| case tmpl.Plugin != nil: | ||
| executeTemplate = ae.executePluginTemplate | ||
| case tmpl.Resource != nil: |
There was a problem hiding this comment.
[coupling] This skip and ResourceAgentExecutor's inverse skip (non-resource tasks) are the only things keeping the two agents from stepping on each other's nodes in the shared WorkflowTaskSet, and the relationship is enforced only by comments. Adding a third agent-served template type would require updating every agent's skip logic or nodes error spuriously. Worth a cross-referencing comment at both sites at minimum (or a shared predicate both agents consume).
| "github.com/argoproj/argo-workflows/v4/workflow/common" | ||
| ) | ||
|
|
||
| func TestManifestDocCount(t *testing.T) { |
There was a problem hiding this comment.
[test nit] TestManifestDocCount tests common.ManifestDocCount (defined in workflow/common/common.go and also used by the controller) — it belongs in workflow/common/common_test.go rather than the executor package's test file.
| // inferPluralName guesses the lowercase plural resource name for a kind. | ||
| // This is the best guess we can do here and is what `kubectl` uses under the hood. Hopefully future versions of the | ||
| // REST client would remove the need to infer the plural name. | ||
| func inferPluralName(kind string) string { |
There was a problem hiding this comment.
[nit] inferPluralName is extracted but has exactly one caller (inferObjectSelfLink, same as before) — the resource agent uses the RESTMapper for pluralization instead. The extraction adds churn without reuse; fine to keep, but it isn't needed for this PR.
| NodeTypeSkipped NodeType = "Skipped" | ||
| NodeTypeSuspend NodeType = "Suspend" | ||
| NodeTypeHTTP NodeType = "HTTP" | ||
| NodeTypeResourceMonitor NodeType = "ResourceMonitor" |
There was a problem hiding this comment.
[naming nit] The node both creates and monitors the resource; ResourceMonitor undersells the create half. Public API enum value, so worth getting right before it ships.
| // template's scope. Pod templates always do; resource templates do too (including the agent | ||
| // variant, which is not IsPodType) so that a manifest referencing {{pod.name}} keeps resolving | ||
| // the same way regardless of resource.agent. | ||
| func (tmpl *Template) ProvidesPodNameVar() bool { |
There was a problem hiding this comment.
[nit] For agent-based resource templates this keeps {{pod.name}} resolving even though no per-node pod exists, so the variable yields a pod name that never exists. The comment says this is intentional for template-compat with the non-agent path — fair, but consider documenting the semantics (users may pass {{pod.name}} to things that expect a real pod).
| WithField("requeueTime", requeueTime). | ||
| Info(ctx, "Starting Agent") | ||
|
|
||
| taskQueue := make(chan string, 32) |
There was a problem hiding this comment.
[scalability nit] taskQueue <- nodeID inside the informer event handler blocks once >32 tasks are in flight beyond the workers, stalling the WorkflowTaskSet informer delivery goroutine until workers drain (each createAndWatchResource can block ~30s on kubectl + cache sync). Backpressure rather than deadlock, and the scale example uses 30 children, but large fan-outs will feel it — a workqueue (as used elsewhere in this file) wouldn't block the handler.
Signed-off-by: isubasinghe <isitha@pipekit.io>
Signed-off-by: isubasinghe <isitha@pipekit.io>
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
workflow/executor/resource.go (1)
407-417: 🚀 Performance & Scalability | 🔵 TrivialGlobal kubectl mutex serializes all resource-agent kubectl calls.
kubectlMutexfixes a legitimate package-level flag race inkubectlcmd.NewKubectlCommand, but it also means every task worker'skubectlinvocation is now fully serialized process-wide. For the large fan-out workflow-of-workflows scenario this feature targets, this could become the throughput bottleneck for a single resource-agent pod handling many concurrent resource creations/patches.Worth keeping an eye on in scale testing; if it becomes a bottleneck, options include using
client-go's dynamic/REST client directly instead of shelling out tokubectl's command tree, or a narrower lock scoped only around command construction rather than the full execute.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@workflow/executor/resource.go` around lines 407 - 417, Evaluate the process-wide kubectlMutex in runKubectl against high-concurrency resource-agent workloads, measuring whether serializing the full command execution creates a throughput bottleneck. If scale testing confirms contention, narrow synchronization to kubectl command construction/parsing or replace the shell command path with client-go dynamic/REST operations while preserving protection against package-level flag races.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@workflow/executor/agent.go`:
- Around line 169-190: Bound the entire retry operation in
patchTaskSetStatusNodes with a context.WithTimeout derived from ctx, including
JSON patch retries and the Patch call itself, so a hung API request cannot block
the patchWorker loop beyond the intended retry window. Ensure the derived
context is canceled and passed to the retry callback and taskSetInterface.Patch
while preserving the existing transient-error backoff behavior.
---
Nitpick comments:
In `@workflow/executor/resource.go`:
- Around line 407-417: Evaluate the process-wide kubectlMutex in runKubectl
against high-concurrency resource-agent workloads, measuring whether serializing
the full command execution creates a throughput bottleneck. If scale testing
confirms contention, narrow synchronization to kubectl command
construction/parsing or replace the shell command path with client-go
dynamic/REST operations while preserving protection against package-level flag
races.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 986015b6-bb63-45ef-a913-061d60ce40d1
📒 Files selected for processing (5)
ui/src/workflows/components/workflow-dag/genres.tsworkflow/executor/agent.goworkflow/executor/resource-agent.goworkflow/executor/resource.goworkflow/executor/resource_test.go
🚧 Files skipped from review as they are similar to previous changes (1)
- workflow/executor/resource-agent.go
| // patchTaskSetStatusNodes merge-patches the given node results into the named WorkflowTaskSet's | ||
| // status subresource, retrying transient errors with backoff. Both agents' patch workers flush | ||
| // through this so the patch mechanics and retry policy cannot drift between them. | ||
| func patchTaskSetStatusNodes(ctx context.Context, taskSetInterface v1alpha1.WorkflowTaskSetInterface, workflowName string, nodes map[string]wfv1.NodeResult) error { | ||
| patch, err := json.Marshal(map[string]any{"status": wfv1.WorkflowTaskSetStatus{Nodes: nodes}}) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| return retry.OnError(wait.Backoff{ | ||
| Duration: time.Second, | ||
| Factor: 2, | ||
| Jitter: 0.1, | ||
| Steps: 5, | ||
| Cap: 30 * time.Second, | ||
| }, func(retryErr error) bool { | ||
| return errors.IsTransientErr(ctx, retryErr) | ||
| }, func() error { | ||
| _, patchErr := taskSetInterface.Patch(ctx, workflowName, types.MergePatchType, patch, metav1.PatchOptions{}, "status") | ||
| return patchErr | ||
| }) | ||
| } | ||
|
|
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Unbounded patch retry blocks the single patchWorker loop.
patchTaskSetStatusNodes retries up to 5 times (backoff to 30s cap, ~31s total) synchronously inside patchWorker's ticker branch, with no per-call context deadline. While blocked, the select loop can't drain responseQueue or fire subsequent ticks, so a degraded/slow API server stalls result propagation for all nodes served by this agent for the duration of the retry (and longer if a single Patch call itself hangs without a deadline).
Consider bounding the whole patch attempt with a context.WithTimeout derived from ctx, so a stuck API call can't extend beyond the intended backoff window.
🔧 Proposed fix
func patchTaskSetStatusNodes(ctx context.Context, taskSetInterface v1alpha1.WorkflowTaskSetInterface, workflowName string, nodes map[string]wfv1.NodeResult) error {
patch, err := json.Marshal(map[string]any{"status": wfv1.WorkflowTaskSetStatus{Nodes: nodes}})
if err != nil {
return err
}
+ ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
+ defer cancel()
return retry.OnError(wait.Backoff{
Duration: time.Second,
Factor: 2,
Jitter: 0.1,
Steps: 5,
Cap: 30 * time.Second,
}, func(retryErr error) bool {
return errors.IsTransientErr(ctx, retryErr)
}, func() error {
_, patchErr := taskSetInterface.Patch(ctx, workflowName, types.MergePatchType, patch, metav1.PatchOptions{}, "status")
return patchErr
})
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| // patchTaskSetStatusNodes merge-patches the given node results into the named WorkflowTaskSet's | |
| // status subresource, retrying transient errors with backoff. Both agents' patch workers flush | |
| // through this so the patch mechanics and retry policy cannot drift between them. | |
| func patchTaskSetStatusNodes(ctx context.Context, taskSetInterface v1alpha1.WorkflowTaskSetInterface, workflowName string, nodes map[string]wfv1.NodeResult) error { | |
| patch, err := json.Marshal(map[string]any{"status": wfv1.WorkflowTaskSetStatus{Nodes: nodes}}) | |
| if err != nil { | |
| return err | |
| } | |
| return retry.OnError(wait.Backoff{ | |
| Duration: time.Second, | |
| Factor: 2, | |
| Jitter: 0.1, | |
| Steps: 5, | |
| Cap: 30 * time.Second, | |
| }, func(retryErr error) bool { | |
| return errors.IsTransientErr(ctx, retryErr) | |
| }, func() error { | |
| _, patchErr := taskSetInterface.Patch(ctx, workflowName, types.MergePatchType, patch, metav1.PatchOptions{}, "status") | |
| return patchErr | |
| }) | |
| } | |
| // patchTaskSetStatusNodes merge-patches the given node results into the named WorkflowTaskSet's | |
| // status subresource, retrying transient errors with backoff. Both agents' patch workers flush | |
| // through this so the patch mechanics and retry policy cannot drift between them. | |
| func patchTaskSetStatusNodes(ctx context.Context, taskSetInterface v1alpha1.WorkflowTaskSetInterface, workflowName string, nodes map[string]wfv1.NodeResult) error { | |
| patch, err := json.Marshal(map[string]any{"status": wfv1.WorkflowTaskSetStatus{Nodes: nodes}}) | |
| if err != nil { | |
| return err | |
| } | |
| ctx, cancel := context.WithTimeout(ctx, 30*time.Second) | |
| defer cancel() | |
| return retry.OnError(wait.Backoff{ | |
| Duration: time.Second, | |
| Factor: 2, | |
| Jitter: 0.1, | |
| Steps: 5, | |
| Cap: 30 * time.Second, | |
| }, func(retryErr error) bool { | |
| return errors.IsTransientErr(ctx, retryErr) | |
| }, func() error { | |
| _, patchErr := taskSetInterface.Patch(ctx, workflowName, types.MergePatchType, patch, metav1.PatchOptions{}, "status") | |
| return patchErr | |
| }) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@workflow/executor/agent.go` around lines 169 - 190, Bound the entire retry
operation in patchTaskSetStatusNodes with a context.WithTimeout derived from
ctx, including JSON patch retries and the Patch call itself, so a hung API
request cannot block the patchWorker loop beyond the intended retry window.
Ensure the derived context is canceled and passed to the retry callback and
taskSetInterface.Patch while preserving the existing transient-error backoff
behavior.
Joibel
left a comment
Joibel
left a comment
There was a problem hiding this comment.
This needs (probably more than one) e2e testing
It feels like the resource agent should have all artifact plugins installed in it, then they'd all just work.
| if resourceAgent { | ||
| // The resource agent's informers need list+watch on whole GVRs — broader than | ||
| // workflow pods should carry — so it runs under a dedicated service account, | ||
| // following the `<name>-executor-plugin` convention. Fails if the SA is missing. |
There was a problem hiding this comment.
Comment says -executor-plugin. Code says "-resource-agent". Nothing anywhere seems to tell the user that their actual SA they requested is being ignored, validation doesn't validate they haven't specified it, and we'd probably want a runtime "nope" as well.
| agent always needs: `create`, `get`, `list` and `watch` at minimum (the agent creates the | ||
| resource, reads it back for restart-safety, and watches/polls it for its success and failure | ||
| conditions), plus `update` for `apply`/`replace`, `patch` for `patch`, and `delete` for `delete`, | ||
| * `get` on `secrets` and `configmaps` if you use `manifestFrom` artifacts, since the agent |
| func runKubectl(ctx context.Context, args ...string) (out []byte, err error) { | ||
| logging.RequireLoggerFromContext(ctx).Info(ctx, strings.Join(args, " ")) | ||
| installKubectlFatalHandler() | ||
| kubectlMutex.Lock() |
There was a problem hiding this comment.
Doesn't this lock pretty much prevent multiple workers from being useful?
| NodeTypeSkipped NodeType = "Skipped" | ||
| NodeTypeSuspend NodeType = "Suspend" | ||
| NodeTypeHTTP NodeType = "HTTP" | ||
| NodeTypeResourceAgent NodeType = "ResourceAgent" |
There was a problem hiding this comment.
Why do we actually need a new node type - it feels like it should be hidden?
| // Agent, when true, runs this resource template on the shared per-workflow resource agent pod | ||
| // (which creates the resource and watches it for its success/failure conditions) instead of a | ||
| // dedicated per-node pod. See docs/resource-template.md. | ||
| Agent bool `json:"agent,omitempty" protobuf:"varint,9,opt,name=agent"` |
There was a problem hiding this comment.
I don't like the name "agent" here, nor the boolean rather than an extendable enum.
| }) | ||
| // Execute parses os.Args unless told otherwise (kubectl's own tests set args the same way). | ||
| cmd.SetArgs(args[1:]) | ||
| if err := cmd.Execute(); err != nil { |
There was a problem hiding this comment.
I'm surprised the linter didn't complain. Prefer ExecuteContext(ctx) here.
| return []byte(tmpl.Resource.Manifest), nil | ||
| } | ||
| name := tmpl.Resource.ManifestFrom.Artifact.Name | ||
| art := tmpl.Inputs.GetArtifactByName(name) |
There was a problem hiding this comment.
Can we DRY and reuse the standard executor artifact code here - this looks like a duplicate of that code.
| # delve debug binaries | ||
| cmd/**/debug | ||
| hack/**/debug | ||
| # binaries built in-place (e.g. `go build ./cmd/argoexec` from its own dir) |
2 participants
Fixes #16126
Motivation
This adds the ability for resources to be monitored by an agent style singular pods
Modifications
Verification
Documentation
Added documentation
AI
Initial design was hand written, but was then wired up with claude.
commit 4005f85 onwards was AI driven.
Summary by CodeRabbit
agent: truefor resource templates to run on a shared per-workflow resource agent pod (including informer-based execution and nativemanifestFromsupport).resource-templateguidance, environment variables, and added RBAC/setup details plus multiple resource-agent examples.agentfield.