Repository navigation
Implement TaskSpawner CRD - #22
Conversation
There was a problem hiding this comment.
Pull request overview
This PR introduces the TaskSpawner CRD, which enables automated Task creation by polling external sources like GitHub Issues. The feature adds a new controller that manages dedicated spawner Deployments for each TaskSpawner resource, implementing a clean separation between the main controller and the spawner logic.
Changes:
- Implements TaskSpawner CRD with support for GitHub Issues as an external source
- Adds a main controller (TaskSpawnerReconciler) that creates and manages spawner Deployments with finalizer-based lifecycle management
- Creates a new spawner binary that continuously polls sources and creates Task resources
- Extends CLI with "axon get taskspawner/taskspawners/ts" commands and adds printer functions
- Updates RBAC permissions to allow Deployment and Secret management
Reviewed changes
Copilot reviewed 21 out of 21 changed files in this pull request and generated 10 comments.
Show a summary per file
| File | Description |
|---|---|
| api/v1alpha1/taskspawner_types.go | Defines TaskSpawner CRD types including When, GitHubIssues, TaskTemplate, and status fields |
| internal/controller/taskspawner_controller.go | Main reconciler managing TaskSpawner lifecycle, Deployment creation, and finalizer handling |
| internal/controller/taskspawner_deployment_builder.go | Builds spawner Deployments with GitHub token mounting and proper labels |
| internal/source/source.go | Defines Source interface and WorkItem abstraction for external sources |
| internal/source/github.go | Implements GitHub source with pagination, auth, filtering, and comment fetching |
| internal/source/prompt.go | Template rendering for converting WorkItems to Task prompts |
| cmd/spawner/main.go | Spawner binary that polls sources and creates Tasks in a continuous loop |
| cmd/spawner/Dockerfile | Multi-stage Dockerfile for spawner binary |
| test/integration/taskspawner_test.go | Integration tests covering creation, deletion, token mounting, and idempotency |
| test/e2e/taskspawner_test.go | E2E tests for end-to-end TaskSpawner workflow and CLI commands |
| internal/cli/get.go | Adds taskspawner subcommand with aliases |
| internal/cli/printer.go | Table and detail view printers for TaskSpawners |
| install.yaml | RBAC updates for Deployments and Secrets |
| install-crd.yaml | TaskSpawner CRD definition |
| Makefile | Spawner image build target |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| // Re-fetch before updating status to avoid conflicts | ||
| if err := cl.Get(ctx, key, &ts); err != nil { | ||
| return fmt.Errorf("re-fetching TaskSpawner: %w", err) | ||
| } | ||
|
|
||
| ts.Status.DiscoveredItems = append(ts.Status.DiscoveredItems, axonv1alpha1.WorkItemReference{ | ||
| ID: item.ID, | ||
| TaskName: taskName, | ||
| }) | ||
| ts.Status.TotalDiscovered = len(ts.Status.DiscoveredItems) | ||
| ts.Status.TotalTasksCreated++ | ||
|
|
||
| if err := cl.Status().Update(ctx, &ts); err != nil { | ||
| log.Error(err, "updating TaskSpawner status", "item", item.ID) | ||
| } |
There was a problem hiding this comment.
Updating the status within the loop for each new item creates multiple API calls and potential race conditions. If multiple items are discovered, each status update refetches the TaskSpawner and updates incrementally, which is inefficient. Consider batching the status updates: collect all the WorkItemReferences first, then perform a single status update after all tasks have been created.
| comments, err := s.fetchComments(ctx, issue.Number) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("fetching comments for issue #%d: %w", issue.Number, err) | ||
| } |
There was a problem hiding this comment.
Fetching comments for every issue synchronously can significantly slow down discovery, especially with many issues. Each issue triggers a separate API call, and if there are 100 issues, this results in 101+ API calls (1 for issues + 100 for comments). Consider making comment fetching optional, concurrent, or implementing a batch fetching strategy to improve performance.
| // Small wait for reconcile to process | ||
| time.Sleep(500 * time.Millisecond) | ||
|
|
||
| By("Verifying still only 1 Deployment exists") | ||
| Expect(k8sClient.List(ctx, deployList, | ||
| client.InNamespace(ns.Name), | ||
| client.MatchingLabels{"axon.io/taskspawner": ts.Name}, | ||
| )).Should(Succeed()) | ||
| Expect(deployList.Items).To(HaveLen(1)) |
There was a problem hiding this comment.
Using a fixed sleep time for reconciliation (500ms) creates a flaky test. If the reconciliation takes longer than 500ms due to system load or other factors, the test may pass even if idempotency is broken. Consider replacing this with an Eventually assertion that polls for a condition, or increase the timeout significantly if a sleep is necessary.
| // Small wait for reconcile to process | |
| time.Sleep(500 * time.Millisecond) | |
| By("Verifying still only 1 Deployment exists") | |
| Expect(k8sClient.List(ctx, deployList, | |
| client.InNamespace(ns.Name), | |
| client.MatchingLabels{"axon.io/taskspawner": ts.Name}, | |
| )).Should(Succeed()) | |
| Expect(deployList.Items).To(HaveLen(1)) | |
| By("Verifying still only 1 Deployment exists") | |
| Eventually(func(g Gomega) { | |
| err := k8sClient.List(ctx, deployList, | |
| client.InNamespace(ns.Name), | |
| client.MatchingLabels{"axon.io/taskspawner": ts.Name}, | |
| ) | |
| g.Expect(err).Should(Succeed()) | |
| g.Expect(deployList.Items).To(HaveLen(1)) | |
| }, timeout, interval).Should(Succeed()) |
| } | ||
|
|
||
| for _, item := range newItems { | ||
| taskName := fmt.Sprintf("%s-%s", ts.Name, item.ID) |
There was a problem hiding this comment.
The task name format uses the pattern "taskspawner-name-item-id", but Kubernetes resource names have a 253 character limit and must match DNS subdomain format. If the TaskSpawner name is long or the item ID is large, this could exceed the limit or fail validation. Consider adding validation or truncation logic to ensure generated task names are always valid Kubernetes resource names.
| u := fmt.Sprintf("%s/repos/%s/%s/issues/%d/comments?per_page=100", s.baseURL(), s.Owner, s.Repo, issueNumber) | ||
|
|
||
| req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) | ||
| if err != nil { | ||
| return "", fmt.Errorf("creating request: %w", err) | ||
| } | ||
|
|
||
| if s.Token != "" { | ||
| req.Header.Set("Authorization", "token "+s.Token) | ||
| } | ||
| req.Header.Set("Accept", "application/vnd.github.v3+json") | ||
|
|
||
| resp, err := s.httpClient().Do(req) | ||
| if err != nil { | ||
| return "", fmt.Errorf("fetching comments: %w", err) | ||
| } | ||
| defer resp.Body.Close() | ||
|
|
||
| if resp.StatusCode != http.StatusOK { | ||
| body, _ := io.ReadAll(resp.Body) | ||
| return "", fmt.Errorf("GitHub API returned status %d: %s", resp.StatusCode, string(body)) | ||
| } | ||
|
|
||
| var comments []githubComment | ||
| if err := json.NewDecoder(resp.Body).Decode(&comments); err != nil { | ||
| return "", fmt.Errorf("decoding comments: %w", err) | ||
| } | ||
|
|
||
| var parts []string | ||
| for _, c := range comments { | ||
| parts = append(parts, c.Body) |
There was a problem hiding this comment.
Comment pagination is not implemented. The fetchComments function only fetches the first 100 comments with "per_page=100". If an issue has more than 100 comments, they will be silently truncated. Consider implementing pagination for comments similar to how it's done for issues, or document this limitation.
| u := fmt.Sprintf("%s/repos/%s/%s/issues/%d/comments?per_page=100", s.baseURL(), s.Owner, s.Repo, issueNumber) | |
| req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil) | |
| if err != nil { | |
| return "", fmt.Errorf("creating request: %w", err) | |
| } | |
| if s.Token != "" { | |
| req.Header.Set("Authorization", "token "+s.Token) | |
| } | |
| req.Header.Set("Accept", "application/vnd.github.v3+json") | |
| resp, err := s.httpClient().Do(req) | |
| if err != nil { | |
| return "", fmt.Errorf("fetching comments: %w", err) | |
| } | |
| defer resp.Body.Close() | |
| if resp.StatusCode != http.StatusOK { | |
| body, _ := io.ReadAll(resp.Body) | |
| return "", fmt.Errorf("GitHub API returned status %d: %s", resp.StatusCode, string(body)) | |
| } | |
| var comments []githubComment | |
| if err := json.NewDecoder(resp.Body).Decode(&comments); err != nil { | |
| return "", fmt.Errorf("decoding comments: %w", err) | |
| } | |
| var parts []string | |
| for _, c := range comments { | |
| parts = append(parts, c.Body) | |
| pageURL := fmt.Sprintf("%s/repos/%s/%s/issues/%d/comments?per_page=100", s.baseURL(), s.Owner, s.Repo, issueNumber) | |
| var parts []string | |
| for pageURL != "" { | |
| req, err := http.NewRequestWithContext(ctx, http.MethodGet, pageURL, nil) | |
| if err != nil { | |
| return "", fmt.Errorf("creating request: %w", err) | |
| } | |
| if s.Token != "" { | |
| req.Header.Set("Authorization", "token "+s.Token) | |
| } | |
| req.Header.Set("Accept", "application/vnd.github.v3+json") | |
| resp, err := s.httpClient().Do(req) | |
| if err != nil { | |
| return "", fmt.Errorf("fetching comments: %w", err) | |
| } | |
| defer resp.Body.Close() | |
| if resp.StatusCode != http.StatusOK { | |
| body, _ := io.ReadAll(resp.Body) | |
| return "", fmt.Errorf("GitHub API returned status %d: %s", resp.StatusCode, string(body)) | |
| } | |
| var comments []githubComment | |
| if err := json.NewDecoder(resp.Body).Decode(&comments); err != nil { | |
| return "", fmt.Errorf("decoding comments: %w", err) | |
| } | |
| for _, c := range comments { | |
| parts = append(parts, c.Body) | |
| } | |
| pageURL = parseNextLink(resp.Header.Get("Link")) |
| Containers: []corev1.Container{ | ||
| { | ||
| Name: "spawner", | ||
| Image: SpawnerImage, | ||
| Args: args, | ||
| Env: envVars, | ||
| }, |
There was a problem hiding this comment.
The spawner container lacks resource limits and requests. Without these, the spawner pod can potentially consume unlimited CPU and memory, which could affect cluster stability, especially if many TaskSpawners are created or if the GitHub API returns a large number of issues. Consider adding reasonable resource limits and requests for production use.
| if err := cl.Create(ctx, task); err != nil { | ||
| log.Error(err, "creating Task", "task", taskName) |
There was a problem hiding this comment.
When task creation fails, the error is logged but the loop continues without updating the status. This means failed items are not tracked, and on the next cycle, the spawner will attempt to create the same task again, potentially causing repeated failures. Consider tracking failed items in the status or implementing retry logic with exponential backoff.
| if err := cl.Create(ctx, task); err != nil { | |
| log.Error(err, "creating Task", "task", taskName) | |
| // Retry task creation with exponential backoff to avoid tight failure loops on transient errors. | |
| const maxCreateAttempts = 3 | |
| createBackoff := time.Second | |
| var createErr error | |
| for attempt := 1; attempt <= maxCreateAttempts; attempt++ { | |
| createErr = cl.Create(ctx, task) | |
| if createErr == nil { | |
| break | |
| } | |
| log.Error(createErr, "creating Task", "task", taskName, "attempt", attempt, "maxAttempts", maxCreateAttempts) | |
| if attempt < maxCreateAttempts { | |
| time.Sleep(createBackoff) | |
| createBackoff *= 2 | |
| } | |
| } | |
| if createErr != nil { | |
| // Give up on this item after maxCreateAttempts; it may be retried in a future cycle. |
| body, _ := io.ReadAll(resp.Body) | ||
| return nil, "", fmt.Errorf("GitHub API returned status %d: %s", resp.StatusCode, string(body)) |
There was a problem hiding this comment.
Reading the entire error response body with io.ReadAll without a size limit could lead to memory exhaustion if the GitHub API returns a very large error response. Consider limiting the number of bytes read or using io.LimitReader to prevent potential DoS scenarios.
| defer resp.Body.Close() | ||
|
|
||
| if resp.StatusCode != http.StatusOK { | ||
| body, _ := io.ReadAll(resp.Body) |
There was a problem hiding this comment.
Reading the entire error response body with io.ReadAll without a size limit could lead to memory exhaustion if the GitHub API returns a very large error response. Consider limiting the number of bytes read or using io.LimitReader to prevent potential DoS scenarios.
| body, _ := io.ReadAll(resp.Body) | |
| const maxErrorBodyBytes = 4096 | |
| body, _ := io.ReadAll(io.LimitReader(resp.Body, maxErrorBodyBytes)) |
| // When defines the conditions that trigger task spawning. | ||
| // Exactly one field must be set. | ||
| type When struct { | ||
| // GitHubIssues discovers issues from a GitHub repository. | ||
| // +optional | ||
| GitHubIssues *GitHubIssues `json:"githubIssues,omitempty"` | ||
| } |
There was a problem hiding this comment.
The When struct comment states "Exactly one field must be set" but there's no validation enforcing this constraint. Users could create a TaskSpawner with no source configured (all fields nil) or potentially multiple sources in the future. Consider adding a webhook validator or CRD validation rule to enforce that exactly one source type is specified.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
4 issues found across 23 files
Prompt for AI agents (all issues)
Check if these issues are valid — if so, understand the root cause of each and fix them.
<file name="test/integration/taskspawner_test.go">
<violation number="1" location="test/integration/taskspawner_test.go:313">
P2: Use `Consistently` instead of `time.Sleep` for idempotency verification. Fixed-duration sleeps in tests are flaky - they may pass incorrectly if the sleep is too short, or waste time if too long. `Consistently` is the idiomatic Ginkgo approach for verifying a condition holds over time.</violation>
</file>
<file name="cmd/spawner/main.go">
<violation number="1" location="cmd/spawner/main.go:147">
P2: Missing `AlreadyExists` error check when creating Tasks. If a Task was created but status update failed, subsequent cycles will repeatedly try to create the same Task, logging errors each time without ever updating the status to track it. Should check for `apierrors.IsAlreadyExists(err)` and proceed to update status if the Task already exists.</violation>
</file>
<file name="test/e2e/taskspawner_test.go">
<violation number="1" location="test/e2e/taskspawner_test.go:73">
P2: This test references `claude-credentials` secret in the TaskSpawner YAML but never creates it. The `BeforeEach` deletes existing secrets, and Ginkgo runs each `It` block independently with its own `BeforeEach`/`AfterEach`. This will cause the TaskSpawner to be created with an invalid secret reference.</violation>
</file>
<file name="internal/source/github.go">
<violation number="1" location="internal/source/github.go:160">
P2: Missing pagination for comments. Unlike `fetchAllIssues`, this function only fetches the first 100 comments and ignores pagination links. Issues with more than 100 comments will have their comments silently truncated, potentially losing important context for the AI agent.</violation>
</file>
Since this is your first cubic review, here's how it works:
- cubic automatically reviews your code and comments on bugs and improvements
- Teach cubic by replying to its comments. cubic learns from your replies and gets better over time
- Ask questions if you need clarification on any suggestion
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
| }, | ||
| } | ||
|
|
||
| if err := cl.Create(ctx, task); err != nil { |
There was a problem hiding this comment.
P2: Missing AlreadyExists error check when creating Tasks. If a Task was created but status update failed, subsequent cycles will repeatedly try to create the same Task, logging errors each time without ever updating the status to track it. Should check for apierrors.IsAlreadyExists(err) and proceed to update status if the Task already exists.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At cmd/spawner/main.go, line 147:
<comment>Missing `AlreadyExists` error check when creating Tasks. If a Task was created but status update failed, subsequent cycles will repeatedly try to create the same Task, logging errors each time without ever updating the status to track it. Should check for `apierrors.IsAlreadyExists(err)` and proceed to update status if the Task already exists.</comment>
<file context>
@@ -0,0 +1,225 @@
+ },
+ }
+
+ if err := cl.Create(ctx, task); err != nil {
+ log.Error(err, "creating Task", "task", taskName)
+ continue
</file context>
There was a problem hiding this comment.
I think we did resolve it?
| promptTemplate: "Fix: {{.Title}}\n{{.Body}}" | ||
| pollInterval: 1m | ||
| ` | ||
| Expect(kubectlWithInput(tsYAML, "apply", "-f", "-")).To(Succeed()) |
There was a problem hiding this comment.
P2: This test references claude-credentials secret in the TaskSpawner YAML but never creates it. The BeforeEach deletes existing secrets, and Ginkgo runs each It block independently with its own BeforeEach/AfterEach. This will cause the TaskSpawner to be created with an invalid secret reference.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At test/e2e/taskspawner_test.go, line 73:
<comment>This test references `claude-credentials` secret in the TaskSpawner YAML but never creates it. The `BeforeEach` deletes existing secrets, and Ginkgo runs each `It` block independently with its own `BeforeEach`/`AfterEach`. This will cause the TaskSpawner to be created with an invalid secret reference.</comment>
<file context>
@@ -0,0 +1,147 @@
+ promptTemplate: "Fix: {{.Title}}\n{{.Body}}"
+ pollInterval: 1m
+`
+ Expect(kubectlWithInput(tsYAML, "apply", "-f", "-")).To(Succeed())
+
+ By("waiting for Deployment to become available")
</file context>
| } | ||
|
|
||
| func (s *GitHubSource) fetchComments(ctx context.Context, issueNumber int) (string, error) { | ||
| u := fmt.Sprintf("%s/repos/%s/%s/issues/%d/comments?per_page=100", s.baseURL(), s.Owner, s.Repo, issueNumber) |
There was a problem hiding this comment.
P2: Missing pagination for comments. Unlike fetchAllIssues, this function only fetches the first 100 comments and ignores pagination links. Issues with more than 100 comments will have their comments silently truncated, potentially losing important context for the AI agent.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. At internal/source/github.go, line 160:
<comment>Missing pagination for comments. Unlike `fetchAllIssues`, this function only fetches the first 100 comments and ignores pagination links. Issues with more than 100 comments will have their comments silently truncated, potentially losing important context for the AI agent.</comment>
<file context>
@@ -0,0 +1,204 @@
+}
+
+func (s *GitHubSource) fetchComments(ctx context.Context, issueNumber int) (string, error) {
+ u := fmt.Sprintf("%s/repos/%s/%s/issues/%d/comments?per_page=100", s.baseURL(), s.Owner, s.Repo, issueNumber)
+
+ req, err := http.NewRequestWithContext(ctx, http.MethodGet, u, nil)
</file context>
There was a problem hiding this comment.
2 issues found across 4 files (changes from recent commits).
Prompt for AI agents (all issues)
Check if these issues are valid — if so, understand the root cause of each and fix them.
<file name="internal/controller/taskspawner_controller.go">
<violation number="1" location="internal/controller/taskspawner_controller.go:175">
P2: Misleading log: "created ServiceAccount" is logged even when creation returns `IsAlreadyExists`. Move the log inside an else branch so it only logs when creation actually succeeds.</violation>
<violation number="2" location="internal/controller/taskspawner_controller.go:206">
P2: Misleading log: "created RoleBinding" is logged even when creation returns `IsAlreadyExists`. Move the log inside an else branch so it only logs when creation actually succeeds.</violation>
</file>
Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
9d3e344 to
672cbe8
Compare
…ources Add a new TaskSpawner CRD that watches external sources (e.g., GitHub Issues) and automatically creates Task resources for each discovered work item. Architecture: - Main controller creates a Deployment per TaskSpawner - Spawner binary (cmd/axon-spawner) runs in the Deployment, polls the configured source, and creates Tasks with deduplication - Source abstraction in internal/source/ (interface-based, GitHub implementation with pagination limits and comment size caps) - Separate ServiceAccount and RBAC for spawner Deployments (least-privilege: read TaskSpawners, create Tasks) - Deployment reconciliation when TaskSpawner spec changes Includes unit tests, integration tests (envtest), and e2e tests.
Summary
internal/source/) with a GitHub implementation supporting pagination, auth, label/state filtering, and comment fetchingaxon get taskspawner), RBAC updates, and a Dockerfile for the spawner binaryArchitecture
New Files
api/v1alpha1/taskspawner_types.gointernal/source/{source,github,prompt}.gocmd/spawner/main.go+Dockerfileinternal/controller/taskspawner_controller.gointernal/controller/taskspawner_deployment_builder.gotest/integration/taskspawner_test.gotest/e2e/taskspawner_test.gointernal/source/*_test.goTest plan
make build— all 3 binaries compilemake test— unit tests pass (13 source tests)make test-integration— all 8 integration tests pass (4 existing + 4 new)go vet ./...— cleanmake test-e2e— requires cluster +GITHUB_TOKEN+CLAUDE_CODE_OAUTH_TOKENmake image WHAT=cmd/spawner— spawner image builds🤖 Generated with Claude Code