You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airavata.apache.org by ya...@apache.org on 2026/10/09 22:33:01 UTC
(airavata-custos) branch master updated: Decide trace status from deliveries and return them with the trace (#615)
This is an automated email from the ASF dual-hosted git repository.
yasithdev pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
The following commit(s) were added to refs/heads/master by this push:
new c85cf4e67 Decide trace status from deliveries and return them with the trace (#615)
c85cf4e67 is described below
commit c85cf4e67f1af335fc8049a7aed07a6739983772
Author: Lahiru Jayathilake <la...@gmail.com>
AuthorDate: Fri Oct 9 18:32:55 2026 -0400
Decide trace status from deliveries and return them with the trace (#615)
---
CONTRIBUTING.md | 1 +
api/core.openapi.yaml | 40 ++++++
.../ACCESS/AMIE-Processor/handler/posix_alloc.go | 2 +-
.../ACCESS/AMIE-Processor/pkg/amie/loader.go | 5 -
.../testdata/scenarios/baseline.yaml | 14 +-
.../Identity-Provisioner/pkg/comanage/loader.go | 5 -
internal/db/migrations/000007_event_bus.up.sql | 1 +
internal/server/audit_handlers.go | 16 ++-
internal/server/audit_handlers_test.go | 11 +-
internal/store/audit_trace_store.go | 145 ++++++++++++++++++---
.../store/audit_trace_store_integration_test.go | 53 +++++++-
internal/store/audit_trace_store_test.go | 62 ++++++++-
internal/tracing/status.go | 56 +-------
internal/tracing/status_test.go | 23 ----
pkg/models/trace.go | 40 +++++-
15 files changed, 336 insertions(+), 138 deletions(-)
diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md
index ace39008e..75b9c0c5d 100644
--- a/CONTRIBUTING.md
+++ b/CONTRIBUTING.md
@@ -69,6 +69,7 @@ Every SQL file must carry the Apache 2.0 license header (see existing files for
- Database work happens inside `s.inTx(ctx, func(*sql.Tx) error { … })`. Do not nest transactions.
- HTTP routing uses the Go 1.22 `ServeMux` with `r.PathValue` for path parameters.
- Publish domain events through `pkg/events.Bus` from the service layer; never from handlers or stores.
+- An audit event that records a failure must have `Failed`, `Error` or `Rejected` in its event type (for example `ComanageProvisioningFailed`, `PosixUsernameBuildFailed`). The trace view reads a step's outcome from its name, so a failure named otherwise shows as ok.
## Tests
diff --git a/api/core.openapi.yaml b/api/core.openapi.yaml
index e6eb90779..f9c247ca9 100644
--- a/api/core.openapi.yaml
+++ b/api/core.openapi.yaml
@@ -445,6 +445,17 @@ definitions:
user_id:
type: string
type: object
+ DeliveryCounts:
+ properties:
+ attempts:
+ type: integer
+ failed:
+ type: integer
+ pending:
+ type: integer
+ succeeded:
+ type: integer
+ type: object
DeliveryHistory:
properties:
attempts:
@@ -695,6 +706,25 @@ definitions:
subscriber:
type: string
type: object
+ TraceDelivery:
+ properties:
+ attempts:
+ type: integer
+ event_type:
+ type: string
+ id:
+ type: string
+ last_error:
+ type: string
+ next_run_at:
+ type: string
+ span_id:
+ type: string
+ status:
+ $ref: '#/definitions/EventDeliveryStatus'
+ subscriber:
+ type: string
+ type: object
TraceEvent:
properties:
created_at:
@@ -707,6 +737,8 @@ definitions:
type: string
event_type:
type: string
+ id:
+ type: string
parent_span_id:
type: string
source:
@@ -732,6 +764,8 @@ definitions:
type: string
event_type:
type: string
+ id:
+ type: string
parent_span_id:
type: string
source:
@@ -743,6 +777,8 @@ definitions:
type: object
TraceSummary:
properties:
+ deliveries:
+ $ref: '#/definitions/DeliveryCounts'
ended_at:
type: string
event_count:
@@ -1150,6 +1186,10 @@ paths:
description: OK
schema:
properties:
+ deliveries:
+ items:
+ $ref: '#/definitions/TraceDelivery'
+ type: array
status:
type: string
trace_id:
diff --git a/connectors/ACCESS/AMIE-Processor/handler/posix_alloc.go b/connectors/ACCESS/AMIE-Processor/handler/posix_alloc.go
index c42e68926..4355251dc 100644
--- a/connectors/ACCESS/AMIE-Processor/handler/posix_alloc.go
+++ b/connectors/ACCESS/AMIE-Processor/handler/posix_alloc.go
@@ -37,7 +37,7 @@ func allocateAndCreateClusterUser(ctx context.Context, svc *service.Service, clu
base, truncated, err := posix.BuildBase(user, posix.Prefix())
if err != nil {
_, _ = svc.CreateAuditEvent(ctx, &models.AuditEvent{
- EventType: "PosixUsernameUnbuildable",
+ EventType: "PosixUsernameBuildFailed",
EntityID: userID,
EntityType: "user",
Details: err.Error(),
diff --git a/connectors/ACCESS/AMIE-Processor/pkg/amie/loader.go b/connectors/ACCESS/AMIE-Processor/pkg/amie/loader.go
index b49e25bb5..8f42536f7 100644
--- a/connectors/ACCESS/AMIE-Processor/pkg/amie/loader.go
+++ b/connectors/ACCESS/AMIE-Processor/pkg/amie/loader.go
@@ -42,16 +42,11 @@ import (
custosconfig "github.com/apache/airavata-custos/internal/config"
"github.com/apache/airavata-custos/internal/db"
corestore "github.com/apache/airavata-custos/internal/store"
- "github.com/apache/airavata-custos/internal/tracing"
"github.com/apache/airavata-custos/pkg/events"
"github.com/apache/airavata-custos/pkg/identity"
coreservice "github.com/apache/airavata-custos/pkg/service"
)
-func init() {
- tracing.RegisterMarkers("amie", "PACKET_RECEIVED", "TRANSACTION_COMPLETE", "REPLY_SENT", "REPLY_HELD")
-}
-
const connectorName = "amie"
// Type is the connector type in the config file.
diff --git a/connectors/ACCESS/AMIE-Processor/testdata/scenarios/baseline.yaml b/connectors/ACCESS/AMIE-Processor/testdata/scenarios/baseline.yaml
index 04ff9767c..84ebcbe84 100644
--- a/connectors/ACCESS/AMIE-Processor/testdata/scenarios/baseline.yaml
+++ b/connectors/ACCESS/AMIE-Processor/testdata/scenarios/baseline.yaml
@@ -55,13 +55,13 @@ expectations:
packets:
total_count: 12
by_type_status:
- - { type: request_project_create, status: DECODED, count: 3 }
- - { type: data_project_create, status: DECODED, count: 1 }
- - { type: data_account_create, status: DECODED, count: 1 }
- - { type: request_user_modify, status: DECODED, count: 2 }
- - { type: request_person_merge, status: DECODED, count: 1 }
- - { type: request_account_create, status: DECODED, count: 3 }
- - { type: inform_transaction_complete, status: DECODED, count: 1 }
+ - { type: request_project_create, status: PROCESSED, count: 3 }
+ - { type: data_project_create, status: PROCESSED, count: 1 }
+ - { type: data_account_create, status: PROCESSED, count: 1 }
+ - { type: request_user_modify, status: PROCESSED, count: 2 }
+ - { type: request_person_merge, status: PROCESSED, count: 1 }
+ - { type: request_account_create, status: PROCESSED, count: 3 }
+ - { type: inform_transaction_complete, status: PROCESSED, count: 1 }
tables:
diff --git a/connectors/COmanage/Identity-Provisioner/pkg/comanage/loader.go b/connectors/COmanage/Identity-Provisioner/pkg/comanage/loader.go
index 8c290bbd6..c71304878 100644
--- a/connectors/COmanage/Identity-Provisioner/pkg/comanage/loader.go
+++ b/connectors/COmanage/Identity-Provisioner/pkg/comanage/loader.go
@@ -33,16 +33,11 @@ import (
"github.com/apache/airavata-custos/connectors/COmanage/Identity-Provisioner/internal/client"
"github.com/apache/airavata-custos/connectors/COmanage/Identity-Provisioner/internal/subscribers"
"github.com/apache/airavata-custos/internal/config"
- "github.com/apache/airavata-custos/internal/tracing"
"github.com/apache/airavata-custos/pkg/events"
"github.com/apache/airavata-custos/pkg/identity"
"github.com/apache/airavata-custos/pkg/service"
)
-func init() {
- tracing.RegisterMarkers("comanage", "ComanageProvisioningStarted", "ComanageClusterAccountAttached")
-}
-
// Type is the connector type in the config file and the subscriber name on the bus.
const Type = "comanage-identity-provisioner"
diff --git a/internal/db/migrations/000007_event_bus.up.sql b/internal/db/migrations/000007_event_bus.up.sql
index 6e0c79931..d11da17df 100644
--- a/internal/db/migrations/000007_event_bus.up.sql
+++ b/internal/db/migrations/000007_event_bus.up.sql
@@ -43,6 +43,7 @@ CREATE TABLE IF NOT EXISTS events
CREATE INDEX IF NOT EXISTS idx_events_type ON events (event_type, created_at);
CREATE INDEX IF NOT EXISTS idx_events_source ON events (source);
CREATE INDEX IF NOT EXISTS idx_events_time ON events (created_at);
+CREATE INDEX IF NOT EXISTS idx_events_trace ON events (trace_id);
-- One row per event subscription. The worker updates it after each try.
-- Failed tries go to audit_events.
diff --git a/internal/server/audit_handlers.go b/internal/server/audit_handlers.go
index d9c55a655..91bd912af 100644
--- a/internal/server/audit_handlers.go
+++ b/internal/server/audit_handlers.go
@@ -95,7 +95,7 @@ func (s *Server) handleListTraces(w http.ResponseWriter, r *http.Request) {
// @Security BearerAuth
// @Produce json
// @Param trace_id path string true "32-char lowercase hex"
-// @Success 200 {object} object{trace_id=string,status=string,tree=[]models.TraceNode,truncated=boolean}
+// @Success 200 {object} object{trace_id=string,status=string,tree=[]models.TraceNode,deliveries=[]models.TraceDelivery,truncated=boolean}
// @Failure 400 {object} object{error=string} "Malformed trace_id"
// @Failure 404 {object} object{error=string} "Trace not found"
// @Failure 503 {object} object{error=string} "Audit trace store not configured"
@@ -110,7 +110,7 @@ func (s *Server) handleGetTrace(w http.ResponseWriter, r *http.Request) {
common.WriteError(w, http.StatusBadRequest, err)
return
}
- tree, truncated, err := ts.GetTraceTree(r.Context(), traceID)
+ tree, deliveries, truncated, err := ts.GetTraceTree(r.Context(), traceID)
if err != nil {
common.WriteError(w, http.StatusInternalServerError, err)
return
@@ -119,12 +119,16 @@ func (s *Server) handleGetTrace(w http.ResponseWriter, r *http.Request) {
common.WriteError(w, http.StatusNotFound, errors.New("trace not found"))
return
}
+ if deliveries == nil {
+ deliveries = []models.TraceDelivery{}
+ }
resp := map[string]any{
- "trace_id": traceID,
- "status": tree.Status,
- "tree": tree.Children,
- "truncated": truncated,
+ "trace_id": traceID,
+ "status": tree.Status,
+ "tree": tree.Children,
+ "deliveries": deliveries,
+ "truncated": truncated,
}
common.WriteJSON(w, http.StatusOK, resp)
}
diff --git a/internal/server/audit_handlers_test.go b/internal/server/audit_handlers_test.go
index 20bd2fb70..c6678cf95 100644
--- a/internal/server/audit_handlers_test.go
+++ b/internal/server/audit_handlers_test.go
@@ -37,6 +37,7 @@ type fakeAuditStore struct {
traces []models.TraceSummary
total int
tree *models.TraceNode
+ deliveries []models.TraceDelivery
truncated bool
events []models.TraceEvent
sources []string
@@ -58,12 +59,12 @@ func (f *fakeAuditStore) ListTraces(_ context.Context, filter store.TraceFilter)
return f.traces, f.total, nil
}
-func (f *fakeAuditStore) GetTraceTree(_ context.Context, traceID string) (*models.TraceNode, bool, error) {
+func (f *fakeAuditStore) GetTraceTree(_ context.Context, traceID string) (*models.TraceNode, []models.TraceDelivery, bool, error) {
f.getTraceArg = traceID
if f.getErr != nil {
- return nil, false, f.getErr
+ return nil, nil, false, f.getErr
}
- return f.tree, f.truncated, nil
+ return f.tree, f.deliveries, f.truncated, nil
}
func (f *fakeAuditStore) ListEvents(_ context.Context, traceID, spanID string) ([]models.TraceEvent, error) {
@@ -292,6 +293,10 @@ func TestGetTraceReturnsTree(t *testing.T) {
if body["status"] != "error" {
t.Errorf("trace status = %v", body["status"])
}
+ // A trace with no deliveries still lists them as an empty array, not null.
+ if deliveries, ok := body["deliveries"].([]any); !ok || len(deliveries) != 0 {
+ t.Errorf("deliveries = %v, want []", body["deliveries"])
+ }
treeNodes := body["tree"].([]any)
if len(treeNodes) != 1 {
t.Fatalf("tree len = %d", len(treeNodes))
diff --git a/internal/store/audit_trace_store.go b/internal/store/audit_trace_store.go
index d2cc09f66..16276fb12 100644
--- a/internal/store/audit_trace_store.go
+++ b/internal/store/audit_trace_store.go
@@ -20,6 +20,7 @@ package store
import (
"context"
"fmt"
+ "maps"
"slices"
"strings"
"time"
@@ -47,7 +48,8 @@ type TraceFilter struct {
type AuditTraceStore interface {
ListTraces(ctx context.Context, filter TraceFilter) ([]models.TraceSummary, int, error)
- GetTraceTree(ctx context.Context, traceID string) (*models.TraceNode, bool, error)
+ // GetTraceTree returns the trace's rows as a tree, its deliveries, and whether the tree was cut at TreeRowLimit.
+ GetTraceTree(ctx context.Context, traceID string) (*models.TraceNode, []models.TraceDelivery, bool, error)
ListEvents(ctx context.Context, traceID, spanID string) ([]models.TraceEvent, error)
ListSources(ctx context.Context) ([]string, error)
}
@@ -59,6 +61,7 @@ type AuditTraceStore interface {
// only by the connector-specific endpoints.
const auditTraceSelect = `
SELECT
+ id,
trace_id,
span_id,
parent_span_id,
@@ -80,6 +83,7 @@ func NewAuditTraceStore(db *sqlx.DB) AuditTraceStore {
}
type rowEvent struct {
+ ID string `db:"id"`
TraceID string `db:"trace_id"`
SpanID string `db:"span_id"`
ParentSpanID *string `db:"parent_span_id"`
@@ -93,6 +97,7 @@ type rowEvent struct {
func (r rowEvent) toTraceEvent() models.TraceEvent {
return models.TraceEvent{
+ ID: r.ID,
SpanID: r.SpanID,
ParentSpanID: r.ParentSpanID,
Source: r.Source,
@@ -105,9 +110,8 @@ func (r rowEvent) toTraceEvent() models.TraceEvent {
}
}
-// ListTraces pages traces in SQL and summarises the page in one more query.
-// Status is computed in Go, so a status filter first loads every matching
-// trace's rows in one query and narrows the trace set before counting and paging.
+// ListTraces pages traces in SQL, then loads the page's rows and deliveries in one query each.
+// Status is computed in Go, so a status filter first loads every matching trace and keeps the ones with that status before counting and paging.
func (s *pgAuditTraceStore) ListTraces(ctx context.Context, f TraceFilter) ([]models.TraceSummary, int, error) {
if len(f.Statuses) > 0 {
whereSQL, args := buildTraceWhere(f)
@@ -115,9 +119,13 @@ func (s *pgAuditTraceStore) ListTraces(ctx context.Context, f TraceFilter) ([]mo
if err != nil {
return nil, 0, err
}
+ deliveries, err := s.traceDeliveries(ctx, slices.Collect(maps.Keys(byTrace)))
+ if err != nil {
+ return nil, 0, err
+ }
f.traceIDs = []string{}
for id, rows := range byTrace {
- if _, _, status := summarise(rows); slices.Contains(f.Statuses, status) {
+ if slices.Contains(f.Statuses, traceStatus(rows, deliveries[id])) {
f.traceIDs = append(f.traceIDs, id)
}
}
@@ -168,25 +176,117 @@ func (s *pgAuditTraceStore) ListTraces(ctx context.Context, f TraceFilter) ([]mo
if err != nil {
return nil, 0, err
}
+ deliveries, err := s.traceDeliveries(ctx, ids)
+ if err != nil {
+ return nil, 0, err
+ }
out := make([]models.TraceSummary, 0, len(raw))
for _, r := range raw {
- rootOp, src, status := summarise(byTrace[r.TraceID])
+ rootOp, src := summarise(byTrace[r.TraceID])
out = append(out, models.TraceSummary{
TraceID: r.TraceID,
RootOperation: rootOp,
Source: src,
- Status: status,
+ Status: traceStatus(byTrace[r.TraceID], deliveries[r.TraceID]),
StartedAt: r.StartedAt,
EndedAt: r.EndedAt,
EventCount: r.EventCount,
+ Deliveries: deliveryCounts(deliveries[r.TraceID]),
})
}
return out, total, nil
}
+// traceDeliveries loads the deliveries of the events each trace published.
+// An event's deliveries are created in one transaction and get the same created_at,
+// so they are also sorted by connector name to keep a fixed order.
+func (s *pgAuditTraceStore) traceDeliveries(ctx context.Context, traceIDs []string) (map[string][]models.TraceDelivery, error) {
+ var rows []models.TraceDelivery
+ err := s.db.SelectContext(ctx, &rows, s.db.Rebind(`
+ SELECT d.id, e.trace_id, e.event_type, d.subscriber, d.status, d.attempts, d.next_run_at, d.last_error, e.span_id
+ FROM event_deliveries d JOIN events e ON e.id = d.event_id
+ WHERE e.trace_id = ANY(?)
+ ORDER BY d.created_at ASC, d.subscriber ASC`), traceIDs)
+ if err != nil {
+ return nil, fmt.Errorf("audit_trace_store: deliveries: %w", err)
+ }
+ byTrace := map[string][]models.TraceDelivery{}
+ for _, d := range rows {
+ byTrace[d.TraceID] = append(byTrace[d.TraceID], d)
+ }
+ return byTrace, nil
+}
+
+// deliveryEntityType is the row entity type the event worker writes for each event delivery try.
+const deliveryEntityType = "event_delivery"
+
+// traceStatus is error when a delivery ran out of tries or a step failed
+// outside any delivery, in_progress while a delivery is pending, else ok.
+// Failed steps inside a delivery try are ignored, the delivery's status covers them.
+func traceStatus(rows []rowEvent, deliveries []models.TraceDelivery) string {
+ pending := false
+ for _, d := range deliveries {
+ switch d.Status {
+ case models.EventDeliveryFailed:
+ return tracing.StatusError
+ case models.EventDeliveryPending:
+ pending = true
+ }
+ }
+ firstRowOfSpan := make(map[string]rowEvent, len(rows))
+ for _, r := range rows {
+ if _, ok := firstRowOfSpan[r.SpanID]; !ok {
+ firstRowOfSpan[r.SpanID] = r
+ }
+ }
+ for _, r := range rows {
+ if tracing.EventStatus(r.EventType) == tracing.StatusError && !insideDelivery(r, firstRowOfSpan) {
+ return tracing.StatusError
+ }
+ }
+ if pending {
+ return tracing.StatusInProgress
+ }
+ return tracing.StatusOk
+}
+
+// insideDelivery reports whether the row was written during a delivery try, by walking up the parent spans to a delivery row.
+func insideDelivery(r rowEvent, firstRowOfSpan map[string]rowEvent) bool {
+ for range len(firstRowOfSpan) + 1 {
+ if r.EntityType == deliveryEntityType {
+ return true
+ }
+ if r.ParentSpanID == nil {
+ return false
+ }
+ parent, ok := firstRowOfSpan[*r.ParentSpanID]
+ if !ok {
+ return false
+ }
+ r = parent
+ }
+ return false
+}
+
+func deliveryCounts(deliveries []models.TraceDelivery) models.DeliveryCounts {
+ var c models.DeliveryCounts
+ for _, d := range deliveries {
+ switch d.Status {
+ case models.EventDeliveryPending:
+ c.Pending++
+ c.Attempts = max(c.Attempts, d.Attempts)
+ case models.EventDeliverySucceeded:
+ c.Succeeded++
+ case models.EventDeliveryFailed:
+ c.Failed++
+ }
+ }
+ return c
+}
+
// traceRows loads the rows of every trace matching where, oldest first, keyed by trace.
func (s *pgAuditTraceStore) traceRows(ctx context.Context, where string, args ...any) (map[string][]rowEvent, error) {
- q := `SELECT trace_id, span_id, parent_span_id, source, event_type, entity_type, entity_id, description, created_at
+ q := `SELECT id, trace_id, span_id, parent_span_id, source, event_type, entity_type, entity_id, description, created_at
FROM (` + auditTraceSelect + `) u
WHERE ` + where + `
ORDER BY created_at ASC, span_id ASC`
@@ -202,12 +302,10 @@ func (s *pgAuditTraceStore) traceRows(ctx context.Context, where string, args ..
}
// summarise names a trace by its root row, the first whose parent span wrote no row.
-func summarise(rows []rowEvent) (rootOp, source, status string) {
+func summarise(rows []rowEvent) (rootOp, source string) {
spans := make(map[string]bool, len(rows))
- statuses := make([]tracing.TraceEventStatus, len(rows))
- for i, r := range rows {
+ for _, r := range rows {
spans[r.SpanID] = true
- statuses[i] = tracing.TraceEventStatus{Source: r.Source, EventType: r.EventType}
}
root := rows[0]
for _, r := range rows {
@@ -216,20 +314,23 @@ func summarise(rows []rowEvent) (rootOp, source, status string) {
break
}
}
- return root.EventType, root.Source, tracing.TraceStatus(statuses)
+ return root.EventType, root.Source
}
-func (s *pgAuditTraceStore) GetTraceTree(ctx context.Context, traceID string) (*models.TraceNode, bool, error) {
+func (s *pgAuditTraceStore) GetTraceTree(ctx context.Context, traceID string) (*models.TraceNode, []models.TraceDelivery, bool, error) {
byTrace, err := s.traceRows(ctx, `trace_id = ?`, traceID)
rows := byTrace[traceID]
if err != nil || len(rows) == 0 {
- return nil, false, err
+ return nil, nil, false, err
+ }
+ deliveries, err := s.traceDeliveries(ctx, []string{traceID})
+ if err != nil {
+ return nil, nil, false, err
}
- _, _, status := summarise(rows)
truncated := len(rows) > TreeRowLimit
tree := buildTree(rows[:min(len(rows), TreeRowLimit)])
- tree.Status = status
- return tree, truncated, nil
+ tree.Status = traceStatus(rows, deliveries[traceID])
+ return tree, deliveries[traceID], truncated, nil
}
// buildTree makes one node per row and hangs children off their parent span's
@@ -258,7 +359,7 @@ func buildTree(rows []rowEvent) *models.TraceNode {
}
func (s *pgAuditTraceStore) ListEvents(ctx context.Context, traceID, spanID string) ([]models.TraceEvent, error) {
- q := `SELECT trace_id, span_id, parent_span_id, source, event_type, entity_type, entity_id, description, created_at
+ q := `SELECT id, trace_id, span_id, parent_span_id, source, event_type, entity_type, entity_id, description, created_at
FROM (` + auditTraceSelect + `) u
WHERE trace_id = ?`
args := []any{traceID}
@@ -303,9 +404,11 @@ func buildTraceWhere(f TraceFilter) (string, []any) {
clauses = append(clauses, "u.created_at <= ?")
args = append(args, f.To)
}
+ // An admin looking into a problem usually searches by a username or an entity id.
+ // Usernames are only in the details text, and the caller always sends a time window, so that scan stays small.
if f.Q != "" {
- clauses = append(clauses, "(u.trace_id ILIKE ? OR u.event_type ILIKE ?)")
- args = append(args, strings.ToLower(f.Q)+"%", "%"+f.Q+"%")
+ clauses = append(clauses, "(u.trace_id ILIKE ? OR u.event_type ILIKE ? OR u.entity_id = ? OR u.description ILIKE ?)")
+ args = append(args, strings.ToLower(f.Q)+"%", "%"+f.Q+"%", f.Q, "%"+f.Q+"%")
}
if f.traceIDs != nil {
diff --git a/internal/store/audit_trace_store_integration_test.go b/internal/store/audit_trace_store_integration_test.go
index 9263c34ff..f1e3fe9cf 100644
--- a/internal/store/audit_trace_store_integration_test.go
+++ b/internal/store/audit_trace_store_integration_test.go
@@ -20,6 +20,7 @@
package store
import (
+ "database/sql"
"fmt"
"testing"
"time"
@@ -57,7 +58,7 @@ func TestListTracesStatusFilterCountsAndPages(t *testing.T) {
t.Fatalf("total=%d page=%+v, want total 3 and the oldest error trace", total, got)
}
- tree, _, err := s.GetTraceTree(ctx, got[0].TraceID)
+ tree, _, _, err := s.GetTraceTree(ctx, got[0].TraceID)
if err != nil {
t.Fatalf("tree: %v", err)
}
@@ -86,7 +87,7 @@ func TestTraceTreeKeepsEveryRowOfASpan(t *testing.T) {
t.Fatalf("commit: %v", err)
}
- tree, _, err := NewAuditTraceStore(database).GetTraceTree(ctx, traceID)
+ tree, _, _, err := NewAuditTraceStore(database).GetTraceTree(ctx, traceID)
if err != nil {
t.Fatalf("tree: %v", err)
}
@@ -94,3 +95,51 @@ func TestTraceTreeKeepsEveryRowOfASpan(t *testing.T) {
t.Fatalf("tree children=%+v, want the sourced row then the core one", tree.Children)
}
}
+
+// Make sure the list and the detail include a trace's deliveries, and that search finds the trace by an entity id or a username.
+func TestTraceReportsItsDeliveriesAndIsFoundByEntity(t *testing.T) {
+ database, ctx := setupTestDB(t), t.Context()
+ const source, traceID, userID = "trace-delivery-test", "00000000000000000000000000000c00", "user-c00"
+ if _, err := database.Exec(`DELETE FROM audit_events WHERE source = $1`, source); err != nil {
+ t.Fatalf("clean: %v", err)
+ }
+ if _, err := database.Exec(`INSERT INTO audit_events
+ (id, event_type, event_time, entity_id, entity_type, details, source, trace_id, span_id, parent_span_id)
+ VALUES ($1, 'CLUSTER_ACCOUNT_APPROVED', $2, $3, 'compute_cluster_user', '{"local_username":"jdoe"}', $4, $5, $6, NULL)`,
+ traceID+"-row", time.Date(2026, 1, 2, 0, 0, 0, 0, time.UTC), userID, source, traceID, traceID[:16]); err != nil {
+ t.Fatalf("insert audit row: %v", err)
+ }
+ event := newEvent(time.Now())
+ event.TraceID, event.SpanID = traceID, traceID[:16]
+ inTx(t, database, func(tx *sql.Tx) error {
+ if err := NewEventDeliveryStore(database).CreateEvent(ctx, tx, event); err != nil {
+ return err
+ }
+ pending := newDelivery(event.ID, "slurm-association-mapper", time.Now().Add(time.Minute))
+ pending.Attempts = 4
+ if err := NewEventDeliveryStore(database).CreateDelivery(ctx, tx, pending); err != nil {
+ return err
+ }
+ return NewEventDeliveryStore(database).CreateDelivery(ctx, tx, newDelivery(event.ID, "email-notifier", time.Now()))
+ })
+
+ s := NewAuditTraceStore(database)
+ for _, q := range []string{userID, "jdoe"} {
+ got, _, err := s.ListTraces(ctx, TraceFilter{Sources: []string{source}, Q: q})
+ if err != nil {
+ t.Fatalf("list q=%s: %v", q, err)
+ }
+ if len(got) != 1 || got[0].Status != "in_progress" || got[0].Deliveries != (models.DeliveryCounts{Pending: 2, Attempts: 4}) {
+ t.Fatalf("q=%s: traces=%+v, want one in-progress trace with two pending deliveries at try 4", q, got)
+ }
+ }
+
+ _, deliveries, _, err := s.GetTraceTree(ctx, traceID)
+ if err != nil {
+ t.Fatalf("tree: %v", err)
+ }
+ if len(deliveries) != 2 || deliveries[0].Subscriber != "email-notifier" || deliveries[1].Subscriber != "slurm-association-mapper" ||
+ deliveries[1].Attempts != 4 || deliveries[0].SpanID != traceID[:16] || deliveries[0].EventType != event.EventType {
+ t.Fatalf("deliveries=%+v, want both, by subscriber, under the publisher's span", deliveries)
+ }
+}
diff --git a/internal/store/audit_trace_store_test.go b/internal/store/audit_trace_store_test.go
index adf35fd71..1f6c32116 100644
--- a/internal/store/audit_trace_store_test.go
+++ b/internal/store/audit_trace_store_test.go
@@ -21,6 +21,9 @@ import (
"strings"
"testing"
"time"
+
+ "github.com/apache/airavata-custos/internal/tracing"
+ "github.com/apache/airavata-custos/pkg/models"
)
func mkSpan(b byte) string {
@@ -110,12 +113,59 @@ func TestBuildTraceWhereTimeAndQ(t *testing.T) {
if !strings.Contains(w, "u.created_at >= ?") || !strings.Contains(w, "u.created_at <= ?") {
t.Errorf("where missing time clauses: %q", w)
}
- if !strings.Contains(w, "u.trace_id ILIKE ?") {
- t.Errorf("where missing trace_id ILIKE: %q", w)
+ if !strings.Contains(w, "u.trace_id ILIKE ?") || !strings.Contains(w, "u.entity_id = ?") || !strings.Contains(w, "u.description ILIKE ?") {
+ t.Errorf("where missing a search clause: %q", w)
+ }
+ // 2 timestamps + 4 search values
+ if len(args) != 6 {
+ t.Errorf("args len = %d, want 6", len(args))
+ }
+}
+
+// Make sure the trace status comes from the deliveries, so a delivery that succeeded after failed tries reads ok, not error.
+func TestTraceStatusComesFromDeliveries(t *testing.T) {
+ publisher, try1, handler1, try2 := mkSpan(0x01), mkSpan(0x02), mkSpan(0x03), mkSpan(0x04)
+ rows := []rowEvent{
+ {SpanID: publisher, EventType: "CLUSTER_ACCOUNT_APPROVED"},
+ {SpanID: handler1, ParentSpanID: &try1, EventType: "ComanageProvisioningFailed"},
+ {SpanID: try1, ParentSpanID: &publisher, EventType: "EVENT_DELIVERY_FAILED", EntityType: deliveryEntityType},
+ {SpanID: try2, ParentSpanID: &publisher, EventType: "EVENT_DELIVERY_SUCCEEDED", EntityType: deliveryEntityType},
+ }
+ outside := append(rows, rowEvent{SpanID: mkSpan(0x05), ParentSpanID: &publisher, EventType: "PosixUsernameBuildFailed"})
+ delivery := func(status models.EventDeliveryStatus) []models.TraceDelivery {
+ return []models.TraceDelivery{{Status: status, Attempts: 2}}
+ }
+ cases := []struct {
+ name string
+ rows []rowEvent
+ deliveries []models.TraceDelivery
+ want string
+ }{
+ {"delivery out of tries", rows, delivery(models.EventDeliveryFailed), tracing.StatusError},
+ {"delivery still retrying", rows, delivery(models.EventDeliveryPending), tracing.StatusInProgress},
+ {"delivery recovered on a later try", rows, delivery(models.EventDeliverySucceeded), tracing.StatusOk},
+ {"step failed outside any delivery", outside, delivery(models.EventDeliverySucceeded), tracing.StatusError},
+ {"nothing delivered, nothing failed", rows[:1], nil, tracing.StatusOk},
+ }
+ for _, c := range cases {
+ if got := traceStatus(c.rows, c.deliveries); got != c.want {
+ t.Errorf("%s: status = %q, want %q", c.name, got, c.want)
+ }
}
- // 2 timestamps + 2 LIKE values
- if len(args) != 4 {
- t.Errorf("args len = %d, want 4", len(args))
+}
+
+// Make sure the list gets how many deliveries are pending, done and failed,
+// and the most tries a pending one has made, which it shows as "retrying N of 10".
+func TestDeliveryCounts(t *testing.T) {
+ got := deliveryCounts([]models.TraceDelivery{
+ {Status: models.EventDeliveryPending, Attempts: 4},
+ {Status: models.EventDeliveryPending, Attempts: 1},
+ {Status: models.EventDeliverySucceeded, Attempts: 1},
+ {Status: models.EventDeliveryFailed, Attempts: 10},
+ })
+ want := models.DeliveryCounts{Pending: 2, Succeeded: 1, Failed: 1, Attempts: 4}
+ if got != want {
+ t.Errorf("counts = %+v, want %+v", got, want)
}
}
@@ -127,7 +177,7 @@ func TestSummarisePicksRootWithUnauditedParent(t *testing.T) {
{SpanID: received, ParentSpanID: &ingest, Source: "amie", EventType: "PACKET_RECEIVED"},
{SpanID: mkSpan(0x03), ParentSpanID: &received, Source: "comanage", EventType: "ComanageLookup"},
}
- if op, src, _ := summarise(rows); op != "PACKET_RECEIVED" || src != "amie" {
+ if op, src := summarise(rows); op != "PACKET_RECEIVED" || src != "amie" {
t.Errorf("root = %s/%s, want PACKET_RECEIVED/amie", op, src)
}
}
diff --git a/internal/tracing/status.go b/internal/tracing/status.go
index c6f4acf63..e375c276a 100644
--- a/internal/tracing/status.go
+++ b/internal/tracing/status.go
@@ -17,11 +17,7 @@
package tracing
-import (
- "slices"
- "strings"
- "sync"
-)
+import "strings"
const (
StatusOk = "ok"
@@ -31,6 +27,9 @@ const (
var errorMarkers = []string{"failed", "error", "rejected"}
+// EventStatus reads a step's outcome from its name, since a step records no
+// status. The trace's status is decided in the audit trace store from its
+// steps and its deliveries.
func EventStatus(eventType string) string {
lower := strings.ToLower(eventType)
for _, marker := range errorMarkers {
@@ -40,50 +39,3 @@ func EventStatus(eventType string) string {
}
return StatusOk
}
-
-type traceMarkers struct {
- start string
- terminals []string
-}
-
-var (
- markersMu sync.RWMutex
- markers = map[string]traceMarkers{}
-)
-
-// RegisterMarkers declares the event that opens a source's work in a trace and
-// the events that close it. Connectors call this at boot so the core stays
-// unaware of connector-specific event names.
-func RegisterMarkers(source, start string, terminals ...string) {
- markersMu.Lock()
- defer markersMu.Unlock()
- markers[source] = traceMarkers{start, terminals}
-}
-
-type TraceEventStatus struct {
- Source string
- EventType string
-}
-
-// TraceStatus is "error" if any event errored, "in_progress" while a source has
-// written its start marker and none of its terminals, else "ok". A source with
-// no registered markers never holds a trace in progress.
-func TraceStatus(events []TraceEventStatus) string {
- markersMu.RLock()
- defer markersMu.RUnlock()
- started, done := map[string]bool{}, map[string]bool{}
- for _, e := range events {
- if EventStatus(e.EventType) == StatusError {
- return StatusError
- }
- m, ok := markers[e.Source]
- started[e.Source] = started[e.Source] || ok && e.EventType == m.start
- done[e.Source] = done[e.Source] || slices.Contains(m.terminals, e.EventType)
- }
- for s := range started {
- if started[s] && !done[s] {
- return StatusInProgress
- }
- }
- return StatusOk
-}
diff --git a/internal/tracing/status_test.go b/internal/tracing/status_test.go
index a03ce470a..d43262f27 100644
--- a/internal/tracing/status_test.go
+++ b/internal/tracing/status_test.go
@@ -35,26 +35,3 @@ func TestEventStatus(t *testing.T) {
}
}
}
-
-func TestTraceStatus(t *testing.T) {
- RegisterMarkers("status-test", "PACKET_RECEIVED", "TRANSACTION_COMPLETE", "REPLY_HELD")
- cases := []struct {
- name string
- events []string
- want string
- }{
- {"start without terminal", []string{"PACKET_RECEIVED", "CREATE_PERSON"}, StatusInProgress},
- {"no start", []string{"CREATE_PERSON"}, StatusOk},
- {"start and terminal", []string{"PACKET_RECEIVED", "REPLY_HELD"}, StatusOk},
- {"error wins", []string{"PACKET_RECEIVED", "REQUEST_REJECTED", "TRANSACTION_COMPLETE"}, StatusError},
- }
- for _, c := range cases {
- events := make([]TraceEventStatus, len(c.events))
- for i, et := range c.events {
- events[i] = TraceEventStatus{Source: "status-test", EventType: et}
- }
- if got := TraceStatus(events); got != c.want {
- t.Errorf("%s: TraceStatus = %q, want %q", c.name, got, c.want)
- }
- }
-}
diff --git a/pkg/models/trace.go b/pkg/models/trace.go
index e724a59bb..f5254ba39 100644
--- a/pkg/models/trace.go
+++ b/pkg/models/trace.go
@@ -23,16 +23,42 @@ import (
)
type TraceSummary struct {
- TraceID string `db:"trace_id" json:"trace_id"`
- RootOperation string `db:"root_operation" json:"root_operation"`
- Source string `db:"source" json:"source"`
- Status string `db:"status" json:"status"`
- StartedAt time.Time `db:"started_at" json:"started_at"`
- EndedAt time.Time `db:"ended_at" json:"ended_at"`
- EventCount int `db:"event_count" json:"event_count"`
+ TraceID string `db:"trace_id" json:"trace_id"`
+ RootOperation string `db:"root_operation" json:"root_operation"`
+ Source string `db:"source" json:"source"`
+ Status string `db:"status" json:"status"`
+ StartedAt time.Time `db:"started_at" json:"started_at"`
+ EndedAt time.Time `db:"ended_at" json:"ended_at"`
+ EventCount int `db:"event_count" json:"event_count"`
+ Deliveries DeliveryCounts `db:"-" json:"deliveries"`
+}
+
+// DeliveryCounts says how many of a trace's deliveries are pending, done and
+// failed. Attempts is the most tries any pending one has made.
+type DeliveryCounts struct {
+ Pending int `json:"pending"`
+ Succeeded int `json:"succeeded"`
+ Failed int `json:"failed"`
+ Attempts int `json:"attempts"`
+}
+
+// TraceDelivery is a connector's delivery of an event the trace published.
+// SpanID is the span of the step that published the event, so the delivery
+// can be shown under that step.
+type TraceDelivery struct {
+ ID string `db:"id" json:"id"`
+ TraceID string `db:"trace_id" json:"-"`
+ EventType string `db:"event_type" json:"event_type"`
+ Subscriber string `db:"subscriber" json:"subscriber"`
+ Status EventDeliveryStatus `db:"status" json:"status"`
+ Attempts int `db:"attempts" json:"attempts"`
+ NextRunAt time.Time `db:"next_run_at" json:"next_run_at"`
+ LastError *string `db:"last_error" json:"last_error,omitempty"`
+ SpanID string `db:"span_id" json:"span_id"`
}
type TraceEvent struct {
+ ID string `db:"id" json:"id"`
SpanID string `db:"span_id" json:"span_id"`
ParentSpanID *string `db:"parent_span_id" json:"parent_span_id,omitempty"`
Source string `db:"source" json:"source"`