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/12 04:21:14 UTC
(airavata-custos) branch fix-trace-list created (now fcc87bfb8)
This is an automated email from the ASF dual-hosted git repository.
yasithdev pushed a change to branch fix-trace-list
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
at fcc87bfb8 simplify audit trace read queries and sourcelist
This branch includes the following new commits:
new fcc87bfb8 simplify audit trace read queries and sourcelist
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(airavata-custos) 01/01: simplify audit trace read queries and sourcelist
Posted by ya...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yasithdev pushed a commit to branch fix-trace-list
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git
commit fcc87bfb8540ebdfc59a342eaf078eb6c219b677
Author: Yasith Jayawardana <ya...@icloud.com>
AuthorDate: Mon Oct 12 00:21:05 2026 -0400
simplify audit trace read queries and sourcelist
---
internal/store/audit_trace_sources_test.go | 42 +++++++++++++++
internal/store/audit_trace_store.go | 62 +++++++---------------
.../store/audit_trace_store_integration_test.go | 20 +++++++
internal/store/audit_trace_store_test.go | 6 +--
4 files changed, 85 insertions(+), 45 deletions(-)
diff --git a/internal/store/audit_trace_sources_test.go b/internal/store/audit_trace_sources_test.go
new file mode 100644
index 000000000..14134c99c
--- /dev/null
+++ b/internal/store/audit_trace_sources_test.go
@@ -0,0 +1,42 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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 store_test
+
+import (
+ "slices"
+ "testing"
+
+ "github.com/apache/airavata-custos/connectors/ACCESS/AMIE-Processor/pkg/amie"
+ "github.com/apache/airavata-custos/connectors/COmanage/Identity-Provisioner/pkg/comanage"
+ "github.com/apache/airavata-custos/connectors/Notification/Email/pkg/email"
+ "github.com/apache/airavata-custos/connectors/SLURM/Association-Mapper/pkg/smapper"
+ "github.com/apache/airavata-custos/connectors/Storage/VAST-Provisioner/pkg/vast"
+ "github.com/apache/airavata-custos/internal/store"
+)
+
+func TestSourcesOfferEverySubscriber(t *testing.T) {
+ sources, err := store.NewAuditTraceStore(nil).ListSources(t.Context())
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, subscriber := range []string{amie.Type, comanage.Type, email.Type, smapper.Type, vast.Type} {
+ if !slices.Contains(sources, subscriber) {
+ t.Errorf("missing %q", subscriber)
+ }
+ }
+}
diff --git a/internal/store/audit_trace_store.go b/internal/store/audit_trace_store.go
index 16276fb12..e34269ba2 100644
--- a/internal/store/audit_trace_store.go
+++ b/internal/store/audit_trace_store.go
@@ -54,25 +54,8 @@ type AuditTraceStore interface {
ListSources(ctx context.Context) ([]string, error)
}
-// auditTraceSelect projects audit_events into the trace-view shape. Every
-// subsystem (core and each connector) writes its audit rows here, tagged by
-// `source`. Connector-specific references (e.g. AMIE's packet_id / event_id
-// on amie_audit_extras) live on connector-owned tables and are joined
-// only by the connector-specific endpoints.
-const auditTraceSelect = `
-SELECT
- id,
- trace_id,
- span_id,
- parent_span_id,
- source,
- event_type,
- entity_type,
- entity_id,
- details AS description,
- event_time AS created_at
-FROM audit_events
-`
+const traceEventColumns = `id, trace_id, span_id, parent_span_id, source, event_type, entity_type, entity_id,
+ details AS description, event_time AS created_at`
type pgAuditTraceStore struct {
db *sqlx.DB
@@ -115,7 +98,7 @@ func (r rowEvent) toTraceEvent() models.TraceEvent {
func (s *pgAuditTraceStore) ListTraces(ctx context.Context, f TraceFilter) ([]models.TraceSummary, int, error) {
if len(f.Statuses) > 0 {
whereSQL, args := buildTraceWhere(f)
- byTrace, err := s.traceRows(ctx, `trace_id IN (SELECT trace_id FROM (`+auditTraceSelect+`) u `+whereSQL+`)`, args...)
+ byTrace, err := s.traceRows(ctx, `trace_id IN (SELECT trace_id FROM audit_events `+whereSQL+`)`, args...)
if err != nil {
return nil, 0, err
}
@@ -137,22 +120,20 @@ func (s *pgAuditTraceStore) ListTraces(ctx context.Context, f TraceFilter) ([]mo
limit = 50
}
- countQuery := `SELECT COUNT(*) FROM (
- SELECT trace_id FROM (` + auditTraceSelect + `) u ` + whereSQL + `
- GROUP BY trace_id
- ) g`
+ countQuery := `SELECT COUNT(DISTINCT trace_id) FROM audit_events ` + whereSQL
var total int
if err := s.db.GetContext(ctx, &total, s.db.Rebind(countQuery), args...); err != nil {
return nil, 0, fmt.Errorf("audit_trace_store: count: %w", err)
}
listQuery := `SELECT trace_id,
- MIN(created_at) AS started_at,
- MAX(created_at) AS ended_at,
+ MIN(event_time) AS started_at,
+ MAX(event_time) AS ended_at,
COUNT(*) AS event_count
- FROM (` + auditTraceSelect + `) u ` + whereSQL + `
+ FROM audit_events
+ WHERE trace_id IN (SELECT trace_id FROM audit_events ` + whereSQL + `)
GROUP BY trace_id
- ORDER BY MAX(created_at) DESC
+ ORDER BY MAX(event_time) DESC
LIMIT ? OFFSET ?`
listArgs := append([]any{}, args...)
listArgs = append(listArgs, limit, f.Offset)
@@ -286,10 +267,7 @@ func deliveryCounts(deliveries []models.TraceDelivery) models.DeliveryCounts {
// 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 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`
+ q := `SELECT ` + traceEventColumns + ` FROM audit_events WHERE ` + where + ` ORDER BY event_time ASC, span_id ASC`
var rows []rowEvent
if err := s.db.SelectContext(ctx, &rows, s.db.Rebind(q), args...); err != nil {
return nil, fmt.Errorf("audit_trace_store: rows: %w", err)
@@ -359,15 +337,13 @@ func buildTree(rows []rowEvent) *models.TraceNode {
}
func (s *pgAuditTraceStore) ListEvents(ctx context.Context, traceID, spanID string) ([]models.TraceEvent, error) {
- 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 = ?`
+ q := `SELECT ` + traceEventColumns + ` FROM audit_events WHERE trace_id = ?`
args := []any{traceID}
if spanID != "" {
q += ` AND span_id = ?`
args = append(args, spanID)
}
- q += ` ORDER BY created_at ASC, span_id ASC`
+ q += ` ORDER BY event_time ASC, span_id ASC`
var rows []rowEvent
if err := s.db.SelectContext(ctx, &rows, s.db.Rebind(q), args...); err != nil {
@@ -380,8 +356,10 @@ func (s *pgAuditTraceStore) ListEvents(ctx context.Context, traceID, spanID stri
return out, nil
}
+// Subscribers are the connectors' Type constants, which this package cannot import without a cycle.
func (s *pgAuditTraceStore) ListSources(_ context.Context) ([]string, error) {
- return []string{"amie", "comanage", "core", "slurm"}, nil
+ return []string{"amie", "amie-processor", "comanage", "comanage-identity-provisioner", "core",
+ "email-notifier", "slurm", "slurm-association-mapper", "vast-storage-provisioner"}, nil
}
func buildTraceWhere(f TraceFilter) (string, []any) {
@@ -394,25 +372,25 @@ func buildTraceWhere(f TraceFilter) (string, []any) {
placeholders[i] = "?"
args = append(args, src)
}
- clauses = append(clauses, "u.source IN ("+strings.Join(placeholders, ",")+")")
+ clauses = append(clauses, "source IN ("+strings.Join(placeholders, ",")+")")
}
if !f.From.IsZero() {
- clauses = append(clauses, "u.created_at >= ?")
+ clauses = append(clauses, "event_time >= ?")
args = append(args, f.From)
}
if !f.To.IsZero() {
- clauses = append(clauses, "u.created_at <= ?")
+ clauses = append(clauses, "event_time <= ?")
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 ? OR u.entity_id = ? OR u.description ILIKE ?)")
+ clauses = append(clauses, "(trace_id ILIKE ? OR event_type ILIKE ? OR entity_id = ? OR details ILIKE ?)")
args = append(args, strings.ToLower(f.Q)+"%", "%"+f.Q+"%", f.Q, "%"+f.Q+"%")
}
if f.traceIDs != nil {
- clauses = append(clauses, "u.trace_id = ANY(?)")
+ clauses = append(clauses, "trace_id = ANY(?)")
args = append(args, f.traceIDs)
}
diff --git a/internal/store/audit_trace_store_integration_test.go b/internal/store/audit_trace_store_integration_test.go
index f1e3fe9cf..613a02a87 100644
--- a/internal/store/audit_trace_store_integration_test.go
+++ b/internal/store/audit_trace_store_integration_test.go
@@ -143,3 +143,23 @@ func TestTraceReportsItsDeliveriesAndIsFoundByEntity(t *testing.T) {
t.Fatalf("deliveries=%+v, want both, by subscriber, under the publisher's span", deliveries)
}
}
+
+func TestListTracesFilterKeepsWholeTraceFigures(t *testing.T) {
+ database := setupTestDB(t)
+ const traceID = "00000000000000000000000000000d00"
+ if _, err := database.Exec(`DELETE FROM audit_events WHERE trace_id = $1`, traceID); err != nil {
+ t.Fatalf("clean: %v", err)
+ }
+ for i, src := range []string{"trace-figures-test", "other", "other"} {
+ if _, err := database.Exec(`INSERT INTO audit_events
+ (id, event_type, event_time, entity_id, details, source, trace_id, span_id, parent_span_id)
+ VALUES ($1, 'ROOT', NOW(), 'e', '', $2, $3, $4, NULL)`,
+ fmt.Sprint(traceID, i), src, traceID, traceID[:16]); err != nil {
+ t.Fatalf("insert: %v", err)
+ }
+ }
+ got, _, err := NewAuditTraceStore(database).ListTraces(t.Context(), TraceFilter{Sources: []string{"trace-figures-test"}})
+ if err != nil || len(got) != 1 || got[0].EventCount != 3 {
+ t.Fatalf("traces=%+v err=%v, want one trace of 3 rows", got, err)
+ }
+}
diff --git a/internal/store/audit_trace_store_test.go b/internal/store/audit_trace_store_test.go
index 1f6c32116..196a9a54f 100644
--- a/internal/store/audit_trace_store_test.go
+++ b/internal/store/audit_trace_store_test.go
@@ -98,7 +98,7 @@ func TestBuildTraceWhereEmpty(t *testing.T) {
func TestBuildTraceWhereSources(t *testing.T) {
w, args := buildTraceWhere(TraceFilter{Sources: []string{"amie", "comanage"}})
- if !strings.Contains(w, "u.source IN (?,?)") {
+ if !strings.Contains(w, "source IN (?,?)") {
t.Errorf("where = %q, missing IN clause", w)
}
if len(args) != 2 {
@@ -110,10 +110,10 @@ func TestBuildTraceWhereTimeAndQ(t *testing.T) {
from := time.Date(2026, 6, 1, 0, 0, 0, 0, time.UTC)
to := from.Add(24 * time.Hour)
w, args := buildTraceWhere(TraceFilter{From: from, To: to, Q: "create"})
- if !strings.Contains(w, "u.created_at >= ?") || !strings.Contains(w, "u.created_at <= ?") {
+ if !strings.Contains(w, "event_time >= ?") || !strings.Contains(w, "event_time <= ?") {
t.Errorf("where missing time clauses: %q", w)
}
- if !strings.Contains(w, "u.trace_id ILIKE ?") || !strings.Contains(w, "u.entity_id = ?") || !strings.Contains(w, "u.description ILIKE ?") {
+ if !strings.Contains(w, "trace_id ILIKE ?") || !strings.Contains(w, "entity_id = ?") || !strings.Contains(w, "details ILIKE ?") {
t.Errorf("where missing a search clause: %q", w)
}
// 2 timestamps + 4 search values