diff --git a/src/control-plane-services/event-ledger/README.md b/src/control-plane-services/event-ledger/README.md index 4bcdc512b1..5a9d777364 100644 --- a/src/control-plane-services/event-ledger/README.md +++ b/src/control-plane-services/event-ledger/README.md @@ -138,6 +138,24 @@ curl 'http://localhost:8080/v3/ledger/namespace/example/stats?eventFilter=ready, The optional `view=filtered` query selects the filtered stats view when it is enabled in configuration. The endpoint returns `404` when that view is disabled. +### Read account-scoped events and stats + +Two routes read the events and stats recorded for one account and function: + +```bash +curl 'http://localhost:8080/v3/ledger/accounts/acct-1/functions//versions//events?instance_id=instance-1' +curl 'http://localhost:8080/v3/ledger/accounts/acct-1/functions//versions//stats' +``` + +The function version ID is the namespace. The routes require the `admin:read` +scope and trust the account in the path; they do not check who owns the +function version. Malformed IDs return `400`, and a request that matches +nothing returns `200` with empty lists. The events route takes the same query +parameters as the namespace events route, and the stats route takes +`eventFilter`. The response shapes have no `nca_id` field, and event details +are returned as stored. Only events stored with `nca_id` and `function_id` can +match, which needs `v3.account-columns-enabled`. + ## Authentication Authentication is enabled by default. For the JWT provider, configure diff --git a/src/control-plane-services/event-ledger/cmd/api/service/BUILD.bazel b/src/control-plane-services/event-ledger/cmd/api/service/BUILD.bazel index 765b2c9277..ff996d0c16 100644 --- a/src/control-plane-services/event-ledger/cmd/api/service/BUILD.bazel +++ b/src/control-plane-services/event-ledger/cmd/api/service/BUILD.bazel @@ -10,6 +10,7 @@ go_library( "v1.go", "v2.go", "v3.go", + "v3_account_reads.go", ], importpath = "github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/cmd/api/service", visibility = ["//visibility:public"], @@ -52,6 +53,7 @@ go_test( "service_test.go", "servicev2_test.go", "v1_test.go", + "v3_account_reads_test.go", "v3_test.go", ], embed = [":service"], diff --git a/src/control-plane-services/event-ledger/cmd/api/service/servicev2_test.go b/src/control-plane-services/event-ledger/cmd/api/service/servicev2_test.go index 8198f2c228..45e3c6f159 100644 --- a/src/control-plane-services/event-ledger/cmd/api/service/servicev2_test.go +++ b/src/control-plane-services/event-ledger/cmd/api/service/servicev2_test.go @@ -128,6 +128,14 @@ func (p *passDBHandlerV2) GetEventsV3(ctx context.Context, namespace, eventConte return []data_access.EventV3Record{}, nil } +func (p *passDBHandlerV2) GetAccountEventsV3(ctx context.Context, namespace, eventContext, ncaID, functionID string) ([]data_access.EventV3Record, error) { + return []data_access.EventV3Record{}, nil +} + +func (p *passDBHandlerV2) GetAccountStatsV3(ctx context.Context, namespace, ncaID, functionID string) ([]data_access.StatsV3Record, error) { + return []data_access.StatsV3Record{}, nil +} + func (p *passDBHandlerV2) Close() error { return nil } // Interface assertion for V2 @@ -202,6 +210,14 @@ func (p *failDBHandlerV2) GetEventsV3(ctx context.Context, namespace, eventConte return nil, fmt.Errorf("v2 get events fail") } +func (p *failDBHandlerV2) GetAccountEventsV3(ctx context.Context, namespace, eventContext, ncaID, functionID string) ([]data_access.EventV3Record, error) { + return nil, fmt.Errorf("v2 get account events fail") +} + +func (p *failDBHandlerV2) GetAccountStatsV3(ctx context.Context, namespace, ncaID, functionID string) ([]data_access.StatsV3Record, error) { + return nil, fmt.Errorf("v2 get account stats fail") +} + func (p *failDBHandlerV2) Close() error { return fmt.Errorf("v2 close fail") } // Add Close method // Interface assertion for V2 diff --git a/src/control-plane-services/event-ledger/cmd/api/service/v3_account_reads.go b/src/control-plane-services/event-ledger/cmd/api/service/v3_account_reads.go new file mode 100644 index 0000000000..6ee0c1da78 --- /dev/null +++ b/src/control-plane-services/event-ledger/cmd/api/service/v3_account_reads.go @@ -0,0 +1,224 @@ +/* +SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package service + +import ( + "encoding/json" + "fmt" + "net/http" + "sort" + "strings" + + "github.com/gorilla/mux" + "go.uber.org/zap" + + "github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/observability/logging" +) + +// accountPathParams reads and validates the account, function, and function +// version from the route. It writes a 400 problem detail and returns false when +// any of them is malformed. +func accountPathParams(w http.ResponseWriter, r *http.Request) (ncaID, functionID, functionVersionID string, ok bool) { + vars := mux.Vars(r) + ncaID = vars["ncaId"] + functionID = vars["functionId"] + functionVersionID = vars["functionVersionId"] + + switch { + case len(ncaID) > maxNcaIDLength || !ncaIDFieldPattern.MatchString(ncaID): + sendProblemDetail(w, http.StatusBadRequest, "Bad Request", + fmt.Sprintf("invalid ncaId: must match %s and be at most %d characters", ncaIDFieldPattern, maxNcaIDLength)) + case !uuidFieldPattern.MatchString(functionID): + sendProblemDetail(w, http.StatusBadRequest, "Bad Request", "invalid functionId: must be a UUID") + case !uuidFieldPattern.MatchString(functionVersionID): + sendProblemDetail(w, http.StatusBadRequest, "Bad Request", "invalid functionVersionId: must be a UUID") + default: + return ncaID, functionID, functionVersionID, true + } + return "", "", "", false +} + +// GetAccountEventsV3 handles GET requests to +// /v3/ledger/accounts/{ncaId}/functions/{functionId}/versions/{functionVersionId}/events +// It returns the events of one context in the function version namespace that +// were recorded for the given account and function. The path account is +// trusted: the caller is a service that has already authenticated the user. +// A request that matches nothing returns 200 with an empty list, so a caller +// cannot tell whether a function exists or who owns it. +// It accepts the same query parameters as GetEventsV3. +func (s *Server) GetAccountEventsV3(w http.ResponseWriter, r *http.Request) { + traceCtx := r.Context() + logger := logging.GetLogger(traceCtx) + + ncaID, functionID, namespace, ok := accountPathParams(w, r) + if !ok { + return + } + + queryParams := r.URL.Query() + contextV3 := ContextV3{ + InstanceID: queryParams.Get(contextFieldInstanceID), + DeploymentID: queryParams.Get(contextFieldDeploymentID), + GPUSpecificationID: queryParams.Get(contextFieldGPUSpecificationID), + ClusterID: queryParams.Get(contextFieldClusterID), + ResourceID: queryParams.Get(contextFieldResourceID), + } + + attributeKey := queryParams.Get(queryParamAttributeKey) + attributeValue := queryParams.Get(queryParamAttributeValue) + if (attributeKey == "") != (attributeValue == "") { + sendProblemDetail(w, http.StatusBadRequest, "Bad Request", + fmt.Sprintf("%s and %s must be provided together", queryParamAttributeKey, queryParamAttributeValue)) + return + } + + eventContext, err := eventContextToCanonical(contextV3) + if err != nil { + sendProblemDetail(w, http.StatusBadRequest, "Bad Request", + fmt.Sprintf("Invalid context parameters: %s", err.Error())) + return + } + + records, err := s.conns.DbHandlerV2.GetAccountEventsV3(traceCtx, namespace, eventContext, ncaID, functionID) + if err != nil { + logger.ErrorContext(traceCtx, "Failed to retrieve account events", + zap.Error(err), + zap.String("namespace", namespace), + zap.String("context", eventContext), + zap.String("nca_id", ncaID), + zap.String("function_id", functionID)) + sendProblemDetail(w, http.StatusInternalServerError, "Internal Server Error", + "Failed to retrieve events from database") + return + } + + // Cassandra cannot ORDER BY timestamp since it is not a clustering key. + sort.Slice(records, func(i, j int) bool { + return records[i].Timestamp.After(records[j].Timestamp) + }) + + response := EventsV3Response{ + Namespace: namespace, + Context: eventContext, + Events: make([]EventV3Item, 0, len(records)), + } + + for _, record := range records { + var details EventDetails + if len(record.Details) > 0 { + if err := json.Unmarshal(record.Details, &details); err != nil { + logger.WarnContext(traceCtx, "Failed to unmarshal event details, using empty details", + zap.Error(err), + zap.String("event_name", record.EventName), + zap.Time("timestamp", record.Timestamp)) + details = EventDetails{} + } + } + + if !attributeMatches(details, attributeKey, attributeValue) { + continue + } + + response.Events = append(response.Events, EventV3Item{ + EventName: record.EventName, + Source: record.Source, + Timestamp: record.Timestamp, + Details: details, + CreatedAt: record.CreatedAt, + UpdatedAt: record.UpdatedAt, + }) + } + + logger.InfoContext(traceCtx, "Successfully retrieved account events", + zap.String("namespace", namespace), + zap.String("nca_id", ncaID), + zap.String("function_id", functionID), + zap.Int("event_count", len(response.Events))) + + sendJSONResponse(w, http.StatusOK, response) +} + +// GetAccountStatsV3 handles GET requests to +// /v3/ledger/accounts/{ncaId}/functions/{functionId}/versions/{functionVersionId}/stats +// It returns stats for the function version namespace built only from rows +// recorded for the given account and function, so the summary counts only +// those rows. The optional eventFilter query parameter works as in +// GetStatsV3. A request that matches nothing returns 200 with empty data. +func (s *Server) GetAccountStatsV3(w http.ResponseWriter, r *http.Request) { + traceCtx := r.Context() + logger := logging.GetLogger(traceCtx) + + ncaID, functionID, namespace, ok := accountPathParams(w, r) + if !ok { + return + } + + eventFilter := r.URL.Query().Get("eventFilter") + var filterEvents map[string]struct{} + if eventFilter != "" { + filterEvents = make(map[string]struct{}) + for _, event := range strings.Split(eventFilter, ",") { + filterEvents[strings.TrimSpace(event)] = struct{}{} + } + } + + records, err := s.conns.DbHandlerV2.GetAccountStatsV3(traceCtx, namespace, ncaID, functionID) + if err != nil { + logger.ErrorContext(traceCtx, "Failed to retrieve account stats", + zap.Error(err), + zap.String("namespace", namespace), + zap.String("nca_id", ncaID), + zap.String("function_id", functionID)) + sendProblemDetail(w, http.StatusInternalServerError, "Internal Server Error", + "Failed to retrieve stats from database") + return + } + + response := StatsV3Response{ + Namespace: namespace, + Summary: StatsSummary{ + ByEvent: make(map[string]int), + }, + Contexts: []ContextStats{}, + } + + for _, record := range records { + if filterEvents != nil { + if _, ok := filterEvents[record.EventName]; !ok { + continue + } + } + + response.Contexts = append(response.Contexts, ContextStats{ + Context: record.Context, + EventName: record.EventName, + Timestamp: record.Timestamp, + FirstSeen: record.CreatedAt, + }) + response.Summary.ByEvent[record.EventName]++ + } + response.Summary.TotalContexts = len(response.Contexts) + + logger.InfoContext(traceCtx, "Successfully retrieved account stats", + zap.String("namespace", namespace), + zap.String("nca_id", ncaID), + zap.String("function_id", functionID), + zap.Int("total_contexts", response.Summary.TotalContexts)) + + sendJSONResponse(w, http.StatusOK, response) +} diff --git a/src/control-plane-services/event-ledger/cmd/api/service/v3_account_reads_test.go b/src/control-plane-services/event-ledger/cmd/api/service/v3_account_reads_test.go new file mode 100644 index 0000000000..6d7b91dd4f --- /dev/null +++ b/src/control-plane-services/event-ledger/cmd/api/service/v3_account_reads_test.go @@ -0,0 +1,265 @@ +/* +SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package service + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/gorilla/mux" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/config" + "github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/data_access" + "github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/observability/logging" + "github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/pkg/testutils" +) + +const ( + accountTestNca = "acct-1" + accountTestFunction = "11111111-1111-1111-1111-111111111111" + accountTestVersion = "22222222-2222-2222-2222-222222222222" +) + +type accountRow struct { + namespace string + context string + ncaID string + functionID string +} + +// accountReadsDB emulates the CQL filter: it returns a row only when every +// column matches, so rows with empty columns (legacy rows) never match. +type accountReadsDB struct { + *mockDBHandlerV3 + eventRows map[accountRow][]data_access.EventV3Record + statsRows map[accountRow][]data_access.StatsV3Record + err error + lastEvents accountRow + lastStats accountRow +} + +func (d *accountReadsDB) GetAccountEventsV3(_ context.Context, namespace, eventContext, ncaID, functionID string) ([]data_access.EventV3Record, error) { + d.lastEvents = accountRow{namespace, eventContext, ncaID, functionID} + if d.err != nil { + return nil, d.err + } + return d.eventRows[d.lastEvents], nil +} + +func (d *accountReadsDB) GetAccountStatsV3(_ context.Context, namespace, ncaID, functionID string) ([]data_access.StatsV3Record, error) { + d.lastStats = accountRow{namespace: namespace, ncaID: ncaID, functionID: functionID} + if d.err != nil { + return nil, d.err + } + return d.statsRows[d.lastStats], nil +} + +func newAccountReadsDB(t *testing.T) *accountReadsDB { + t.Helper() + now := time.Now() + details := func(attrs map[string]any) json.RawMessage { + raw, err := json.Marshal(EventDetails{Body: "body", Attributes: attrs}) + require.NoError(t, err) + return raw + } + match := accountRow{accountTestVersion, "instance_id=pod-1", accountTestNca, accountTestFunction} + statsMatch := accountRow{namespace: accountTestVersion, ncaID: accountTestNca, functionID: accountTestFunction} + return &accountReadsDB{ + mockDBHandlerV3: &mockDBHandlerV3{}, + eventRows: map[accountRow][]data_access.EventV3Record{ + match: { + {EventName: "pending", Source: "src", Details: details(map[string]any{"request": "r1"}), Timestamp: now.Add(-time.Minute)}, + {EventName: "ready", Source: "src", Details: details(map[string]any{"request": "r2"}), Timestamp: now}, + }, + }, + statsRows: map[accountRow][]data_access.StatsV3Record{ + statsMatch: { + {Context: "pod-1", EventName: "pending", Timestamp: now}, + {Context: "pod-2", EventName: "ready", Timestamp: now}, + {Context: "pod-3", EventName: "ready", Timestamp: now}, + }, + }, + } +} + +func serveAccountRead(t *testing.T, db data_access.DBHandlerV2, route, query string, vars map[string]string) *httptest.ResponseRecorder { + t.Helper() + logger := testutils.InitTestLogger(t) + server := NewServer(Connections{DbHandlerV2: db}, logger, nil, "test", &config.HTTPClientConfig{}, + config.PaginationConfig{}, config.StatsConfig{}) + + req := httptest.NewRequest(http.MethodGet, "/"+route+query, nil) + ctx := context.WithValue(req.Context(), logging.LoggerKey, logging.NewTraceLogger(req.Context(), logger)) + req = mux.SetURLVars(req.WithContext(ctx), vars) + + w := httptest.NewRecorder() + switch route { + case "events": + server.GetAccountEventsV3(w, req) + case "stats": + server.GetAccountStatsV3(w, req) + default: + t.Fatalf("unknown route %q", route) + } + return w +} + +func accountVars(nca, function, version string) map[string]string { + return map[string]string{"ncaId": nca, "functionId": function, "functionVersionId": version} +} + +func TestGetAccountEventsV3(t *testing.T) { + const podQuery = "?instance_id=pod-1" + + tests := []struct { + name string + query string + vars map[string]string + wantStatus int + wantEvents []string + }{ + {"match returns rows newest first", podQuery, accountVars(accountTestNca, accountTestFunction, accountTestVersion), http.StatusOK, []string{"ready", "pending"}}, + {"wrong account returns empty 200", podQuery, accountVars("acct-2", accountTestFunction, accountTestVersion), http.StatusOK, nil}, + {"wrong function returns empty 200", podQuery, accountVars(accountTestNca, "33333333-3333-3333-3333-333333333333", accountTestVersion), http.StatusOK, nil}, + {"wrong version returns empty 200", podQuery, accountVars(accountTestNca, accountTestFunction, "33333333-3333-3333-3333-333333333333"), http.StatusOK, nil}, + {"other context returns empty 200", "?instance_id=pod-9", accountVars(accountTestNca, accountTestFunction, accountTestVersion), http.StatusOK, nil}, + {"attribute filter keeps matching event", podQuery + "&attribute_key=request&attribute_value=r2", accountVars(accountTestNca, accountTestFunction, accountTestVersion), http.StatusOK, []string{"ready"}}, + {"attribute filter with no match returns empty 200", podQuery + "&attribute_key=request&attribute_value=nope", accountVars(accountTestNca, accountTestFunction, accountTestVersion), http.StatusOK, nil}, + {"attribute key without value", podQuery + "&attribute_key=request", accountVars(accountTestNca, accountTestFunction, accountTestVersion), http.StatusBadRequest, nil}, + {"invalid context field", "?instance_id=bad%20value", accountVars(accountTestNca, accountTestFunction, accountTestVersion), http.StatusBadRequest, nil}, + {"malformed ncaId", podQuery, accountVars("acct.1", accountTestFunction, accountTestVersion), http.StatusBadRequest, nil}, + {"empty ncaId", podQuery, accountVars("", accountTestFunction, accountTestVersion), http.StatusBadRequest, nil}, + {"overlong ncaId", podQuery, accountVars(strings.Repeat("a", maxNcaIDLength+1), accountTestFunction, accountTestVersion), http.StatusBadRequest, nil}, + {"malformed functionId", podQuery, accountVars(accountTestNca, "not-a-uuid", accountTestVersion), http.StatusBadRequest, nil}, + {"malformed functionVersionId", podQuery, accountVars(accountTestNca, accountTestFunction, "not-a-uuid"), http.StatusBadRequest, nil}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w := serveAccountRead(t, newAccountReadsDB(t), "events", tc.query, tc.vars) + require.Equal(t, tc.wantStatus, w.Code, w.Body.String()) + if tc.wantStatus != http.StatusOK { + assert.Equal(t, "application/problem+json", w.Header().Get("Content-Type")) + return + } + + var response EventsV3Response + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &response)) + assert.Equal(t, tc.vars["functionVersionId"], response.Namespace) + require.NotNil(t, response.Events) + var got []string + for _, event := range response.Events { + got = append(got, event.EventName) + } + assert.Equal(t, tc.wantEvents, got) + }) + } +} + +func TestGetAccountEventsV3_PassesPathValuesToDB(t *testing.T) { + db := newAccountReadsDB(t) + w := serveAccountRead(t, db, "events", "?instance_id=pod-1", + accountVars(accountTestNca, accountTestFunction, accountTestVersion)) + require.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, accountRow{accountTestVersion, "instance_id=pod-1", accountTestNca, accountTestFunction}, db.lastEvents) +} + +func TestGetAccountEventsV3_ResponseOmitsNcaID(t *testing.T) { + w := serveAccountRead(t, newAccountReadsDB(t), "events", "?instance_id=pod-1", + accountVars(accountTestNca, accountTestFunction, accountTestVersion)) + require.Equal(t, http.StatusOK, w.Code) + assert.NotContains(t, w.Body.String(), "nca_id") + assert.NotContains(t, w.Body.String(), accountTestNca) +} + +func TestGetAccountEventsV3_DBErrorReturns500(t *testing.T) { + db := newAccountReadsDB(t) + db.err = fmt.Errorf("cassandra unavailable") + w := serveAccountRead(t, db, "events", "?instance_id=pod-1", + accountVars(accountTestNca, accountTestFunction, accountTestVersion)) + assert.Equal(t, http.StatusInternalServerError, w.Code) + assert.Equal(t, "application/problem+json", w.Header().Get("Content-Type")) + assert.NotContains(t, w.Body.String(), "cassandra unavailable") +} + +func TestGetAccountStatsV3(t *testing.T) { + otherFunction := "33333333-3333-3333-3333-333333333333" + + tests := []struct { + name string + query string + vars map[string]string + wantStatus int + wantTotal int + wantByEvent map[string]int + wantContextsN int + }{ + {"match counts only returned rows", "", accountVars(accountTestNca, accountTestFunction, accountTestVersion), http.StatusOK, 3, map[string]int{"pending": 1, "ready": 2}, 3}, + {"event filter narrows contexts and summary", "?eventFilter=ready", accountVars(accountTestNca, accountTestFunction, accountTestVersion), http.StatusOK, 2, map[string]int{"ready": 2}, 2}, + {"event filter trims names", "?eventFilter=pending,%20ready", accountVars(accountTestNca, accountTestFunction, accountTestVersion), http.StatusOK, 3, map[string]int{"pending": 1, "ready": 2}, 3}, + {"view parameter is ignored", "?view=bogus", accountVars(accountTestNca, accountTestFunction, accountTestVersion), http.StatusOK, 3, map[string]int{"pending": 1, "ready": 2}, 3}, + {"wrong account returns empty 200", "", accountVars("acct-2", accountTestFunction, accountTestVersion), http.StatusOK, 0, map[string]int{}, 0}, + {"wrong function returns empty 200", "", accountVars(accountTestNca, otherFunction, accountTestVersion), http.StatusOK, 0, map[string]int{}, 0}, + {"malformed ncaId", "", accountVars("acct 1", accountTestFunction, accountTestVersion), http.StatusBadRequest, 0, nil, 0}, + {"malformed functionId", "", accountVars(accountTestNca, "nope", accountTestVersion), http.StatusBadRequest, 0, nil, 0}, + {"malformed functionVersionId", "", accountVars(accountTestNca, accountTestFunction, "nope"), http.StatusBadRequest, 0, nil, 0}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + w := serveAccountRead(t, newAccountReadsDB(t), "stats", tc.query, tc.vars) + require.Equal(t, tc.wantStatus, w.Code, w.Body.String()) + if tc.wantStatus != http.StatusOK { + assert.Equal(t, "application/problem+json", w.Header().Get("Content-Type")) + return + } + + var response StatsV3Response + require.NoError(t, json.Unmarshal(w.Body.Bytes(), &response)) + assert.Equal(t, accountTestVersion, response.Namespace) + require.NotNil(t, response.Contexts) + require.NotNil(t, response.Summary.ByEvent) + assert.Len(t, response.Contexts, tc.wantContextsN) + assert.Equal(t, tc.wantTotal, response.Summary.TotalContexts) + assert.Equal(t, tc.wantByEvent, response.Summary.ByEvent) + }) + } +} + +func TestGetAccountStatsV3_PassesPathValuesToDB(t *testing.T) { + db := newAccountReadsDB(t) + w := serveAccountRead(t, db, "stats", "", accountVars(accountTestNca, accountTestFunction, accountTestVersion)) + require.Equal(t, http.StatusOK, w.Code) + assert.Equal(t, accountRow{namespace: accountTestVersion, ncaID: accountTestNca, functionID: accountTestFunction}, db.lastStats) +} + +func TestGetAccountStatsV3_DBErrorReturns500(t *testing.T) { + db := newAccountReadsDB(t) + db.err = fmt.Errorf("cassandra unavailable") + w := serveAccountRead(t, db, "stats", "", accountVars(accountTestNca, accountTestFunction, accountTestVersion)) + assert.Equal(t, http.StatusInternalServerError, w.Code) + assert.Equal(t, "application/problem+json", w.Header().Get("Content-Type")) +} 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 725c7b4445..069a488821 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 @@ -221,6 +221,14 @@ func (m *mockDBHandlerV3) ReadDeploymentDeploymentStats(context.Context, uuid.UU func (m *mockDBHandlerV3) ListDeploymentInstancesPaginated(ctx context.Context, functionVersionId, deploymentId uuid.UUID, paginationParams data_access.PaginationParams) (data_access.PaginatedInstancesResponse, error) { return data_access.PaginatedInstancesResponse{}, nil } +func (m *mockDBHandlerV3) GetAccountEventsV3(ctx context.Context, namespace, eventContext, ncaID, functionID string) ([]data_access.EventV3Record, error) { + return []data_access.EventV3Record{}, nil +} + +func (m *mockDBHandlerV3) GetAccountStatsV3(ctx context.Context, namespace, ncaID, functionID string) ([]data_access.StatsV3Record, error) { + return []data_access.StatsV3Record{}, nil +} + func (m *mockDBHandlerV3) Close() error { return nil } diff --git a/src/control-plane-services/event-ledger/cmd/api/startup/run_service.go b/src/control-plane-services/event-ledger/cmd/api/startup/run_service.go index 34488fb840..0dccca437a 100644 --- a/src/control-plane-services/event-ledger/cmd/api/startup/run_service.go +++ b/src/control-plane-services/event-ledger/cmd/api/startup/run_service.go @@ -484,6 +484,14 @@ func runService(cfg config.Config) error { middleware.MaybeRequireScopes(logger, requireLocalScopeCheck, middleware.ReadScopes, middleware.RequireAnyScopes)(http.HandlerFunc(server.GetEventsV3)), ).Methods("GET", "OPTIONS") + // Account-scoped V3 reads for KAS. The account in the path is trusted. + authRouter.Handle("/v3/ledger/accounts/{ncaId}/functions/{functionId}/versions/{functionVersionId}/events", + middleware.MaybeRequireScopes(logger, requireLocalScopeCheck, middleware.AdminReadScopes, middleware.RequireAnyScopes)(http.HandlerFunc(server.GetAccountEventsV3)), + ).Methods("GET", "OPTIONS") + authRouter.Handle("/v3/ledger/accounts/{ncaId}/functions/{functionId}/versions/{functionVersionId}/stats", + middleware.MaybeRequireScopes(logger, requireLocalScopeCheck, middleware.AdminReadScopes, middleware.RequireAnyScopes)(http.HandlerFunc(server.GetAccountStatsV3)), + ).Methods("GET", "OPTIONS") + // Cross Account Endpoints for KAS (V2) // Remove when legacy clients migrate to V3. if !cfg.DeprecateEndpoints { diff --git a/src/control-plane-services/event-ledger/internal/data_access/db.go b/src/control-plane-services/event-ledger/internal/data_access/db.go index 29e3cb59b5..daf1e45912 100644 --- a/src/control-plane-services/event-ledger/internal/data_access/db.go +++ b/src/control-plane-services/event-ledger/internal/data_access/db.go @@ -102,6 +102,8 @@ type DBHandlerV2 interface { GetStatsV3(traceCtx context.Context, namespace string) ([]StatsV3Record, error) GetFilteredStatsV3(traceCtx context.Context, namespace string) ([]StatsV3Record, error) GetEventsV3(traceCtx context.Context, namespace, eventContext string) ([]EventV3Record, error) + GetAccountEventsV3(traceCtx context.Context, namespace, eventContext, ncaID, functionID string) ([]EventV3Record, error) + GetAccountStatsV3(traceCtx context.Context, namespace, ncaID, functionID string) ([]StatsV3Record, error) Close() error // Include Close in V2 as well, assuming the underlying connection needs closing. } diff --git a/src/control-plane-services/event-ledger/internal/db_client/cassandra/BUILD.bazel b/src/control-plane-services/event-ledger/internal/db_client/cassandra/BUILD.bazel index 585fb5bcbb..ddb0641140 100644 --- a/src/control-plane-services/event-ledger/internal/db_client/cassandra/BUILD.bazel +++ b/src/control-plane-services/event-ledger/internal/db_client/cassandra/BUILD.bazel @@ -6,6 +6,7 @@ go_library( "common.go", "v1.go", "v2.go", + "v3_account_reads.go", ], importpath = "github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/db_client/cassandra", visibility = ["//:__subpackages__"], diff --git a/src/control-plane-services/event-ledger/internal/db_client/cassandra/v3_account_reads.go b/src/control-plane-services/event-ledger/internal/db_client/cassandra/v3_account_reads.go new file mode 100644 index 0000000000..58f98e7671 --- /dev/null +++ b/src/control-plane-services/event-ledger/internal/db_client/cassandra/v3_account_reads.go @@ -0,0 +1,98 @@ +/* +SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +SPDX-License-Identifier: Apache-2.0 + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package cassandra + +import ( + "context" + "fmt" + "time" + + "github.com/NVIDIA/nvcf/src/control-plane-services/event-ledger/internal/data_access" +) + +// The account reads filter inside the partition being read. Rows written +// without nca_id or function_id hold NULL there, so they never match. +const ( + accountEventsV3Query = `SELECT event_name, source, details, timestamp, created_at, updated_at + FROM events_v3 + WHERE namespace = ? AND context = ? AND nca_id = ? AND function_id = ? + ALLOW FILTERING` + + accountStatsV3Query = `SELECT context, event_name, timestamp, created_at, updated_at + FROM stats_v3 + WHERE namespace = ? AND nca_id = ? AND function_id = ? + ALLOW FILTERING` +) + +// GetAccountEventsV3 retrieves the events of a namespace and context that were +// recorded for the given account and function. +func (c *CassandraHandler) GetAccountEventsV3(traceCtx context.Context, namespace, eventContext, ncaID, functionID string) ([]data_access.EventV3Record, error) { + var records []data_access.EventV3Record + + err := c.executeWithSessionRecreation(traceCtx, func() error { + records = nil + + iter := c.session.Query(accountEventsV3Query, namespace, eventContext, ncaID, functionID).WithContext(traceCtx).Iter() + + var record data_access.EventV3Record + for iter.Scan(&record.EventName, &record.Source, &record.Details, &record.Timestamp, &record.CreatedAt, &record.UpdatedAt) { + records = append(records, record) + record = data_access.EventV3Record{} + } + + return iter.Close() + }, "getAccountEventsV3") + if err != nil { + return nil, fmt.Errorf("failed to get account events from events_v3: %w", err) + } + + return records, nil +} + +// GetAccountStatsV3 retrieves the stats_v3 rows of a namespace that were +// recorded for the given account and function. +func (c *CassandraHandler) GetAccountStatsV3(traceCtx context.Context, namespace, ncaID, functionID string) ([]data_access.StatsV3Record, error) { + var records []data_access.StatsV3Record + + err := c.executeWithSessionRecreation(traceCtx, func() error { + records = nil + + iter := c.session.Query(accountStatsV3Query, namespace, ncaID, functionID).WithContext(traceCtx).Iter() + + var ( + statsContext, eventName string + timestamp, createdAt, updatedAt time.Time + ) + for iter.Scan(&statsContext, &eventName, ×tamp, &createdAt, &updatedAt) { + records = append(records, data_access.StatsV3Record{ + Context: statsContext, + EventName: eventName, + Timestamp: timestamp, + CreatedAt: createdAt, + UpdatedAt: updatedAt, + }) + } + + return iter.Close() + }, "getAccountStatsV3") + if err != nil { + return nil, fmt.Errorf("failed to get account stats from stats_v3: %w", err) + } + + return records, nil +}