feat: use new cluster job endpoint in place of synchronous set cluster endpoint

Author: ajbeattie

cmd/deployment-tracker/main.go

@@ -96,6 +96,7 @@ func main() {
 		GHAppPrivateKey:     ghAppPrivateKey,
 		GHAppPrivateKeyPath: getEnvOrDefault("GH_APP_PRIV_KEY_PATH", ""),
 		Organization:        os.Getenv("GITHUB_ORG"),
+		BulkClusterSync:     strings.EqualFold(getEnvOrDefault("BULK_CLUSTER_SYNC", "false"), "true"),
 	}
 
 	if len(cntrlCfg.GHAppPrivateKey) > 0 && cntrlCfg.GHAppPrivateKeyPath != "" {

internal/controller/config.go

@@ -29,6 +29,13 @@ type Config struct {
 	GHAppPrivateKey     []byte
 	GHAppPrivateKeyPath string
 	Organization        string
+	// BulkClusterSync enables the async cluster job endpoint for startup
+	// state sync. When false, startup sync is skipped and only individual
+	// PostOne calls are used. Note: this is experimental and requires
+	// enablement of a feature flag by GitHub. If enabled without the API-side
+	// feature flag, the controller will not post deployment records at startup,
+	// only ongoing pod events that arrive after the initial informer sync.
+	BulkClusterSync bool
 }
 
 // ValidTemplate verifies that at least one placeholder is present

internal/controller/controller.go

@@ -49,7 +49,8 @@ type ttlCache interface {
 
 type deploymentRecordPoster interface {
 	PostOne(ctx context.Context, record *deploymentrecord.Record) error
-	PostCluster(ctx context.Context, records []*deploymentrecord.Record, cluster string) ([]byte, error)
+	CreateClusterJob(ctx context.Context, records []*deploymentrecord.Record, cluster string) (*deploymentrecord.JobResponse, error)
+	WaitForClusterJob(ctx context.Context, cluster string, jobID int64) (*deploymentrecord.JobStatus, error)
 }
 
 type podMetadataAggregator interface {
@@ -89,7 +90,10 @@ type Controller struct {
 	// informerSyncTimeout is the maximum time allowed for all informers to sync
 	// and prevents sync from hanging indefinitely.
 	informerSyncTimeout time.Duration
-	// syncing tracks if the kubernetes informers have finished syncing
+	// syncing gates informer event handlers during startup. When true,
+	// pod add events are suppressed so they can be reported via the bulk
+	// cluster job instead of individual PostOne calls. Only set when
+	// BulkClusterSync is enabled.
 	syncing atomic.Bool
 }
 
@@ -151,7 +155,12 @@ func New(clientset kubernetes.Interface, metadataAggregator podMetadataAggregato
 		unknownArtifacts:    amcache.NewExpiring(),
 		informerSyncTimeout: informerSyncTimeoutDuration,
 	}
-	cntrl.syncing.Store(true)
+	// Only gate informer events when bulk cluster sync is enabled.
+	// When disabled, all pods discovered during informer sync will be
+	// enqueued as individual events.
+	if cfg.BulkClusterSync {
+		cntrl.syncing.Store(true)
+	}
 
 	// Add event handlers to the informer
 	_, err = podInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{

internal/controller/controller_integration_test.go

@@ -37,8 +37,12 @@ func (m *mockRecordPoster) PostOne(_ context.Context, record *deploymentrecord.R
 	return m.err
 }
 
-func (m *mockRecordPoster) PostCluster(_ context.Context, _ []*deploymentrecord.Record, _ string) ([]byte, error) {
-	return nil, nil
+func (m *mockRecordPoster) CreateClusterJob(_ context.Context, _ []*deploymentrecord.Record, _ string) (*deploymentrecord.JobResponse, error) {
+	return &deploymentrecord.JobResponse{}, nil
+}
+
+func (m *mockRecordPoster) WaitForClusterJob(_ context.Context, _ string, _ int64) (*deploymentrecord.JobStatus, error) {
+	return &deploymentrecord.JobStatus{Status: "completed"}, nil
 }
 
 // Helper that allows tests to read captured records safely.

internal/controller/controller_test.go

@@ -26,11 +26,14 @@ import (
 type mockPoster struct {
 	mu                 sync.Mutex
 	calls              int
-	clusterCalls       int
 	clusterRecordCount int
+	jobCalls           int
+	jobWaitCalls       int
 	lastErr            error
-	clusterResp        []byte
-	clusterErr         error
+	jobResp            *deploymentrecord.JobResponse
+	jobErr             error
+	jobStatus          *deploymentrecord.JobStatus
+	jobWaitErr         error
 }
 
 func (m *mockPoster) PostOne(_ context.Context, _ *deploymentrecord.Record) error {
@@ -40,12 +43,19 @@ func (m *mockPoster) PostOne(_ context.Context, _ *deploymentrecord.Record) erro
 	return m.lastErr
 }
 
-func (m *mockPoster) PostCluster(_ context.Context, records []*deploymentrecord.Record, _ string) ([]byte, error) {
+func (m *mockPoster) CreateClusterJob(_ context.Context, records []*deploymentrecord.Record, _ string) (*deploymentrecord.JobResponse, error) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	m.clusterCalls++
+	m.jobCalls++
 	m.clusterRecordCount = len(records)
-	return m.clusterResp, m.clusterErr
+	return m.jobResp, m.jobErr
+}
+
+func (m *mockPoster) WaitForClusterJob(_ context.Context, _ string, _ int64) (*deploymentrecord.JobStatus, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.jobWaitCalls++
+	return m.jobStatus, m.jobWaitErr
 }
 
 func (m *mockPoster) getPostOneCalls() int {
@@ -54,10 +64,16 @@ func (m *mockPoster) getPostOneCalls() int {
 	return m.calls
 }
 
-func (m *mockPoster) getPostClusterCalls() int {
+func (m *mockPoster) getCreateClusterJobCalls() int {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	return m.jobCalls
+}
+
+func (m *mockPoster) getWaitForClusterJobCalls() int {
 	m.mu.Lock()
 	defer m.mu.Unlock()
-	return m.clusterCalls
+	return m.jobWaitCalls
 }
 
 // mockResolver is a test double for the workloadResolver interface.
@@ -90,6 +106,7 @@ func newTestController(poster *mockPoster) *Controller {
 			LogicalEnvironment:  "test",
 			PhysicalEnvironment: "test",
 			Cluster:             "test",
+			BulkClusterSync:     true,
 		},
 		workloadResolver:    &mockResolver{},
 		metadataAggregator:  &mockMetadataAggregator{},

internal/controller/reporting.go

@@ -2,7 +2,6 @@ package controller
 
 import (
 	"context"
-	"encoding/json"
 	"errors"
 	"fmt"
 	"log/slog"
@@ -118,42 +117,73 @@ func (c *Controller) processEvent(ctx context.Context, event PodEvent) error {
 }
 
 func (c *Controller) processSyncEvents(ctx context.Context, syncClusterPods []any) error {
+	if !c.cfg.BulkClusterSync {
+		slog.Info("Async cluster sync disabled, skipping startup sync")
+		return nil
+	}
+
 	syncRecords := c.makeSyncRecords(ctx, syncClusterPods)
 	if len(syncRecords) == 0 {
 		slog.Info("No sync records to post")
 		return nil
 	}
 
-	respBody, err := c.apiClient.PostCluster(ctx, syncRecords, c.cfg.Cluster)
-	var clusterNoRepositoriesError *deploymentrecord.ClusterNoRepositoriesError
+	jobResp, err := c.apiClient.CreateClusterJob(ctx, syncRecords, c.cfg.Cluster)
 	if err != nil {
-		if errors.As(err, &clusterNoRepositoriesError) {
-			slog.Info("Cluster sync found no creatable records",
+		var conflictErr *deploymentrecord.ClusterJobConflictError
+		var noReposErr *deploymentrecord.ClusterNoRepositoriesError
+
+		switch {
+		case errors.As(err, &conflictErr):
+			slog.Warn("Cluster job already in progress, skipping startup sync",
 				"org", c.cfg.Organization,
 			)
+			c.fillCachesFromSubmitted(syncRecords)
 			return nil
+
+		case errors.As(err, &noReposErr):
+			slog.Info("Async cluster endpoint not available, skipping startup sync",
+				"org", c.cfg.Organization,
+			)
+			return nil
+
+		default:
+			slog.Error("Failed to create cluster job",
+				"error", err,
+				"record_count", len(syncRecords),
+			)
+			return fmt.Errorf("failed to create cluster job: %w", err)
 		}
-		slog.Error("Failed to post sync cluster records",
-			"error", err,
-			"record_count", len(syncRecords),
+	}
+
+	if len(jobResp.Errors) > 0 {
+		slog.Warn("Some deployments rejected from job submission",
+			"job_id", jobResp.JobID,
+			"rejected_count", len(jobResp.Errors),
 		)
-		return fmt.Errorf("failed to post sync cluster records: %w", err)
 	}
-	var deploymentRecords deploymentrecord.RecordsClusterResp
-	err = json.Unmarshal(respBody, &deploymentRecords)
+
+	// Wait for job completion with a timeout to prevent indefinite startup delay.
+	jobCtx, cancel := context.WithTimeout(ctx, 5*time.Minute)
+	defer cancel()
+	jobStatus, err := c.apiClient.WaitForClusterJob(jobCtx, c.cfg.Cluster, jobResp.JobID)
 	if err != nil {
-		slog.Error("Failed to unmarshall response",
+		slog.Error("Failed waiting for cluster job, filling caches from submitted records",
+			"job_id", jobResp.JobID,
 			"error", err,
-			"record_count", len(syncRecords),
 		)
+		c.fillCachesFromSubmitted(syncRecords)
 		return nil
 	}
-	slog.Info("Successfully posted sync cluster records",
-		"created", len(deploymentRecords.DeploymentRecords),
-		"errors", len(deploymentRecords.Errors),
+
+	slog.Info("Cluster job completed",
+		"job_id", jobResp.JobID,
+		"status", jobStatus.Status,
+		"total_count", jobStatus.TotalCount,
+		"errors", len(jobStatus.Errors),
 	)
 
-	c.fillCaches(deploymentRecords)
+	c.fillCachesFromJobResult(syncRecords, jobResp, jobStatus)
 	return nil
 }
 
@@ -224,20 +254,51 @@ func (c *Controller) makeSyncRecords(ctx context.Context, syncClusterPods []any)
 	return syncRecords
 }
 
-func (c *Controller) fillCaches(deploymentRecords deploymentrecord.RecordsClusterResp) {
-	slog.Info("Filling caches after posting sync cluster records")
-	// Fill observedDeployments cache with successful digests
-	for _, r := range deploymentRecords.DeploymentRecords {
+// fillCachesFromSubmitted populates the observedDeployments cache from the
+// records we submitted, without waiting for a response. Used when we can't
+// get a response (409 conflict, wait timeout, job failure).
+func (c *Controller) fillCachesFromSubmitted(records []*deploymentrecord.Record) {
+	slog.Info("Filling observedDeployments cache from submitted records",
+		"count", len(records),
+	)
+	for _, r := range records {
 		cacheKey := getCacheKey(EventCreated, r.DeploymentName, r.Digest)
 		c.observedDeployments.Set(cacheKey, true, 2*time.Minute)
 	}
+}
+
+// fillCachesFromJobResult populates both caches after an async job completes.
+// observedDeployments is filled from submitted records, and unknownArtifacts
+// is filled from error responses with cause "not_found".
+func (c *Controller) fillCachesFromJobResult(records []*deploymentrecord.Record, jobResp *deploymentrecord.JobResponse, jobStatus *deploymentrecord.JobStatus) {
+	slog.Info("Filling caches after cluster job completion",
+		"record_count", len(records),
+	)
 
-	// Fill unknownArtifacts cache with unknown digests
-	for _, r := range deploymentRecords.Errors {
-		if r.Cause == "not_found" {
-			c.unknownArtifacts.Set(r.Digest, true, unknownArtifactTTL)
+	// Build a name→digests lookup from submitted records so we can
+	// key unknownArtifacts by digest (which is how recordContainer looks them up).
+	// Multiple records can share the same image name with different digests,
+	// so we only cache when the mapping is unambiguous (exactly one digest per name).
+	nameToDigests := make(map[string][]string, len(records))
+	for _, r := range records {
+		cacheKey := getCacheKey(EventCreated, r.DeploymentName, r.Digest)
+		c.observedDeployments.Set(cacheKey, true, 2*time.Minute)
+		nameToDigests[r.Name] = append(nameToDigests[r.Name], r.Digest)
+	}
+
+	cacheUnknownDigests := func(errors []deploymentrecord.JobError) {
+		for _, e := range errors {
+			if e.Cause == "not_found" {
+				if digests, ok := nameToDigests[e.Name]; ok && len(digests) == 1 {
+					c.unknownArtifacts.Set(digests[0], true, unknownArtifactTTL)
+				}
+			}
 		}
 	}
+
+	// Fill unknownArtifacts from job submission and completion errors
+	cacheUnknownDigests(jobResp.Errors)
+	cacheUnknownDigests(jobStatus.Errors)
 }
 
 // recordContainer records a single container's deployment info.