Method flush

coderd/agentapi/metadatabatcher/metadata_batcher.go:300–398

flush flushes the current batch to the database and pubsub.

(ctx context.Context, reason string)

// flush flushes the current batch to the database and pubsub.
func (b *Batcher) flush(ctx context.Context, reason string) {
	count := len(b.batch)

	if count == 0 {
		return
	}

	start := b.clock.Now()
	b.log.Debug(ctx, "flushing metadata batch",
		slog.F("reason", reason),
		slog.F("count", count),
	)

	// Convert batch map to parallel arrays for the batch query.
	// Also build map of agent IDs for per-agent metrics and pubsub.
	var (
		agentIDs    = make([]uuid.UUID, 0, count)
		keys        = make([]string, 0, count)
		values      = make([]string, 0, count)
		errors      = make([]string, 0, count)
		collectedAt = make([]time.Time, 0, count)
		agentKeys   = make(map[uuid.UUID]int) // Track keys per agent for metrics
	)

	for ck, mv := range b.batch {
		agentIDs = append(agentIDs, ck.agentID)
		keys = append(keys, ck.key)
		values = append(values, mv.v)
		errors = append(errors, mv.error)
		collectedAt = append(collectedAt, mv.collectedAt)
		agentKeys[ck.agentID]++
	}

	// Batch has been processed into slices for our DB request, so we can clear it.
	// It's safe to clear before we know whether the flush is successful as agent metadata is not critical, and therefore
	// we do not retry failed flushes and losing a batch of metadata is okay.
	b.batch = make(map[compositeKey]value)
	b.currentBatchLen.Store(0)

	// Record per-agent utilization metrics.
	for _, keyCount := range agentKeys {
		b.Metrics.BatchUtilization.Observe(float64(keyCount))
	}

	// Update the database with all metadata updates in a single query.
	err := b.store.BatchUpdateWorkspaceAgentMetadata(ctx, database.BatchUpdateWorkspaceAgentMetadataParams{
		WorkspaceAgentID: agentIDs,
		Key:              keys,
		Value:            values,
		Error:            errors,
		CollectedAt:      collectedAt,
	})
	elapsed := b.clock.Since(start)

	if err != nil {
		if database.IsQueryCanceledError(err) {
			b.log.Debug(ctx, "query canceled, skipping update of workspace agent metadata", slog.F("elapsed", elapsed))
			return