Merge pull request #549 from buildkite/chore/informer_debug_logs

chore: add additional logging and metrics for limiter

Author: petetomasik

internal/controller/limiter/limiter.go

@@ -73,6 +73,18 @@ func (l *MaxInFlight) RegisterInformer(ctx context.Context, factory informers.Sh
 		return fmt.Errorf("failed to sync informer cache")
 	}
 
+	informerStore := jobInformer.GetStore()
+	informerCacheKeys := informerStore.ListKeys()
+	l.logger.Debug("informer cache sync complete, dump informer cache keys...",
+		zap.Int("cache-keys-found", len(informerCacheKeys)),
+	)
+
+	for _, cacheKey := range informerCacheKeys {
+		l.logger.Debug("informer cache key found",
+			zap.String("informer-cache-key", fmt.Sprint(cacheKey)),
+		)
+	}
+
 	return nil
 }
 
@@ -83,6 +95,17 @@ func (l *MaxInFlight) Handle(ctx context.Context, job model.Job) error {
 	// Block until there's a token in the bucket, or cancel if the job
 	// information becomes too stale.
 	start := time.Now()
+
+	l.logger.Debug("Waiting for token",
+		zap.String("job-uuid", job.Uuid),
+		zap.Int("max-in-flight", l.MaxInFlight),
+		zap.Int("available-tokens", len(l.tokenBucket)),
+		zap.Int("running-jobs", l.MaxInFlight-len(l.tokenBucket)),
+	)
+
+	waitingForTokenGauge.Inc()
+	defer waitingForTokenGauge.Dec()
+
 	select {
 	case <-ctx.Done():
 		return context.Cause(ctx)
@@ -92,8 +115,11 @@ func (l *MaxInFlight) Handle(ctx context.Context, job model.Job) error {
 		l.logger.Debug("Unable to create job before stale-job-data-timeout",
 			zap.String("job-uuid", job.Uuid),
 			zap.Int("max-in-flight", l.MaxInFlight),
+			zap.Int("available-tokens", len(l.tokenBucket)),
 			zap.Int("running-jobs", l.MaxInFlight-len(l.tokenBucket)),
+			zap.Duration("wait-time", time.Since(start)),
 		)
+		jobStaleWhileWaitingCounter.Inc()
 		return model.ErrStaleJob
 
 	case <-l.tokenBucket:
@@ -198,7 +224,15 @@ func (l *MaxInFlight) tryTakeToken(source string) {
 	select {
 	case <-l.tokenBucket:
 		// Success.
+		l.logger.Debug("Token taken successfully",
+			zap.String("source", source),
+			zap.Int("remaining-tokens", len(l.tokenBucket)),
+		)
 	default:
+		l.logger.Debug("Failed to take token - bucket empty",
+			zap.String("source", source),
+			zap.Int("max-in-flight", l.MaxInFlight),
+		)
 		tokenUnderflowCounter.WithLabelValues(source).Inc()
 	}
 }
@@ -208,7 +242,15 @@ func (l *MaxInFlight) tryReturnToken(source string) {
 	select {
 	case l.tokenBucket <- struct{}{}:
 		// Success.
+		l.logger.Debug("Token returned successfully",
+			zap.String("source", source),
+			zap.Int("available-tokens", len(l.tokenBucket)),
+		)
 	default:
+		l.logger.Warn("Failed to return token - bucket full",
+			zap.String("source", source),
+			zap.Int("max-in-flight", l.MaxInFlight),
+		)
 		tokenOverflowCounter.WithLabelValues(source).Inc()
 	}
 }

internal/controller/limiter/metrics.go

@@ -35,6 +35,20 @@ var (
 		NativeHistogramZeroThreshold: 0.01,
 	})
 
+	waitingForTokenGauge = promauto.NewGauge(prometheus.GaugeOpts{
+		Namespace: promNamespace,
+		Subsystem: promSubsystem,
+		Name:      "waiting_for_token",
+		Help:      "Number of jobs currently waiting for a token",
+	})
+
+	jobStaleWhileWaitingCounter = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: promNamespace,
+		Subsystem: promSubsystem,
+		Name:      "jobs_stale_while_waiting_total",
+		Help:      "Count of jobs that became stale while waiting for a token",
+	})
+
 	jobHandlerCallsCounter = promauto.NewCounter(prometheus.CounterOpts{
 		Namespace: promNamespace,
 		Subsystem: promSubsystem,

internal/controller/monitor/monitor.go

@@ -109,6 +109,10 @@ func (r clusteredJobResp) CommandJobs() []*api.JobJobTypeCommand {
 // getScheduledCommandJobs calls either the clustered or unclustered GraphQL API
 // methods, depending on if a cluster uuid was provided in the config
 func (m *Monitor) getScheduledCommandJobs(ctx context.Context, queue string) (jobResp jobResp, err error) {
+	m.logger.Debug("getting scheduled jobs via GQL query...",
+		zap.Duration("poll-interval", m.cfg.PollInterval),
+	)
+
 	jobQueryCounter.Inc()
 	start := time.Now()
 	defer func() {
@@ -274,6 +278,9 @@ func (m *Monitor) Start(ctx context.Context, handler model.JobHandler) <-chan er
 			}
 
 			jobs := resp.CommandJobs()
+			m.logger.Debug("jobs in GQL query response(s)...",
+				zap.Int("jobs-processed", len(jobs)),
+			)
 			if len(jobs) == 0 {
 				continue
 			}