From 5fa3e60158489e32b334b9d04fc73e5bc02f2e03 Mon Sep 17 00:00:00 2001 From: Bora Oztekin Date: Mon, 24 Aug 2026 20:36:49 +0000 Subject: [PATCH 1/4] fix(event-ledger): support dotted instance IDs and bulk writes Signed-off-by: Bora Oztekin --- .../event-ledger/cmd/api/service/v3.go | 156 +++++++++++++----- .../event-ledger/cmd/api/service/v3_test.go | 57 +++++++ 2 files changed, 175 insertions(+), 38 deletions(-) diff --git a/src/control-plane-services/event-ledger/cmd/api/service/v3.go b/src/control-plane-services/event-ledger/cmd/api/service/v3.go index c4f1fe41c..6f2ff8847 100644 --- a/src/control-plane-services/event-ledger/cmd/api/service/v3.go +++ b/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,108 @@ 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 { + 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 +820,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) diff --git a/src/control-plane-services/event-ledger/cmd/api/service/v3_test.go b/src/control-plane-services/event-ledger/cmd/api/service/v3_test.go index 39601e61f..dedc10657 100644 --- a/src/control-plane-services/event-ledger/cmd/api/service/v3_test.go +++ b/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,22 @@ func createOTLPLogRecord(eventName, namespace, source, instanceID string, extraA } } +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 +727,41 @@ func TestPostCloudEventV3_BatchMissingSpecversion(t *testing.T) { assert.Contains(t, w.Body.String(), "specversion") } +func TestProcessCloudEvents_UsesBulkPersistence(t *testing.T) { + mockDB := &mockDBHandlerV3{} + server := newServerWithMock(t, mockDB) + ctx := makeStoreCtx(server) + + makeCloudEvent := func(id, eventType, instanceID string) *cloudevents.Event { + event := cloudevents.NewEvent() + event.SetSpecVersion(cloudevents.VersionV1) + event.SetID(id) + event.SetType(eventType) + event.SetSource("/test") + event.SetTime(time.Now()) + event.SetExtension("namespace", "test-namespace") + event.SetExtension("instanceId", instanceID) + require.NoError(t, event.SetData(cloudevents.ApplicationJSON, map[string]string{"status": "ok"})) + return &event + } + + result := server.processCloudEvents(ctx, []*cloudevents.Event{ + makeCloudEvent("event-1", "pod.ready", "00000000-0000-4000-8000-000000000001.synthetic-instance"), + makeCloudEvent("event-2", "pod.pending", "pod-2"), + }) + + 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", + ) +} + // Test GetStatsV3 success func TestGetStatsV3_Success(t *testing.T) { mockDB := &mockDBHandlerV3{} From 43304406e16fe6e6595d4e6df8dbdcdcad36f07b Mon Sep 17 00:00:00 2001 From: Bora Oztekin Date: Mon, 24 Aug 2026 20:46:51 +0000 Subject: [PATCH 2/4] docs(event-ledger): document dotted instance IDs Signed-off-by: Bora Oztekin --- src/control-plane-services/event-ledger/README.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/control-plane-services/event-ledger/README.md b/src/control-plane-services/event-ledger/README.md index 7a5453deb..001e6ad05 100644 --- a/src/control-plane-services/event-ledger/README.md +++ b/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' From f38b1337b57cf20f06afdad98e16a2f7ac9d824a Mon Sep 17 00:00:00 2001 From: Bora Oztekin Date: Mon, 24 Aug 2026 22:18:33 +0000 Subject: [PATCH 3/4] fix(event-ledger): reject null CloudEvents Signed-off-by: Bora Oztekin --- .../event-ledger/cmd/api/service/v3.go | 8 ++++++++ .../event-ledger/cmd/api/service/v3_test.go | 7 +++++++ 2 files changed, 15 insertions(+) diff --git a/src/control-plane-services/event-ledger/cmd/api/service/v3.go b/src/control-plane-services/event-ledger/cmd/api/service/v3.go index 6f2ff8847..f9533f26d 100644 --- a/src/control-plane-services/event-ledger/cmd/api/service/v3.go +++ b/src/control-plane-services/event-ledger/cmd/api/service/v3.go @@ -561,6 +561,14 @@ func (s *Server) processCloudEvents(traceCtx context.Context, cloudEvents []*clo 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)) diff --git a/src/control-plane-services/event-ledger/cmd/api/service/v3_test.go b/src/control-plane-services/event-ledger/cmd/api/service/v3_test.go index dedc10657..0c0ce3ad4 100644 --- a/src/control-plane-services/event-ledger/cmd/api/service/v3_test.go +++ b/src/control-plane-services/event-ledger/cmd/api/service/v3_test.go @@ -727,6 +727,13 @@ 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) From 57c90589a3de7fb112148efdb3263641e75abd17 Mon Sep 17 00:00:00 2001 From: Bora Oztekin Date: Fri, 28 Aug 2026 19:29:01 +0000 Subject: [PATCH 4/4] fix(event-ledger): preserve latest stats event --- .../event-ledger/cmd/api/service/v3_test.go | 67 ++++++++++++---- .../internal/db_client/cassandra/v2.go | 80 +++++++++---------- .../internal/db_client/cassandra/v2_test.go | 35 ++++++++ 3 files changed, 127 insertions(+), 55 deletions(-) diff --git a/src/control-plane-services/event-ledger/cmd/api/service/v3_test.go b/src/control-plane-services/event-ledger/cmd/api/service/v3_test.go index 0c0ce3ad4..cebba676c 100644 --- a/src/control-plane-services/event-ledger/cmd/api/service/v3_test.go +++ b/src/control-plane-services/event-ledger/cmd/api/service/v3_test.go @@ -251,6 +251,20 @@ 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" @@ -738,23 +752,11 @@ func TestProcessCloudEvents_UsesBulkPersistence(t *testing.T) { mockDB := &mockDBHandlerV3{} server := newServerWithMock(t, mockDB) ctx := makeStoreCtx(server) - - makeCloudEvent := func(id, eventType, instanceID string) *cloudevents.Event { - event := cloudevents.NewEvent() - event.SetSpecVersion(cloudevents.VersionV1) - event.SetID(id) - event.SetType(eventType) - event.SetSource("/test") - event.SetTime(time.Now()) - event.SetExtension("namespace", "test-namespace") - event.SetExtension("instanceId", instanceID) - require.NoError(t, event.SetData(cloudevents.ApplicationJSON, map[string]string{"status": "ok"})) - return &event - } + now := time.Now() result := server.processCloudEvents(ctx, []*cloudevents.Event{ - makeCloudEvent("event-1", "pod.ready", "00000000-0000-4000-8000-000000000001.synthetic-instance"), - makeCloudEvent("event-2", "pod.pending", "pod-2"), + 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) @@ -769,6 +771,41 @@ func TestProcessCloudEvents_UsesBulkPersistence(t *testing.T) { ) } +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{} diff --git a/src/control-plane-services/event-ledger/internal/db_client/cassandra/v2.go b/src/control-plane-services/event-ledger/internal/db_client/cassandra/v2.go index 1fedee918..2111061db 100644 --- a/src/control-plane-services/event-ledger/internal/db_client/cassandra/v2.go +++ b/src/control-plane-services/event-ledger/internal/db_client/cassandra/v2.go @@ -1339,23 +1339,16 @@ func (c *CassandraHandler) UpsertFilteredStatsV3(traceCtx context.Context, names return c.upsertStatsRow(traceCtx, filteredStatsV3Table, namespace, eventContext, eventName, timestamp) } -// upsertStatsRow is the shared LWT upsert for stats_v3-shaped tables. +// upsertStatsRow is the shared latest-wins LWT upsert for stats_v3-shaped tables. // table must be a trusted, code-controlled identifier (not user input). func (c *CassandraHandler) upsertStatsRow(traceCtx context.Context, table, namespace, eventContext, eventName string, timestamp time.Time) error { logger := logging.GetLogger(traceCtx) err := c.executeWithSessionRecreation(traceCtx, func() error { - // Use LWT to atomically insert if not exists insertQuery := fmt.Sprintf(`INSERT INTO %s (namespace, context, event_name, timestamp, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?) IF NOT EXISTS`, table) - // Variables to scan existing row if IF NOT EXISTS fails. - // When CAS fails, Cassandra returns all columns in this order: - // PK: namespace, context - // Other (alphabetical): created_at, event_name, timestamp, updated_at - var existingNamespace, existingContext, existingEventName string - var existingCreatedAt, existingTimestamp, existingUpdatedAt time.Time - + previous := make(map[string]any) applied, err := c.session.Query(insertQuery, namespace, eventContext, @@ -1363,15 +1356,7 @@ func (c *CassandraHandler) upsertStatsRow(traceCtx context.Context, table, names timestamp, timestamp, // created_at timestamp, // updated_at - ).WithContext(traceCtx).ScanCAS( - &existingNamespace, - &existingContext, - &existingCreatedAt, - &existingEventName, - &existingTimestamp, - &existingUpdatedAt, - ) - + ).WithContext(traceCtx).MapScanCAS(previous) if err != nil { logger.ErrorContext(traceCtx, "Failed to insert stats", zap.Error(err), @@ -1379,33 +1364,48 @@ func (c *CassandraHandler) upsertStatsRow(traceCtx context.Context, table, names zap.String("namespace", namespace), zap.String("context", eventContext), zap.String("event_name", eventName)) - return err + return fmt.Errorf("failed to insert stats into %s: %w", table, err) } - if !applied { - // Record exists - update it preserving created_at, replacing event_name - updateQuery := fmt.Sprintf(`INSERT INTO %s (namespace, context, event_name, timestamp, created_at, updated_at) - VALUES (?, ?, ?, ?, ?, ?)`, table) + if applied { + return nil + } - if err := c.session.Query(updateQuery, - namespace, - eventContext, - eventName, - timestamp, // event timestamp - existingCreatedAt, // preserve original created_at - time.Now(), // updated_at = current time - ).WithContext(traceCtx).Exec(); err != nil { - logger.ErrorContext(traceCtx, "Failed to update stats", - zap.Error(err), - zap.String("table", table), - zap.String("namespace", namespace), - zap.String("context", eventContext), - zap.String("event_name", eventName)) - return err - } + updateQuery := fmt.Sprintf(`UPDATE %s + SET event_name = ?, timestamp = ?, updated_at = ? + WHERE namespace = ? AND context = ? + IF timestamp < ?`, table) + + previous = make(map[string]any) + applied, err = c.session.Query(updateQuery, + eventName, + timestamp, + time.Now(), + namespace, + eventContext, + timestamp, + ).WithContext(traceCtx).MapScanCAS(previous) + if err != nil { + logger.ErrorContext(traceCtx, "Failed to conditionally update stats", + zap.Error(err), + zap.String("table", table), + zap.String("namespace", namespace), + zap.String("context", eventContext), + zap.String("event_name", eventName)) + return fmt.Errorf("failed to conditionally update stats in %s: %w", table, err) + } + + if !applied { + logger.DebugContext(traceCtx, "Skipped stale stats update", + zap.String("table", table), + zap.String("namespace", namespace), + zap.String("context", eventContext), + zap.String("event_name", eventName), + zap.Time("timestamp", timestamp)) + return nil } - logger.DebugContext(traceCtx, "Upserted stats", + logger.DebugContext(traceCtx, "Updated stats", zap.String("table", table), zap.String("namespace", namespace), zap.String("context", eventContext), diff --git a/src/control-plane-services/event-ledger/internal/db_client/cassandra/v2_test.go b/src/control-plane-services/event-ledger/internal/db_client/cassandra/v2_test.go index 06e762fa7..9dbe00d10 100644 --- a/src/control-plane-services/event-ledger/internal/db_client/cassandra/v2_test.go +++ b/src/control-plane-services/event-ledger/internal/db_client/cassandra/v2_test.go @@ -35,6 +35,41 @@ func TestFilteredStatsTableNameCompatibility(t *testing.T) { assert.Equal(t, "stats_v3_ngc", filteredStatsV3Table) } +func TestUpsertFilteredStatsV3KeepsLatestEvent(t *testing.T) { + session := getTestSession(t) + if session == nil { + t.Skip("Cassandra not available for testing") + } + + handler := &CassandraHandler{session: session} + logger := otelzap.New(zap.NewNop()) + ctx := logging.AttachLoggerToContext(context.Background(), logger) + + _ = session.Query("TRUNCATE " + filteredStatsV3Table).Exec() + t.Cleanup(func() { _ = session.Query("TRUNCATE " + filteredStatsV3Table).Exec() }) + + createdAt := time.Now().Add(-2 * time.Minute).Truncate(time.Millisecond) + latestTimestamp := createdAt.Add(2 * time.Minute) + staleTimestamp := createdAt.Add(time.Minute) + + require.NoError(t, handler.UpsertFilteredStatsV3(ctx, "ns-latest", "ctx-1", "pod.pending", createdAt)) + require.NoError(t, handler.UpsertFilteredStatsV3(ctx, "ns-latest", "ctx-1", "pod.ready", latestTimestamp)) + require.NoError(t, handler.UpsertFilteredStatsV3(ctx, "ns-latest", "ctx-1", "pod.starting", staleTimestamp)) + + var eventName string + var storedTimestamp time.Time + var storedCreatedAt time.Time + err := session.Query( + `SELECT event_name, timestamp, created_at FROM stats_v3_ngc WHERE namespace = ? AND context = ?`, + "ns-latest", + "ctx-1", + ).Scan(&eventName, &storedTimestamp, &storedCreatedAt) + require.NoError(t, err) + assert.Equal(t, "pod.ready", eventName) + assert.Equal(t, latestTimestamp.UTC(), storedTimestamp.UTC().Truncate(time.Millisecond)) + assert.Equal(t, createdAt.UTC(), storedCreatedAt.UTC().Truncate(time.Millisecond)) +} + func TestBulkUpsertEventsV3(t *testing.T) { session := getTestSession(t) if session == nil {