Use cluster endpoint at startup (#90)

* add logic to process syncEvents, prevent sync events from entering queue

Signed-off-by: Eric Pickard <piceri@github.com>

* add PostCluster to client

Signed-off-by: Eric Pickard <piceri@github.com>

* add logic to fillCaches after PostCluster

Signed-off-by: Eric Pickard <piceri@github.com>

* enable partial success, refactors and improvements

Signed-off-by: Eric Pickard <piceri@github.com>

* add tests for processSyncEvents

Signed-off-by: Eric Pickard <piceri@github.com>

* add tests for PostCluster

Signed-off-by: Eric Pickard <piceri@github.com>

* update url variable names

Signed-off-by: Eric Pickard <piceri@github.com>

* refactor record.go, address linting items

Signed-off-by: Eric Pickard <piceri@github.com>

* add logging and comments

Signed-off-by: Eric Pickard <piceri@github.com>

* adjust syncRecords to include terminated jobs

Signed-off-by: Eric Pickard <piceri@github.com>

* add logging to non-rate limit 4xx responses

Signed-off-by: Eric Pickard <piceri@github.com>

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

* fix: lint error

* fix: address PR comments

* fix: address PR comments

* fix: update comment

* chore: remove dead code

---------

Signed-off-by: Eric Pickard <piceri@github.com>
Co-authored-by: Austin Beattie <ajbeattie@github.com>

Author: piceri

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,11 @@ 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 not yet available
+	// for public use.**
+	BulkClusterSync bool
 }
 
 // ValidTemplate verifies that at least one placeholder is present

internal/controller/controller.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"log/slog"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	"github.com/github/deployment-tracker/internal/metadata"
@@ -47,7 +48,9 @@ type ttlCache interface {
 }
 
 type deploymentRecordPoster interface {
-	PostOne(ctx context.Context, record *deploymentrecord.DeploymentRecord) error
+	PostOne(ctx context.Context, record *deploymentrecord.Record) 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 {
@@ -87,6 +90,11 @@ type Controller struct {
 	// informerSyncTimeout is the maximum time allowed for all informers to sync
 	// and prevents sync from hanging indefinitely.
 	informerSyncTimeout time.Duration
+	// 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
 }
 
 // New creates a new deployment tracker controller.
@@ -147,10 +155,21 @@ func New(clientset kubernetes.Interface, metadataAggregator podMetadataAggregato
 		unknownArtifacts:    amcache.NewExpiring(),
 		informerSyncTimeout: informerSyncTimeoutDuration,
 	}
+	// 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{
 		AddFunc: func(obj any) {
+			// Skip adding sync events
+			if cntrl.syncing.Load() {
+				return
+			}
+
 			pod, ok := obj.(*corev1.Pod)
 			if !ok {
 				slog.Error("Invalid object returned",
@@ -314,6 +333,12 @@ func (c *Controller) Run(ctx context.Context, workers int) error {
 			return errors.New("timed out waiting for caches to sync - please ensure deployment tracker has the correct kubernetes permissions")
 		}
 	}
+	c.syncing.Store(false)
+	syncClusterPods := c.podInformer.GetIndexer().List()
+	err := c.processSyncEvents(ctx, syncClusterPods)
+	if err != nil {
+		return fmt.Errorf("sync events failed: %w", err)
+	}
 
 	slog.Info("Starting workers",
 		"count", workers,

internal/controller/controller_integration_test.go

@@ -26,19 +26,27 @@ import (
 
 type mockRecordPoster struct {
 	mu      sync.Mutex
-	records []*deploymentrecord.DeploymentRecord
+	records []*deploymentrecord.Record
 	err     error // to simulate failures
 }
 
-func (m *mockRecordPoster) PostOne(_ context.Context, record *deploymentrecord.DeploymentRecord) error {
+func (m *mockRecordPoster) PostOne(_ context.Context, record *deploymentrecord.Record) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	m.records = append(m.records, record)
 	return m.err
 }
 
+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.
-func (m *mockRecordPoster) getRecords() []*deploymentrecord.DeploymentRecord {
+func (m *mockRecordPoster) getRecords() []*deploymentrecord.Record {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	return slices.Clone(m.records)

internal/controller/controller_test.go

@@ -8,6 +8,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/github/deployment-tracker/internal/metadata"
 	"github.com/github/deployment-tracker/internal/workload"
 	"github.com/github/deployment-tracker/pkg/deploymentrecord"
 	"github.com/stretchr/testify/assert"
@@ -23,35 +24,78 @@ import (
 
 // mockPoster records all PostOne calls and returns a configurable error.
 type mockPoster struct {
-	mu      sync.Mutex
-	calls   int
-	lastErr error
+	mu                 sync.Mutex
+	calls              int
+	clusterRecordCount int
+	jobCalls           int
+	jobWaitCalls       int
+	lastErr            error
+	jobResp            *deploymentrecord.JobResponse
+	jobErr             error
+	jobStatus          *deploymentrecord.JobStatus
+	jobWaitErr         error
 }
 
-func (m *mockPoster) PostOne(_ context.Context, _ *deploymentrecord.DeploymentRecord) error {
+func (m *mockPoster) PostOne(_ context.Context, _ *deploymentrecord.Record) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	m.calls++
 	return m.lastErr
 }
 
-func (m *mockPoster) getCalls() int {
+func (m *mockPoster) CreateClusterJob(_ context.Context, records []*deploymentrecord.Record, _ string) (*deploymentrecord.JobResponse, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	m.jobCalls++
+	m.clusterRecordCount = len(records)
+	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 {
 	m.mu.Lock()
 	defer m.mu.Unlock()
 	return m.calls
 }
 
+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.jobWaitCalls
+}
+
 // mockResolver is a test double for the workloadResolver interface.
-type mockResolver struct{}
+type mockResolver struct {
+	name string
+}
 
-func (*mockResolver) Resolve(_ *corev1.Pod) workload.Identity {
-	return workload.Identity{}
+func (m *mockResolver) Resolve(_ *corev1.Pod) workload.Identity {
+	return workload.Identity{Name: m.name}
 }
 
 func (*mockResolver) IsActive(_ string, _ workload.Identity) bool {
 	return false
 }
 
+// mockMetadataAggregator is a test double for the podMetadataAggregator interface.
+type mockMetadataAggregator struct{}
+
+func (*mockMetadataAggregator) BuildAggregatePodMetadata(_ context.Context, _ *metav1.PartialObjectMetadata) *metadata.AggregatePodMetadata {
+	return nil
+}
+
 // newTestController creates a minimal Controller suitable for unit-testing
 // recordContainer without a real Kubernetes cluster.
 func newTestController(poster *mockPoster) *Controller {
@@ -62,8 +106,10 @@ func newTestController(poster *mockPoster) *Controller {
 			LogicalEnvironment:  "test",
 			PhysicalEnvironment: "test",
 			Cluster:             "test",
+			BulkClusterSync:     true,
 		},
 		workloadResolver:    &mockResolver{},
+		metadataAggregator:  &mockMetadataAggregator{},
 		observedDeployments: amcache.NewExpiring(),
 		unknownArtifacts:    amcache.NewExpiring(),
 	}