fix(event-ledger): support dotted instance IDs and bulk CloudEvent writes

src/control-plane-services/event-ledger/README.md

@@ -78,8 +78,9 @@ It may also include `instance_id`, `deployment_id`, `gpu_specification_id`, and
 
 ### Read events
 
-Pass the same context fields as query parameters. Values may contain letters,
-numbers, and dashes.
+Pass the same context fields as query parameters. Context values may contain
+letters, numbers, and dashes. Instance IDs may also contain dots between
+non-empty segments.
 
 ```bash
 curl 'http://localhost:8080/v3/ledger/namespace/example/events?instance_id=instance-1'

src/control-plane-services/event-ledger/cmd/api/service/v3.go

@@ -47,8 +47,10 @@ import (
 
 var (
 	// contextFieldPattern validates context field values (alphanumeric and dashes only)
-	contextFieldPattern   = regexp.MustCompile(`^[a-zA-Z0-9-]+$`)
-	namespaceFieldPattern = regexp.MustCompile(`^[a-zA-Z0-9-]+$`)
+	contextFieldPattern = regexp.MustCompile(`^[a-zA-Z0-9-]+$`)
+	// instanceIDFieldPattern additionally permits dot-separated instance ID segments.
+	instanceIDFieldPattern = regexp.MustCompile(`^[a-zA-Z0-9-]+(?:\.[a-zA-Z0-9-]+)*$`)
+	namespaceFieldPattern  = regexp.MustCompile(`^[a-zA-Z0-9-]+$`)
 
 	ErrMissingEventName = errors.New("missing required field: event_name")
 	ErrMissingNamespace = errors.New("missing required field: namespace")
@@ -363,12 +365,20 @@ func deduplicateEvents(events []*EventV3) []*EventV3 {
 
 // eventContextToCanonical converts a ContextV3 struct to a canonical string representation
 // Format: key1=value1,key2=value2 (alphabetical order: cluster_id, deployment_id, gpu_specification_id, instance_id)
-// Validates that values contain only alphanumeric characters and dashes. Empty fields are omitted.
+// Validates that values contain only alphanumeric characters and dashes. Instance IDs may also
+// contain dots between non-empty segments. Empty fields are omitted.
 func eventContextToCanonical(eventContext ContextV3) (string, error) {
 	// Helper to validate field values
 	validate := func(name, value string) error {
-		if value != "" && !contextFieldPattern.MatchString(value) {
-			return fmt.Errorf("invalid %s '%s': must contain only alphanumeric characters and dashes", name, value)
+		pattern := contextFieldPattern
+		allowedCharacters := "alphanumeric characters and dashes"
+		if name == "instance_id" {
+			pattern = instanceIDFieldPattern
+			allowedCharacters = "alphanumeric characters, dashes, and dots between segments"
+		}
+
+		if value != "" && !pattern.MatchString(value) {
+			return fmt.Errorf("invalid %s '%s': must contain only %s", name, value, allowedCharacters)
 		}
 
 		if len(value) > MaxContextLength {
@@ -542,6 +552,116 @@ func extractCloudEvent(ce *cloudevents.Event) (*EventV3, error) {
 	return event, nil
 }
 
+// processCloudEvents validates a CloudEvents request and persists accepted events in bulk.
+// Response counts continue to describe the input events, while duplicate storage keys are
+// reduced to their latest timestamp before persistence.
+func (s *Server) processCloudEvents(traceCtx context.Context, cloudEvents []*cloudevents.Event) EventProcessingResult {
+	logger := logging.GetLogger(traceCtx)
+	result := EventProcessingResult{ProcessedEvents: make([]ProcessedEventSummary, 0, len(cloudEvents))}
+
+	acceptedEvents := make([]*EventV3, 0, len(cloudEvents))
+	for _, cloudEvent := range cloudEvents {
+		if cloudEvent == nil {
+			err := errors.New("CloudEvent must not be null")
+			logger.WarnContext(traceCtx, "Skipping null CloudEvent", zap.Error(err))
+			result.FailureCount++
+			result.LastError = err
+			continue
+		}
+
+		event, err := extractCloudEvent(cloudEvent)
+		if err != nil {
+			logger.WarnContext(traceCtx, "Skipping event", zap.Error(err))
+			result.FailureCount++
+			result.LastError = err
+			continue
+		}
+
+		if !middleware.IsTenantAuthorized(traceCtx, event.Namespace) {
+			err := errors.New("tenant is not authorized")
+			logger.WarnContext(traceCtx, "Skipping unauthorized tenant event")
+			result.FailureCount++
+			result.LastError = err
+			continue
+		}
+
+		acceptedEvents = append(acceptedEvents, event)
+	}
+
+	storageEvents := deduplicateEvents(acceptedEvents)
+	records := make([]data_access.EventV3UpsertRecord, len(storageEvents))
+	for i, event := range storageEvents {
+		records[i] = data_access.EventV3UpsertRecord{
+			Namespace: event.Namespace,
+			Context:   event.Context,
+			EventName: event.EventName,
+			Source:    event.Source,
+			Details:   event.DetailsJSON,
+			Timestamp: event.Timestamp,
+		}
+	}
+
+	if len(records) > 0 {
+		if err := s.conns.DbHandlerV2.BulkUpsertEventsV3(traceCtx, records); err != nil {
+			logger.ErrorContext(traceCtx, "Failed to bulk upsert CloudEvents", zap.Error(err))
+			result.FailureCount += len(acceptedEvents)
+			result.LastError = err
+			return result
+		}
+
+		statsRecords := make([]data_access.EventV3UpsertRecord, 0, len(records))
+		for _, record := range records {
+			if s.isStatsEnabled(record.EventName) {
+				statsRecords = append(statsRecords, record)
+			}
+		}
+		if len(statsRecords) > 0 {
+			if err := s.conns.DbHandlerV2.BulkUpsertStatsV3(traceCtx, statsRecords); err != nil {
+				logger.ErrorContext(traceCtx, "Failed to bulk upsert CloudEvent stats", zap.Error(err))
+				result.LastError = err
+				for _, event := range acceptedEvents {
+					if s.isStatsEnabled(event.EventName) {
+						result.FailureCount++
+						continue
+					}
+					s.completeCloudEvent(traceCtx, event, &result)
+				}
+				return result
+			}
+		}
+	}
+
+	for _, event := range acceptedEvents {
+		s.completeCloudEvent(traceCtx, event, &result)
+	}
+
+	return result
+}
+
+// completeCloudEvent preserves filtered-view writes, which do not have a bulk interface yet,
+// without putting event and primary-stats persistence back on the per-event LWT path.
+func (s *Server) completeCloudEvent(traceCtx context.Context, event *EventV3, result *EventProcessingResult) {
+	if s.isFilteredStatsEnabled(event.EventName) {
+		if err := s.conns.DbHandlerV2.UpsertFilteredStatsV3(traceCtx, event.Namespace, event.Context, event.EventName, event.Timestamp); err != nil {
+			logging.GetLogger(traceCtx).ErrorContext(traceCtx, "Failed to store event in filtered stats view", zap.Error(err))
+			result.FailureCount++
+			result.LastError = err
+			return
+		}
+	}
+	result.addProcessedEvent(event)
+}
+
+func (result *EventProcessingResult) addProcessedEvent(event *EventV3) {
+	result.SuccessCount++
+	result.ProcessedEvents = append(result.ProcessedEvents, ProcessedEventSummary{
+		Namespace: event.Namespace,
+		Context:   event.Context,
+		Name:      event.EventName,
+		Timestamp: event.Timestamp.Format(time.RFC3339),
+	})
+}
+
 // storeK8sEvent persists an event to both events_v3 and optionally stats_v3
 func (s *Server) storeK8sEvent(traceCtx context.Context, event *EventV3) error {
 	logger := logging.GetLogger(traceCtx)
@@ -708,39 +828,7 @@ func (s *Server) PostCloudEventV3(w http.ResponseWriter, r *http.Request) {
 
 	logger.InfoContext(traceCtx, "Parsed CloudEvents", zap.Int("count", len(events)))
 
-	result := EventProcessingResult{ProcessedEvents: make([]ProcessedEventSummary, 0, len(events))}
-	for _, event := range events {
-		eventV3, err := extractCloudEvent(event)
-		if err != nil {
-			logger.WarnContext(traceCtx, "Skipping event", zap.Error(err))
-			result.FailureCount++
-			result.LastError = err
-			continue
-		}
-
-		if !middleware.IsTenantAuthorized(traceCtx, eventV3.Namespace) {
-			err := errors.New("tenant is not authorized")
-			logger.WarnContext(traceCtx, "Skipping unauthorized tenant event")
-			result.FailureCount++
-			result.LastError = err
-			continue
-		}
-
-		if err := s.storeK8sEvent(traceCtx, eventV3); err != nil {
-			logger.ErrorContext(traceCtx, "Failed to store event", zap.Error(err))
-			result.FailureCount++
-			result.LastError = err
-			continue
-		}
-
-		result.SuccessCount++
-		result.ProcessedEvents = append(result.ProcessedEvents, ProcessedEventSummary{
-			Namespace: eventV3.Namespace,
-			Context:   eventV3.Context,
-			Name:      eventV3.EventName,
-			Timestamp: eventV3.Timestamp.Format(time.RFC3339),
-		})
-	}
+	result := s.processCloudEvents(traceCtx, events)
 
 	// Send response using the common response handler
 	s.sendEventResponse(w, traceCtx, result)

src/control-plane-services/event-ledger/cmd/api/service/v3_test.go

@@ -72,12 +72,16 @@ type mockDBHandlerV3 struct {
 	getStatsCalls            int
 	getFilteredStatsCalls    int
 	storedStatsEvents        []data_access.EventV3UpsertRecord
+	upsertEventCalls         int
+	bulkUpsertEventsCalls    int
+	bulkUpsertStatsCalls     int
 	bulkUpsertEventsErr      error
 	bulkUpsertStatsErr       error
 }
 
 // V3 methods
 func (m *mockDBHandlerV3) UpsertEventV3(ctx context.Context, namespace, eventContext, eventName, source string, details json.RawMessage, timestamp time.Time) error {
+	m.upsertEventCalls++
 	m.storedEvents = append(m.storedEvents, struct {
 		namespace string
 		context   string
@@ -100,6 +104,7 @@ func (m *mockDBHandlerV3) UpsertFilteredStatsV3(ctx context.Context, namespace,
 }
 
 func (m *mockDBHandlerV3) BulkUpsertEventsV3(ctx context.Context, events []data_access.EventV3UpsertRecord) error {
+	m.bulkUpsertEventsCalls++
 	if m.bulkUpsertEventsErr != nil {
 		return m.bulkUpsertEventsErr
 	}
@@ -117,6 +122,7 @@ func (m *mockDBHandlerV3) BulkUpsertEventsV3(ctx context.Context, events []data_
 }
 
 func (m *mockDBHandlerV3) BulkUpsertStatsV3(ctx context.Context, events []data_access.EventV3UpsertRecord) error {
+	m.bulkUpsertStatsCalls++
 	if m.bulkUpsertStatsErr != nil {
 		return m.bulkUpsertStatsErr
 	}
@@ -245,6 +251,36 @@ func createOTLPLogRecord(eventName, namespace, source, instanceID string, extraA
 	}
 }
 
+func makeCloudEvent(t *testing.T, id, eventType, instanceID string, timestamp time.Time) *cloudevents.Event {
+	t.Helper()
+	event := cloudevents.NewEvent()
+	event.SetSpecVersion(cloudevents.VersionV1)
+	event.SetID(id)
+	event.SetType(eventType)
+	event.SetSource("/test")
+	event.SetTime(timestamp)
+	event.SetExtension("namespace", "test-namespace")
+	event.SetExtension("instanceId", instanceID)
+	require.NoError(t, event.SetData(cloudevents.ApplicationJSON, map[string]string{"status": "ok"}))
+	return &event
+}
+
+func TestEventContextToCanonical_DotSeparatedInstanceID(t *testing.T) {
+	instanceID := "00000000-0000-4000-8000-000000000001.synthetic-instance"
+
+	canonical, err := eventContextToCanonical(ContextV3{InstanceID: instanceID})
+
+	require.NoError(t, err)
+	assert.Equal(t, "instance_id="+instanceID, canonical)
+}
+
+func TestEventContextToCanonical_DotsRemainInvalidForOtherContextFields(t *testing.T) {
+	_, err := eventContextToCanonical(ContextV3{DeploymentID: "deployment.invalid"})
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "invalid deployment_id")
+}
+
 // Test that namespace is required
 func TestPostK8sEventV3_NamespaceRequired(t *testing.T) {
 	mockDB := &mockDBHandlerV3{}
@@ -705,6 +741,71 @@ func TestPostCloudEventV3_BatchMissingSpecversion(t *testing.T) {
 	assert.Contains(t, w.Body.String(), "specversion")
 }
 
+func TestPostCloudEventV3_BatchRejectsNullEvent(t *testing.T) {
+	w, _ := executeCloudEventsRequest(t, []byte(`[null]`), "application/cloudevents-batch+json")
+
+	assert.Equal(t, http.StatusBadRequest, w.Code)
+	assert.Contains(t, w.Body.String(), "CloudEvent must not be null")
+}
+
+func TestProcessCloudEvents_UsesBulkPersistence(t *testing.T) {
+	mockDB := &mockDBHandlerV3{}
+	server := newServerWithMock(t, mockDB)
+	ctx := makeStoreCtx(server)
+	now := time.Now()
+
+	result := server.processCloudEvents(ctx, []*cloudevents.Event{
+		makeCloudEvent(t, "event-1", "pod.ready", "00000000-0000-4000-8000-000000000001.synthetic-instance", now),
+		makeCloudEvent(t, "event-2", "pod.pending", "pod-2", now),
+	})
+
+	assert.Equal(t, 2, result.SuccessCount)
+	assert.Equal(t, 0, result.FailureCount)
+	assert.Equal(t, 0, mockDB.upsertEventCalls, "per-event LWT path must not be used")
+	assert.Equal(t, 1, mockDB.bulkUpsertEventsCalls)
+	assert.Equal(t, 1, mockDB.bulkUpsertStatsCalls)
+	require.Len(t, mockDB.storedEvents, 2)
+	contexts := []string{mockDB.storedEvents[0].context, mockDB.storedEvents[1].context}
+	assert.Contains(t, contexts,
+		"instance_id=00000000-0000-4000-8000-000000000001.synthetic-instance",
+	)
+}
+
+func TestProcessCloudEvents_DeduplicatesStorageWithoutChangingResultCounts(t *testing.T) {
+	mockDB := &mockDBHandlerV3{}
+	server := newServerWithMock(t, mockDB)
+	ctx := makeStoreCtx(server)
+	latestTimestamp := time.Now()
+
+	result := server.processCloudEvents(ctx, []*cloudevents.Event{
+		makeCloudEvent(t, "event-1", "pod.ready", "pod-1", latestTimestamp.Add(-time.Minute)),
+		makeCloudEvent(t, "event-2", "pod.ready", "pod-1", latestTimestamp),
+	})
+
+	assert.Equal(t, 2, result.SuccessCount)
+	assert.Equal(t, 0, result.FailureCount)
+	assert.Len(t, result.ProcessedEvents, 2)
+	require.Len(t, mockDB.storedEvents, 1)
+	assert.Equal(t, latestTimestamp, mockDB.storedEvents[0].timestamp)
+}
+
+func TestProcessCloudEvents_StatsFailureCountsAcceptedEvents(t *testing.T) {
+	mockDB := &mockDBHandlerV3{bulkUpsertStatsErr: fmt.Errorf("stats unavailable")}
+	server := newServerWithMock(t, mockDB)
+	ctx := makeStoreCtx(server)
+	latestTimestamp := time.Now()
+
+	result := server.processCloudEvents(ctx, []*cloudevents.Event{
+		makeCloudEvent(t, "event-1", "pod.ready", "pod-1", latestTimestamp.Add(-time.Minute)),
+		makeCloudEvent(t, "event-2", "pod.ready", "pod-1", latestTimestamp),
+	})
+
+	assert.Equal(t, 0, result.SuccessCount)
+	assert.Equal(t, 2, result.FailureCount)
+	assert.ErrorContains(t, result.LastError, "stats unavailable")
+	require.Len(t, mockDB.storedEvents, 1)
+}
+
 // Test GetStatsV3 success
 func TestGetStatsV3_Success(t *testing.T) {
 	mockDB := &mockDBHandlerV3{}