From 6511de764e395a1a84c683993a82abe11672eeb2 Mon Sep 17 00:00:00 2001 From: Moshe Immerman Date: Thu, 16 Jul 2026 10:31:28 +0300 Subject: [PATCH 01/12] fix(db): Bound migration lock waits and retry contention Prevent migration DDL and security reconciliation from waiting indefinitely on locks. Retry transient lock contention so live readers no longer cause migration runs to hang or fail unnecessarily. --- migrate/README.md | 13 +++++++++++++ migrate/migrate.go | 27 +++++++++++++++++++++++++-- migrate/migrate_test.go | 29 +++++++++++++++++++++++++++++ migrate/scripts.go | 32 ++++++++++++++++++++++---------- migrate/security.go | 6 ++++++ migrate/view_deps.go | 25 ++++++++++++++++++++++++- 6 files changed, 119 insertions(+), 13 deletions(-) diff --git a/migrate/README.md b/migrate/README.md index 769b89b..b56f49c 100644 --- a/migrate/README.md +++ b/migrate/README.md @@ -19,6 +19,19 @@ comment header: topologically ordered and changes rerun all transitive dependents. - scripts are transactional by default; the SQL and its SHA-256 migration log commit atomically. +- `runs: always` re-runs a script on **every** apply, including a plain process + boot against an unchanged database. It is an **escape hatch**, not the norm. + Normal scripts are hash-gated: they run once and re-run only when their content + changes. Views are restored automatically when Atlas reshapes a base table they + depend on — `migrate.Apply` drops the dependent view and re-runs the script + that creates it — so a view file never needs `runs: always` to stay current. + Reserving `always` for the rare script that must reconcile on every apply keeps + steady-state boots DDL-free; a bundle full of `always` scripts re-issues + `ACCESS EXCLUSIVE`-locking DDL on every connection and will deadlock live + readers. Prefer splitting DDL into small hash-gated files — one file per view + (or per shared dependency), triggers and constraints in their own run-once + files — so a table reshape only re-creates the views that actually depend on + it and never disturbs unrelated triggers or constraints. Atlas-style PostgreSQL `role` and `permission` blocks may live in the same HCL files as tables. The OSS migration layer supports roles and memberships plus diff --git a/migrate/migrate.go b/migrate/migrate.go index 24054df..24bc40a 100644 --- a/migrate/migrate.go +++ b/migrate/migrate.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io/fs" + "net/url" "path" "strings" @@ -116,7 +117,7 @@ func Apply(ctx context.Context, connection string, schemaFS fs.FS, opts ...Optio return err } - client, err := sqlclient.Open(ctx, connection) + client, err := sqlclient.Open(ctx, connectionWithLockTimeout(connection)) if err != nil { return fmt.Errorf("open database: %w", err) } @@ -169,7 +170,9 @@ func Apply(ctx context.Context, connection string, schemaFS fs.FS, opts ...Optio if err := runScriptPhase(ctx, db, cfg.name, ordered, phasePost); err != nil { return err } - if err := reconcileSecurity(ctx, db, cfg.name, security); err != nil { + if err := retryOnLockContention(ctx, "reconcile database security", func() error { + return reconcileSecurity(ctx, db, cfg.name, security) + }); err != nil { return fmt.Errorf("reconcile database security: %w", err) } return nil @@ -186,3 +189,23 @@ func withoutTableDrops(changes []schema.Change) []schema.Change { } return filtered } + +// connectionWithLockTimeout appends a libpq `options` runtime parameter so every +// connection Atlas opens bounds how long its ALTER TABLE DDL waits for a lock. A +// migration that would otherwise block indefinitely against a live reader fails +// fast (55P03) and the next apply re-plans from a fresh inspect, instead of +// camping on an ACCESS EXCLUSIVE lock and starving concurrent traffic. The value +// is returned unchanged when it is not a URL-form DSN or already sets options. +func connectionWithLockTimeout(connection string) string { + u, err := url.Parse(connection) + if err != nil || u.Scheme == "" { + return connection + } + q := u.Query() + if q.Get("options") != "" { + return connection + } + q.Set("options", "-c lock_timeout="+migrationLockTimeout) + u.RawQuery = q.Encode() + return u.String() +} diff --git a/migrate/migrate_test.go b/migrate/migrate_test.go index 144fc62..ee02616 100644 --- a/migrate/migrate_test.go +++ b/migrate/migrate_test.go @@ -22,3 +22,32 @@ func TestApplyValidatesInputsBeforeConnecting(t *testing.T) { assert.EqualError(t, Apply(t.Context(), "", fstest.MapFS{}), "connection string is empty") assert.EqualError(t, Apply(t.Context(), "postgres://unused", nil), "schema filesystem is nil") } + +func TestConnectionWithLockTimeout(t *testing.T) { + tests := []struct { + name string + connection string + want string + }{ + { + name: "url without query gains a bounded lock_timeout", + connection: "postgres://user@localhost/app?sslmode=disable", + want: "postgres://user@localhost/app?options=-c+lock_timeout%3D" + migrationLockTimeout + "&sslmode=disable", + }, + { + name: "existing options are preserved", + connection: "postgres://user@localhost/app?options=-c+statement_timeout%3D5s", + want: "postgres://user@localhost/app?options=-c+statement_timeout%3D5s", + }, + { + name: "keyword dsn is left untouched", + connection: "host=localhost user=app dbname=app sslmode=disable", + want: "host=localhost user=app dbname=app sslmode=disable", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + assert.Equal(t, tt.want, connectionWithLockTimeout(tt.connection)) + }) + } +} diff --git a/migrate/scripts.go b/migrate/scripts.go index 7a53206..7e71049 100644 --- a/migrate/scripts.go +++ b/migrate/scripts.go @@ -288,16 +288,18 @@ func runScriptPhase(ctx context.Context, db *sql.DB, scope string, ordered []*sc return nil } -// runTransactionalScript applies one transactional migration, retrying on lock -// contention. A transactional DDL script (e.g. 50_views_and_triggers.sql) takes -// ACCESS EXCLUSIVE on tables a concurrently-running process may be reading; the -// bounded lock_timeout converts the resulting hang into a retryable 55P03, and a -// deadlock (40P01) aborts one party — either way retrying lets the migration win -// once the reader's short query completes, instead of failing the whole run. -func runTransactionalScript(ctx context.Context, db *sql.DB, scope string, s *script) error { +// retryOnLockContention runs fn, retrying on transient lock contention (a +// detected deadlock, a lock_timeout, or a serialization failure) with linear +// backoff. A DDL statement (a transactional script, a dependent-view DROP, or a +// security reconciliation) takes ACCESS EXCLUSIVE on tables a concurrently +// running process may be reading; the bounded lock_timeout converts the +// resulting hang into a retryable 55P03, and a deadlock (40P01) aborts one party +// — either way retrying lets the migration win once the reader's short query +// completes, instead of failing the whole run. fn MUST be idempotent. +func retryOnLockContention(ctx context.Context, desc string, fn func() error) error { var lastErr error for attempt := 1; attempt <= migrationMaxAttempts; attempt++ { - err := applyTransactionalScript(ctx, db, scope, s) + err := fn() if err == nil { return nil } @@ -306,7 +308,7 @@ func runTransactionalScript(ctx context.Context, db *sql.DB, scope string, s *sc } lastErr = err if attempt < migrationMaxAttempts { - logger.Debugf("migration %s hit lock contention (attempt %d/%d): %v", s.path, attempt, migrationMaxAttempts, err) + logger.Debugf("%s hit lock contention (attempt %d/%d): %v", desc, attempt, migrationMaxAttempts, err) select { case <-ctx.Done(): return ctx.Err() @@ -314,7 +316,17 @@ func runTransactionalScript(ctx context.Context, db *sql.DB, scope string, s *sc } } } - return fmt.Errorf("execute SQL migration %s: gave up after %d attempts under lock contention: %w", s.path, migrationMaxAttempts, lastErr) + return fmt.Errorf("%s: gave up after %d attempts under lock contention: %w", desc, migrationMaxAttempts, lastErr) +} + +// runTransactionalScript applies one transactional migration under a bounded +// lock_timeout, retrying on lock contention. The view/trigger scripts split out +// of the historical 50_views_and_triggers monolith are the canonical DDL that +// contends with live readers. +func runTransactionalScript(ctx context.Context, db *sql.DB, scope string, s *script) error { + return retryOnLockContention(ctx, "execute SQL migration "+s.path, func() error { + return applyTransactionalScript(ctx, db, scope, s) + }) } // applyTransactionalScript runs the script (and records it) in a single diff --git a/migrate/security.go b/migrate/security.go index 459cbe6..abc3281 100644 --- a/migrate/security.go +++ b/migrate/security.go @@ -88,6 +88,12 @@ func reconcileSecurity(ctx context.Context, db *sql.DB, scope string, spec secur } defer tx.Rollback() //nolint:errcheck + // GRANT/REVOKE take strong locks on the targeted objects; bound the wait so + // reconciliation cannot camp against live traffic (the caller retries). + if _, err := tx.ExecContext(ctx, "SET LOCAL lock_timeout = '"+migrationLockTimeout+"'"); err != nil { + return fmt.Errorf("set lock_timeout for security reconciliation: %w", err) + } + previous, err := readSecurityState(ctx, tx, scope) if err != nil { return err diff --git a/migrate/view_deps.go b/migrate/view_deps.go index e0c4488..1e59839 100644 --- a/migrate/view_deps.go +++ b/migrate/view_deps.go @@ -228,9 +228,32 @@ func dropDependentView(ctx context.Context, db *sql.DB, v viewRef) error { } stmt := fmt.Sprintf("DROP %s IF EXISTS %s.%s CASCADE", kind, pq.QuoteIdentifier(v.schemaName), pq.QuoteIdentifier(v.name)) - if _, err := db.ExecContext(ctx, stmt); err != nil { + // DROP takes ACCESS EXCLUSIVE on the view; bound the wait and retry so a + // live reader cannot deadlock the reconciliation indefinitely. The DROP is + // idempotent (IF EXISTS), so re-running the transaction is safe. + if err := retryOnLockContention(ctx, "drop dependent view "+v.qualified(), func() error { + return execWithLockTimeout(ctx, db, stmt) + }); err != nil { return fmt.Errorf("drop dependent view %s: %w", v.qualified(), err) } logger.GetLogger("migrate").V(1).Infof("dropped dependent view %s for table reconciliation", v.qualified()) return nil } + +// execWithLockTimeout runs a single statement in a transaction that first bounds +// its lock waits, so DDL cannot camp on an ACCESS EXCLUSIVE lock against live +// traffic. +func execWithLockTimeout(ctx context.Context, db *sql.DB, stmt string) error { + tx, err := db.BeginTx(ctx, nil) + if err != nil { + return err + } + defer tx.Rollback() //nolint:errcheck + if _, err := tx.ExecContext(ctx, "SET LOCAL lock_timeout = '"+migrationLockTimeout+"'"); err != nil { + return err + } + if _, err := tx.ExecContext(ctx, stmt); err != nil { + return err + } + return tx.Commit() +} From 69d3ae7ab091548f2776d96a67031a934f97d82a Mon Sep 17 00:00:00 2001 From: Moshe Immerman Date: Thu, 16 Jul 2026 20:03:26 +0300 Subject: [PATCH 02/12] feat(database): enable embedded PostgreSQL diagnostics --- db/embedded.go | 72 +++++++++++++++++++++++++++++++++++++++++-- db/embedded_test.go | 74 +++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 144 insertions(+), 2 deletions(-) diff --git a/db/embedded.go b/db/embedded.go index 575c3af..e14064e 100644 --- a/db/embedded.go +++ b/db/embedded.go @@ -33,6 +33,9 @@ type EmbeddedConfig struct { // to os.Stderr (the library default is os.Stdout, which would corrupt callers // that write binary or structured data to stdout). Logger io.Writer + // PerformanceDiagnostics preloads pg_stat_statements, enables I/O timing, + // installs the extension, and fails startup if any part is unavailable. + PerformanceDiagnostics bool } // StartEmbedded launches a fergusstrange/embedded-postgres under cfg.DataDir @@ -87,7 +90,7 @@ func StartEmbedded(cfg EmbeddedConfig) (dsn string, stop func() error, err error if pgLog == nil { pgLog = os.Stderr } - server := embeddedpostgres.NewDatabase(embeddedpostgres.DefaultConfig(). + serverConfig := embeddedpostgres.DefaultConfig(). Port(port). DataPath(dataPath). RuntimePath(filepath.Join(cfg.DataDir, "runtime")). @@ -95,7 +98,11 @@ func StartEmbedded(cfg EmbeddedConfig) (dsn string, stop func() error, err error Version(pgVersion). Username(cfg.Username).Password(cfg.Password). Database(cfg.Database). - Logger(pgLog)) + Logger(pgLog) + if parameters := performanceDiagnosticStartParameters(cfg.PerformanceDiagnostics); len(parameters) > 0 { + serverConfig = serverConfig.StartParameters(parameters) + } + server := embeddedpostgres.NewDatabase(serverConfig) ownedByUs := true if err := server.Start(); err != nil { @@ -125,9 +132,70 @@ func StartEmbedded(cfg EmbeddedConfig) (dsn string, stop func() error, err error _ = stop() return "", nil, fmt.Errorf("embedded postgres never became ready: %w", err) } + if cfg.PerformanceDiagnostics { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := EnsurePerformanceDiagnostics(ctx, dsn); err != nil { + _ = stop() + return "", nil, err + } + } return dsn, stop, nil } +func performanceDiagnosticStartParameters(enabled bool) map[string]string { + if !enabled { + return nil + } + return map[string]string{ + "shared_preload_libraries": "pg_stat_statements", + "track_io_timing": "on", + } +} + +func validatePerformanceDiagnosticSettings(preloadedLibraries, trackIOTiming string) error { + foundStatementStats := false + for _, library := range strings.Split(preloadedLibraries, ",") { + if strings.TrimSpace(library) == "pg_stat_statements" { + foundStatementStats = true + break + } + } + if !foundStatementStats { + return errors.New("PostgreSQL performance diagnostics require shared_preload_libraries=pg_stat_statements; update the server configuration and restart PostgreSQL") + } + if trackIOTiming != "on" { + return errors.New("PostgreSQL performance diagnostics require track_io_timing=on; update the server configuration and restart PostgreSQL") + } + return nil +} + +// EnsurePerformanceDiagnostics validates server-level settings before +// installing pg_stat_statements in the selected database. Server settings are +// checked first so an externally managed instance fails without partial DDL. +func EnsurePerformanceDiagnostics(ctx context.Context, dsn string) error { + conn, err := pgx.Connect(ctx, dsn) + if err != nil { + return fmt.Errorf("connect for PostgreSQL performance diagnostics: %w", err) + } + defer conn.Close(context.Background()) //nolint:errcheck + + var preloadedLibraries, trackIOTiming string + if err := conn.QueryRow(ctx, "SHOW shared_preload_libraries").Scan(&preloadedLibraries); err != nil { + return fmt.Errorf("read shared_preload_libraries: %w", err) + } + if err := conn.QueryRow(ctx, "SHOW track_io_timing").Scan(&trackIOTiming); err != nil { + return fmt.Errorf("read track_io_timing: %w", err) + } + if err := validatePerformanceDiagnosticSettings(preloadedLibraries, trackIOTiming); err != nil { + return err + } + if _, err := conn.Exec(ctx, "CREATE EXTENSION IF NOT EXISTS pg_stat_statements"); err != nil { + return fmt.Errorf("install pg_stat_statements: %w", err) + } + return nil +} + // FreePort binds :0 to discover a free TCP port. Public so callers can reuse // it for adjacent services (e.g. postgrest) that need an unclaimed port. func FreePort() (int, error) { diff --git a/db/embedded_test.go b/db/embedded_test.go index b86b4f2..761e412 100644 --- a/db/embedded_test.go +++ b/db/embedded_test.go @@ -20,6 +20,50 @@ func TestFreePort_ReturnsPositive(t *testing.T) { assert.Less(t, port, 65536) } +func TestPerformanceDiagnosticStartParameters(t *testing.T) { + assert.Empty(t, performanceDiagnosticStartParameters(false)) + assert.Equal(t, map[string]string{ + "shared_preload_libraries": "pg_stat_statements", + "track_io_timing": "on", + }, performanceDiagnosticStartParameters(true)) +} + +func TestValidatePerformanceDiagnosticSettings(t *testing.T) { + tests := []struct { + name string + preloadedLibraries string + trackIOTiming string + wantError string + }{ + { + name: "ready", + preloadedLibraries: "auto_explain, pg_stat_statements", + trackIOTiming: "on", + }, + { + name: "statement statistics not preloaded", + trackIOTiming: "on", + wantError: "PostgreSQL performance diagnostics require shared_preload_libraries=pg_stat_statements; update the server configuration and restart PostgreSQL", + }, + { + name: "IO timing disabled", + preloadedLibraries: "pg_stat_statements", + trackIOTiming: "off", + wantError: "PostgreSQL performance diagnostics require track_io_timing=on; update the server configuration and restart PostgreSQL", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := validatePerformanceDiagnosticSettings(tt.preloadedLibraries, tt.trackIOTiming) + if tt.wantError == "" { + require.NoError(t, err) + return + } + require.EqualError(t, err, tt.wantError) + }) + } +} + // TestStartEmbedded_StartPingStop is an integration test. The // fergusstrange/embedded-postgres library downloads a ~50MB postgres tarball // on first run — in CI environments without network we skip rather than fail. @@ -51,3 +95,33 @@ func TestStartEmbedded_StartPingStop(t *testing.T) { _, statErr := os.Stat(filepath.Join(dir, "data", "PG_VERSION")) assert.NoError(t, statErr, "embedded postgres should have initialized a cluster") } + +func TestStartEmbedded_PerformanceDiagnostics(t *testing.T) { + if os.Getenv("COMMONS_DB_EMBEDDED_TEST") == "" { + t.Skip("set COMMONS_DB_EMBEDDED_TEST=1 to run embedded-postgres integration tests") + } + dir := t.TempDir() + + dsn, stop, err := StartEmbedded(EmbeddedConfig{ + DataDir: dir, + Database: "diagnostics", + PerformanceDiagnostics: true, + }) + require.NoError(t, err) + t.Cleanup(func() { _ = stop() }) + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + conn, err := pgx.Connect(ctx, dsn) + require.NoError(t, err) + defer conn.Close(context.Background()) //nolint:errcheck + + var preloadedLibraries, trackIOTiming, extension string + require.NoError(t, conn.QueryRow(ctx, "SHOW shared_preload_libraries").Scan(&preloadedLibraries)) + require.NoError(t, conn.QueryRow(ctx, "SHOW track_io_timing").Scan(&trackIOTiming)) + require.NoError(t, conn.QueryRow(ctx, + "SELECT extname FROM pg_extension WHERE extname = 'pg_stat_statements'").Scan(&extension)) + assert.Contains(t, preloadedLibraries, "pg_stat_statements") + assert.Equal(t, "on", trackIOTiming) + assert.Equal(t, "pg_stat_statements", extension) +} From ddeadf32f6dc5504c2059fa87e5674f8f69fefab Mon Sep 17 00:00:00 2001 From: Moshe Immerman Date: Fri, 17 Jul 2026 17:46:11 +0300 Subject: [PATCH 03/12] feat(query): add OpenTelemetry trace querying via nested OpenSearch connections Adds first-class OpenTelemetry connections backed by validated nested OpenSearch connections. Supports Jaeger and flat trace querying, normalized result fields, filtering, discovery, testing, and schema-driven connection selection. --- cmd/query/connections/actions.go | 11 + cmd/query/connections/browser.go | 12 + cmd/query/connections/info.go | 10 + cmd/query/connections/service.go | 30 +++ cmd/query/connections/service_test.go | 20 ++ .../www/src/connectionSchemaMonaco.test.ts | 4 +- connection/opentelemetry.go | 48 ++++ connection/opentelemetry_test.go | 50 ++++ models/connections.go | 1 + query/providers/new_providers_test.go | 63 +++++ query/providers/opensearch.go | 12 + query/providers/opentelemetry.go | 241 ++++++++++++++++++ query/providers/opentelemetry_parser.go | 215 ++++++++++++++++ query/schema/connection.go | 24 +- query/schema/connection_providers.go | 37 +-- query/schema/connection_types.go | 3 +- schemas/src/connection.json | 17 ++ schemas/src/connections/opentelemetry.json | 58 +++++ 18 files changed, 835 insertions(+), 21 deletions(-) create mode 100644 connection/opentelemetry.go create mode 100644 connection/opentelemetry_test.go create mode 100644 query/providers/opentelemetry.go create mode 100644 query/providers/opentelemetry_parser.go create mode 100644 schemas/src/connections/opentelemetry.json diff --git a/cmd/query/connections/actions.go b/cmd/query/connections/actions.go index 64d2ce4..865a6d9 100644 --- a/cmd/query/connections/actions.go +++ b/cmd/query/connections/actions.go @@ -146,6 +146,17 @@ func maskedConnection(c *models.Connection) resolvedConnection { // testConnection probes the resolved URL: a TCP connect proves the host:port is // reachable, and http/https URLs additionally report the response status. func testConnection(ctx dbcontext.Context, c *models.Connection) testResult { + if c.Type == models.ConnectionTypeOpenTelemetry { + openTelemetry, err := dbconnection.NewOpenTelemetry(c) + if err != nil { + return testResult{OK: false, Message: err.Error()} + } + nested, err := openTelemetry.ResolveOpenSearch(ctx) + if err != nil { + return testResult{OK: false, Message: err.Error()} + } + return testConnection(ctx, nested) + } if c.URL == "" { return testResult{OK: false, Message: "connection has no URL to test"} } diff --git a/cmd/query/connections/browser.go b/cmd/query/connections/browser.go index 6f5c7ea..e041313 100644 --- a/cmd/query/connections/browser.go +++ b/cmd/query/connections/browser.go @@ -112,6 +112,18 @@ func (h *connectionBrowserHandler) ServeHTTP(w http.ResponseWriter, r *http.Requ http.Error(w, err.Error(), http.StatusNotFound) return } + if conn.Type == models.ConnectionTypeOpenTelemetry { + openTelemetry, resolveErr := dbconnection.NewOpenTelemetry(conn) + if resolveErr != nil { + http.Error(w, resolveErr.Error(), http.StatusUnprocessableEntity) + return + } + conn, resolveErr = openTelemetry.ResolveOpenSearch(h.ctx) + if resolveErr != nil { + http.Error(w, resolveErr.Error(), http.StatusUnprocessableEntity) + return + } + } tail = strings.TrimSuffix(tail, "/") switch { diff --git a/cmd/query/connections/info.go b/cmd/query/connections/info.go index 9b2d9f9..032ad38 100644 --- a/cmd/query/connections/info.go +++ b/cmd/query/connections/info.go @@ -135,6 +135,16 @@ func discoverServer(ctx context.Context, connectionContext dbcontext.Context, co info, err = discoverSQLServer(ctx, connectionContext, connection) case models.ConnectionTypeOpenSearch: info, err = discoverOpenSearch(ctx, connectionContext, connection) + case models.ConnectionTypeOpenTelemetry: + var openTelemetry dbconnection.OpenTelemetry + openTelemetry, err = dbconnection.NewOpenTelemetry(connection) + if err == nil { + var nested *models.Connection + nested, err = openTelemetry.ResolveOpenSearch(connectionContext) + if err == nil { + info, err = discoverOpenSearch(ctx, connectionContext, nested) + } + } case models.ConnectionTypePrometheus: info, err = discoverPrometheus(ctx, connectionContext, connection) case models.ConnectionTypeRedis: diff --git a/cmd/query/connections/service.go b/cmd/query/connections/service.go index deed6bd..7d17b80 100644 --- a/cmd/query/connections/service.go +++ b/cmd/query/connections/service.go @@ -9,6 +9,7 @@ import ( "strings" "github.com/flanksource/clicky" + dbconnection "github.com/flanksource/commons-db/connection" dbcontext "github.com/flanksource/commons-db/context" "github.com/flanksource/commons-db/models" "github.com/google/uuid" @@ -191,6 +192,9 @@ func createConnection(db *gorm.DB, body map[string]any) (*models.Connection, err if err != nil { return nil, err } + if err := validateNestedConnection(db, c); err != nil { + return nil, err + } c.ID = uuid.Nil // let the DB assign the id if err := db.Create(c).Error; err != nil { return nil, fmt.Errorf("create connection: %w", err) @@ -209,6 +213,9 @@ func updateConnection(db *gorm.DB, id string, body map[string]any) (*models.Conn if err != nil { return nil, err } + if err := validateNestedConnection(db, incoming); err != nil { + return nil, err + } applyConnectionUpdate(existing, incoming) if err := db.Save(existing).Error; err != nil { return nil, fmt.Errorf("update connection: %w", err) @@ -217,6 +224,29 @@ func updateConnection(db *gorm.DB, id string, body map[string]any) (*models.Conn return existing, nil } +func validateNestedConnection(db *gorm.DB, candidate *models.Connection) error { + if candidate.Type != models.ConnectionTypeOpenTelemetry { + return nil + } + openTelemetry, err := dbconnection.NewOpenTelemetry(candidate) + if err != nil { + return err + } + name := strings.TrimPrefix(openTelemetry.Connection, "connection://") + if strings.Contains(name, "/") { + parts := strings.Split(name, "/") + name = parts[len(parts)-1] + } + nested, err := findConnection(db, name) + if err != nil { + return fmt.Errorf("resolve nested OpenSearch connection: %w", err) + } + if nested.Type != models.ConnectionTypeOpenSearch { + return fmt.Errorf("nested connection %q has type %q, expected %q", nested.Name, nested.Type, models.ConnectionTypeOpenSearch) + } + return nil +} + // deleteConnection removes a connection by id, erroring when absent. func deleteConnection(db *gorm.DB, id string) error { res := db.Where("id = ?", id).Delete(&models.Connection{}) diff --git a/cmd/query/connections/service_test.go b/cmd/query/connections/service_test.go index 51679da..14fd11d 100644 --- a/cmd/query/connections/service_test.go +++ b/cmd/query/connections/service_test.go @@ -4,6 +4,7 @@ import ( "testing" "github.com/flanksource/commons-db/models" + "github.com/flanksource/commons-db/types" "github.com/google/uuid" ) @@ -41,6 +42,25 @@ func TestConnectionFromBodyRequiresNameAndType(t *testing.T) { } } +func TestValidateOpenTelemetryRequiresNestedOpenSearchConnection(t *testing.T) { + database := connectionInfoTestDB(t) + search := models.Connection{ID: uuid.New(), Name: "OS", Type: models.ConnectionTypeOpenSearch} + if err := database.Create(&search).Error; err != nil { + t.Fatal(err) + } + candidate := &models.Connection{ + Name: "traces", Type: models.ConnectionTypeOpenTelemetry, + Properties: types.JSONStringMap{"connection": "connection://OS"}, + } + if err := validateNestedConnection(database, candidate); err != nil { + t.Fatal(err) + } + candidate.Properties["connection"] = "connection://traces" + if err := validateNestedConnection(database, candidate); err == nil { + t.Fatal("expected self-referencing OpenTelemetry connection to fail") + } +} + func TestApplyConnectionUpdatePreservesSecretWhenBlank(t *testing.T) { existing, _ := connectionFromBody(map[string]any{"name": "db", "type": "postgres", "password": "old"}) incoming, _ := connectionFromBody(map[string]any{"name": "db2", "type": "mysql"}) // no password diff --git a/cmd/query/www/src/connectionSchemaMonaco.test.ts b/cmd/query/www/src/connectionSchemaMonaco.test.ts index bb866e6..3cd5d41 100644 --- a/cmd/query/www/src/connectionSchemaMonaco.test.ts +++ b/cmd/query/www/src/connectionSchemaMonaco.test.ts @@ -10,10 +10,10 @@ describe("connection schema Monaco compatibility", () => { expect(result.schema).toBeDefined(); const definitions = result.schema?.definitions as Record; - expect(Object.keys(definitions)).toHaveLength(55); + expect(Object.keys(definitions)).toHaveLength(56); const serialized = JSON.stringify(result.schema); expect(serialized).not.toContain("#/$defs/"); - expect(serialized.match(/#\/definitions\//g)).toHaveLength(55); + expect(serialized.match(/#\/definitions\//g)).toHaveLength(56); }); }); diff --git a/connection/opentelemetry.go b/connection/opentelemetry.go new file mode 100644 index 0000000..811ea5e --- /dev/null +++ b/connection/opentelemetry.go @@ -0,0 +1,48 @@ +package connection + +import ( + "fmt" + "strings" + + "github.com/flanksource/commons-db/models" +) + +type OpenTelemetry struct { + Name string + Connection string +} + +type connectionHydrator interface { + HydrateConnectionByURL(string) (*models.Connection, error) +} + +func NewOpenTelemetry(connection *models.Connection) (OpenTelemetry, error) { + if connection == nil { + return OpenTelemetry{}, fmt.Errorf("opentelemetry connection is required") + } + if connection.Type != models.ConnectionTypeOpenTelemetry { + return OpenTelemetry{}, fmt.Errorf("connection %q has type %q, expected %q", connection.Name, connection.Type, models.ConnectionTypeOpenTelemetry) + } + nested := strings.TrimSpace(connection.Properties["connection"]) + if nested == "" { + return OpenTelemetry{}, fmt.Errorf("opentelemetry connection %q is missing properties.connection", connection.Name) + } + if nested == "connection://"+connection.Name { + return OpenTelemetry{}, fmt.Errorf("opentelemetry connection %q cannot reference itself", connection.Name) + } + return OpenTelemetry{Name: connection.Name, Connection: nested}, nil +} + +func (connection OpenTelemetry) ResolveOpenSearch(ctx connectionHydrator) (*models.Connection, error) { + nested, err := ctx.HydrateConnectionByURL(connection.Connection) + if err != nil { + return nil, fmt.Errorf("resolve nested OpenSearch connection %q: %w", connection.Connection, err) + } + if nested == nil { + return nil, fmt.Errorf("nested OpenSearch connection %q not found", connection.Connection) + } + if nested.Type != models.ConnectionTypeOpenSearch { + return nil, fmt.Errorf("nested connection %q has type %q, expected %q", nested.Name, nested.Type, models.ConnectionTypeOpenSearch) + } + return nested, nil +} diff --git a/connection/opentelemetry_test.go b/connection/opentelemetry_test.go new file mode 100644 index 0000000..1a86b42 --- /dev/null +++ b/connection/opentelemetry_test.go @@ -0,0 +1,50 @@ +package connection + +import ( + "testing" + + "github.com/flanksource/commons-db/models" + "github.com/flanksource/commons-db/types" +) + +type openTelemetryHydrator struct { + connection *models.Connection + requested string +} + +func (h *openTelemetryHydrator) HydrateConnectionByURL(connection string) (*models.Connection, error) { + h.requested = connection + return h.connection, nil +} + +func TestOpenTelemetryResolvesNestedOpenSearchConnection(t *testing.T) { + outer, err := NewOpenTelemetry(&models.Connection{ + Name: "traces", Type: models.ConnectionTypeOpenTelemetry, + Properties: types.JSONStringMap{"connection": "connection://OS"}, + }) + if err != nil { + t.Fatal(err) + } + hydrator := &openTelemetryHydrator{connection: &models.Connection{Name: "OS", Type: models.ConnectionTypeOpenSearch}} + nested, err := outer.ResolveOpenSearch(hydrator) + if err != nil { + t.Fatal(err) + } + if hydrator.requested != "connection://OS" || nested.Name != "OS" { + t.Fatalf("unexpected nested resolution: requested=%q connection=%+v", hydrator.requested, nested) + } +} + +func TestOpenTelemetryRejectsWrongNestedConnectionType(t *testing.T) { + outer, err := NewOpenTelemetry(&models.Connection{ + Name: "traces", Type: models.ConnectionTypeOpenTelemetry, + Properties: types.JSONStringMap{"connection": "connection://db"}, + }) + if err != nil { + t.Fatal(err) + } + _, err = outer.ResolveOpenSearch(&openTelemetryHydrator{connection: &models.Connection{Name: "db", Type: models.ConnectionTypePostgres}}) + if err == nil { + t.Fatal("expected wrong nested connection type to fail") + } +} diff --git a/models/connections.go b/models/connections.go index d9d38d7..dfb38ca 100644 --- a/models/connections.go +++ b/models/connections.go @@ -58,6 +58,7 @@ const ( ConnectionTypeNtfy = "ntfy" ConnectionTypeOllama = "ollama" ConnectionTypeOpenAI = "openai" + ConnectionTypeOpenTelemetry = "opentelemetry" ConnectionTypeOpenSearch = "opensearch" ConnectionTypeOpsGenie = "opsgenie" ConnectionTypePostgres = "postgres" diff --git a/query/providers/new_providers_test.go b/query/providers/new_providers_test.go index 37cdc2f..106ccae 100644 --- a/query/providers/new_providers_test.go +++ b/query/providers/new_providers_test.go @@ -1,14 +1,77 @@ package providers_test import ( + "encoding/json" + "fmt" "net/http" + "net/http/httptest" context "github.com/flanksource/commons-db/context" + "github.com/flanksource/commons-db/models" "github.com/flanksource/commons-db/query" + "github.com/flanksource/commons-db/types" + "github.com/google/uuid" . "github.com/onsi/ginkgo/v2" . "github.com/onsi/gomega" + "gorm.io/driver/sqlite" + "gorm.io/gorm" ) +var _ = Describe("opentelemetry provider", func() { + It("queries Jaeger spans through its nested OpenSearch connection", func() { + var requestBody map[string]any + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodHead { + w.WriteHeader(http.StatusOK) + return + } + Expect(json.NewDecoder(r.Body).Decode(&requestBody)).To(Succeed()) + w.Header().Set("Content-Type", "application/json") + _, _ = fmt.Fprint(w, `{"hits":{"total":{"value":1,"relation":"eq"},"hits":[{"_id":"one","_source":{"traceID":"trace-1","spanID":"span-1","operationName":"process message","startTimeMillis":1710000000000,"duration":123000,"process":{"serviceName":"prod-api"},"tags":[{"key":"otel@status_code","value":"ERROR"},{"key":"input@xml","value":""}]}}]}}`) + })) + defer server.Close() + + database, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + Expect(err).ToNot(HaveOccurred()) + Expect(database.Exec(`CREATE TABLE connections ( +id TEXT PRIMARY KEY, name TEXT, namespace TEXT, source TEXT, type TEXT, +url TEXT, username TEXT, password TEXT, properties TEXT, certificate TEXT, +insecure_tls NUMERIC, created_at DATETIME, updated_at DATETIME, created_by TEXT +)`).Error).ToNot(HaveOccurred()) + Expect(database.Create(&models.Connection{ + ID: uuid.New(), Name: "OS", Type: models.ConnectionTypeOpenSearch, URL: server.URL, + }).Error).ToNot(HaveOccurred()) + Expect(database.Create(&models.Connection{ + ID: uuid.New(), Name: "traces", Type: models.ConnectionTypeOpenTelemetry, + Properties: types.JSONStringMap{"connection": "connection://OS"}, + }).Error).ToNot(HaveOccurred()) + + result, err := query.Execute(context.New().WithDB(database, nil), query.Profile{ + Name: "jms", + Provider: query.ProviderConfig{ + Type: "opentelemetry", Connection: "connection://traces", + Options: map[string]any{ + "format": "jaeger", "index": "jaeger-span*", "dateField": "startTimeMillis", + "traceIdField": "traceID", "spanIdField": "spanID", "serviceField": "process.serviceName", + "operationField": "operationName", "statusFields": []string{"tag.otel@status_code"}, + "params": map[string]any{"namespace": map[string]any{"field": "process.serviceName", "operator": "term"}}, + }, + }, + Params: []query.ParamDef{{Name: "namespace", Template: "{value}-api"}}, + }, map[string]any{"namespace": "prod"}) + Expect(err).ToNot(HaveOccurred()) + Expect(result.Rows).To(HaveLen(1)) + Expect(result.Rows[0]).To(HaveKeyWithValue("trace_id", "trace-1")) + Expect(result.Rows[0]).To(HaveKeyWithValue("service", "prod-api")) + Expect(result.Rows[0]).To(HaveKeyWithValue("input.xml", "")) + Expect(result.Rows[0]).To(HaveKeyWithValue("duration_ms", float64(123))) + + boolQuery := requestBody["query"].(map[string]any)["bool"].(map[string]any) + filter := boolQuery["filter"].([]any) + Expect(filter[0]).To(Equal(map[string]any{"term": map[string]any{"process.serviceName": "prod-api"}})) + }) +}) + var _ = Describe("postgrest provider", func() { It("returns rows from a PostgREST JSON array response", func() { srv := jsonServer(http.StatusOK, `[{"id":1,"name":"alpha"},{"id":2,"name":"beta"}]`) diff --git a/query/providers/opensearch.go b/query/providers/opensearch.go index ef62123..a18e0f5 100644 --- a/query/providers/opensearch.go +++ b/query/providers/opensearch.go @@ -10,6 +10,7 @@ import ( dbconnection "github.com/flanksource/commons-db/connection" "github.com/flanksource/commons-db/context" "github.com/flanksource/commons-db/logs/opensearch" + "github.com/flanksource/commons-db/models" "github.com/flanksource/commons-db/query" ) @@ -140,6 +141,17 @@ func openSearchClient(ctx context.Context, req query.ProviderRequest) (*opensear return searcher, opts, nil } +func openSearchClientForConnection(ctx context.Context, conn *models.Connection) (*opensearch.Searcher, error) { + if conn == nil || conn.URL == "" { + return nil, fmt.Errorf("OpenSearch connection URL is required") + } + httpConnection, err := dbconnection.NewHTTPConnection(ctx, *conn) + if err != nil { + return nil, err + } + return opensearch.NewWithTransport(ctx, opensearch.Backend{Address: conn.URL}, nil, httpConnection.Transport()) +} + type openSearchRowIterator struct { ctx context.Context cleanupCtx context.Context diff --git a/query/providers/opentelemetry.go b/query/providers/opentelemetry.go new file mode 100644 index 0000000..56a85b4 --- /dev/null +++ b/query/providers/opentelemetry.go @@ -0,0 +1,241 @@ +package providers + +import ( + "encoding/json" + "fmt" + "sort" + "strconv" + "strings" + + dbconnection "github.com/flanksource/commons-db/connection" + "github.com/flanksource/commons-db/context" + "github.com/flanksource/commons-db/logs/opensearch" + "github.com/flanksource/commons-db/query" +) + +func init() { + query.RegisterProvider(&openTelemetryProvider{}) +} + +type openTelemetryProvider struct{} + +func (openTelemetryProvider) Type() string { return "opentelemetry" } + +type openTelemetryOptions struct { + Format string `json:"format,omitempty"` + Index string `json:"index,omitempty"` + DateField string `json:"dateField,omitempty"` + TraceIDField string `json:"traceIdField,omitempty"` + SpanIDField string `json:"spanIdField,omitempty"` + ParentIDField string `json:"parentIdField,omitempty"` + ParentRefType string `json:"parentRefType,omitempty"` + ServiceField string `json:"serviceField,omitempty"` + OperationField string `json:"operationField,omitempty"` + StatusFields []string `json:"statusFields,omitempty"` + SelectFields []string `json:"selectFields,omitempty"` + SourceExcludes []string `json:"sourceExcludes,omitempty"` + Params map[string]openTelemetryParam `json:"params,omitempty"` + Limit int `json:"limit,omitempty"` +} + +type openTelemetryParam struct { + Field string `json:"field,omitempty"` + Operator string `json:"operator,omitempty"` + Clause string `json:"clause,omitempty"` + Format string `json:"format,omitempty"` + Internal bool `json:"internal,omitempty"` +} + +func (openTelemetryProvider) Execute(ctx context.Context, req query.ProviderRequest) ([]query.Row, error) { + if req.Connection == "" { + return nil, fmt.Errorf("opentelemetry connection is required") + } + options, err := query.DecodeOptions[openTelemetryOptions](req.Options) + if err != nil { + return nil, err + } + options.withDefaults() + if options.Format != "jaeger" && options.Format != "flat" { + return nil, fmt.Errorf("unsupported opentelemetry format %q", options.Format) + } + outerModel, err := ctx.HydrateConnectionByURL(req.Connection) + if err != nil { + return nil, fmt.Errorf("hydrate opentelemetry connection %q: %w", req.Connection, err) + } + outer, err := dbconnection.NewOpenTelemetry(outerModel) + if err != nil { + return nil, err + } + nested, err := outer.ResolveOpenSearch(ctx) + if err != nil { + return nil, err + } + searcher, err := openSearchClientForConnection(ctx, nested) + if err != nil { + return nil, err + } + body, err := buildOpenTelemetryQuery(options, req.Params) + if err != nil { + return nil, err + } + encoded, err := json.Marshal(body) + if err != nil { + return nil, fmt.Errorf("encode OpenSearch trace query: %w", err) + } + limit := options.Limit + if limit <= 0 { + limit = 500 + } + result, err := searcher.SearchRaw(ctx, opensearch.Request{Index: options.Index, Query: string(encoded), Limit: strconv.Itoa(limit)}) + if err != nil { + return nil, err + } + rows := make([]query.Row, 0, len(result.Hits.Hits)) + for _, hit := range result.Hits.Hits { + rows = append(rows, openTelemetryRow(hit.Source, options)) + } + return rows, nil +} + +func (options *openTelemetryOptions) withDefaults() { + if options.Format == "" { + options.Format = "flat" + } + if options.Index == "" { + options.Index = "otel-traces-*" + } + if options.DateField == "" { + options.DateField = "@timestamp" + } + if options.TraceIDField == "" { + options.TraceIDField = "trace_id" + } + if options.SpanIDField == "" { + options.SpanIDField = "span_id" + } + if options.ParentIDField == "" { + options.ParentIDField = "parent_id" + } + if options.ServiceField == "" { + options.ServiceField = "service_name" + } + if options.OperationField == "" { + options.OperationField = "operation_name" + } +} + +func buildOpenTelemetryQuery(options openTelemetryOptions, params map[string]any) (map[string]any, error) { + clauses := map[string][]map[string]any{"filter": {}, "must": {}, "should": {}, "must_not": {}} + names := make([]string, 0, len(params)) + for name := range params { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + definition, ok := options.Params[name] + if !ok { + return nil, fmt.Errorf("filter %q is not supported by opentelemetry profile", name) + } + if definition.Internal { + return nil, fmt.Errorf("filter %q is internal to the opentelemetry profile", name) + } + built, err := buildOpenTelemetryParam(definition, params[name]) + if err != nil { + return nil, fmt.Errorf("build filter %q: %w", name, err) + } + clause := definition.Clause + if clause == "" { + clause = "filter" + } + if _, ok := clauses[clause]; !ok { + return nil, fmt.Errorf("unsupported clause %q for filter %q", clause, name) + } + clauses[clause] = append(clauses[clause], built...) + } + body := map[string]any{"sort": []map[string]any{{options.DateField: map[string]any{"order": "desc"}}}} + if len(options.SelectFields) > 0 { + body["stored_fields"] = []string{"*"} + body["fields"] = options.SelectFields + } + if len(options.SourceExcludes) > 0 { + body["_source"] = map[string]any{"excludes": options.SourceExcludes} + } + boolQuery := map[string]any{} + for _, clause := range []string{"filter", "must", "should", "must_not"} { + if len(clauses[clause]) > 0 { + boolQuery[clause] = clauses[clause] + } + } + if len(boolQuery) == 0 { + body["query"] = map[string]any{"match_all": map[string]any{}} + } else { + if len(clauses["should"]) > 0 { + boolQuery["minimum_should_match"] = 1 + } + body["query"] = map[string]any{"bool": boolQuery} + } + return body, nil +} + +func buildOpenTelemetryParam(param openTelemetryParam, value any) ([]map[string]any, error) { + if param.Field == "" { + return nil, fmt.Errorf("field is required") + } + values := normalizeOpenTelemetryValues(value) + if len(values) == 0 { + return nil, nil + } + operator := param.Operator + if operator == "" { + operator = "term" + } + switch operator { + case "term": + if len(values) > 1 { + return []map[string]any{{"terms": map[string]any{param.Field: values}}}, nil + } + return []map[string]any{{"term": map[string]any{param.Field: values[0]}}}, nil + case "terms": + return []map[string]any{{"terms": map[string]any{param.Field: values}}}, nil + case "match_phrase", "wildcard": + result := make([]map[string]any, 0, len(values)) + for _, item := range values { + result = append(result, map[string]any{operator: map[string]any{param.Field: item}}) + } + return result, nil + case "query_string": + result := make([]map[string]any, 0, len(values)) + for _, item := range values { + result = append(result, map[string]any{"query_string": map[string]any{"fields": []string{param.Field}, "query": item}}) + } + return result, nil + case "exists": + return []map[string]any{{"exists": map[string]any{"field": param.Field}}}, nil + default: + return nil, fmt.Errorf("unsupported operator %q", operator) + } +} + +func normalizeOpenTelemetryValues(value any) []any { + switch typed := value.(type) { + case []any: + return typed + case []string: + values := make([]any, len(typed)) + for index := range typed { + values[index] = typed[index] + } + return values + case string: + parts := strings.Split(typed, ",") + values := make([]any, 0, len(parts)) + for _, part := range parts { + if part = strings.TrimSpace(part); part != "" { + values = append(values, part) + } + } + return values + default: + return []any{typed} + } +} diff --git a/query/providers/opentelemetry_parser.go b/query/providers/opentelemetry_parser.go new file mode 100644 index 0000000..ea75402 --- /dev/null +++ b/query/providers/opentelemetry_parser.go @@ -0,0 +1,215 @@ +package providers + +import ( + "encoding/json" + "fmt" + "math" + "strconv" + "strings" + "time" + + "github.com/flanksource/commons-db/query" +) + +func openTelemetryRow(document map[string]any, options openTelemetryOptions) query.Row { + attributes := flattenTraceDocument(document) + row := query.Row{ + "timestamp": normalizeTraceTimestamp(stringTraceValue(firstTracePath(document, options.DateField, "@timestamp", "timestamp", "startTimeMillis"))), + "trace_id": stringTraceValue(firstTracePath(document, options.TraceIDField, "trace_id", "traceID")), + "span_id": stringTraceValue(firstTracePath(document, options.SpanIDField, "span_id", "spanID")), + "parent_id": traceParentID(document, options), + "service": stringTraceValue(firstTracePath(document, options.ServiceField, "service_name", "process.serviceName")), + "operation": stringTraceValue(firstTracePath(document, options.OperationField, "operation_name", "operationName")), + "status": traceStatus(document, attributes, options), + "duration_ms": traceDurationMillis(firstTracePath(document, "duration_ms", "duration", "durationNano", "duration_nano"), options.Format), + "_attributes": attributes, + } + row["id"] = row["span_id"] + row["service_name"] = row["service"] + row["operation_name"] = row["operation"] + for name, value := range attributes { + normalized := normalizeTraceAttributeName(name) + if _, exists := row[normalized]; !exists { + row[normalized] = value + } + } + return row +} + +func traceParentID(document map[string]any, options openTelemetryOptions) string { + if options.Format == "jaeger" { + if references, ok := document["references"].([]any); ok { + for _, raw := range references { + reference, _ := raw.(map[string]any) + if options.ParentRefType != "" && stringTraceValue(reference["refType"]) != options.ParentRefType { + continue + } + if id := stringTraceValue(firstTracePath(reference, "spanID", "span_id")); id != "" { + return id + } + } + } + } + return stringTraceValue(firstTracePath(document, options.ParentIDField, "parent_id", "parentID")) +} + +func traceStatus(document, attributes map[string]any, options openTelemetryOptions) string { + fields := append(append([]string{}, options.StatusFields...), "status", "status.code", "tag.error", "error") + for _, field := range fields { + value := stringTraceValue(firstTracePath(document, field)) + if value == "" { + value = stringTraceValue(attributes[field]) + } + if value != "" { + return strings.ToLower(value) + } + } + return "" +} + +func flattenTraceDocument(document map[string]any) map[string]any { + result := map[string]any{} + var walk func(string, any) + walk = func(prefix string, value any) { + switch typed := value.(type) { + case map[string]any: + for name, child := range typed { + next := name + if prefix != "" { + next = prefix + "." + name + } + walk(next, child) + } + case []any: + if isJaegerTagList(prefix, typed) { + for _, raw := range typed { + tag, _ := raw.(map[string]any) + name := stringTraceValue(tag["key"]) + if name != "" { + result["tag."+name] = firstTracePath(tag, "value", "vStr", "vDouble", "vBool", "vLong") + } + } + return + } + result[prefix] = typed + default: + result[prefix] = typed + } + } + walk("", document) + return result +} + +func normalizeTraceAttributeName(name string) string { + name = strings.TrimPrefix(name, "tag.") + return strings.ReplaceAll(name, "@", ".") +} + +func isJaegerTagList(prefix string, values []any) bool { + if prefix != "tags" && prefix != "process.tags" || len(values) == 0 { + return false + } + first, ok := values[0].(map[string]any) + if !ok { + return false + } + _, ok = first["key"] + return ok +} + +func firstTracePath(document map[string]any, paths ...string) any { + for _, path := range paths { + if value := lookupTracePath(document, path); value != nil { + return value + } + } + return nil +} + +func lookupTracePath(document map[string]any, path string) any { + if document == nil || path == "" { + return nil + } + if value, ok := document[path]; ok { + return value + } + var current any = document + for _, part := range strings.Split(path, ".") { + mapping, ok := current.(map[string]any) + if !ok { + return nil + } + current = mapping[part] + if current == nil { + return nil + } + } + return current +} + +func unwrapTraceValue(value any) any { + if values, ok := value.([]any); ok && len(values) == 1 { + return values[0] + } + return value +} + +func stringTraceValue(value any) string { + switch typed := unwrapTraceValue(value).(type) { + case nil: + return "" + case string: + return typed + case json.Number: + return typed.String() + case float64: + if math.Trunc(typed) == typed { + return strconv.FormatInt(int64(typed), 10) + } + return strconv.FormatFloat(typed, 'f', -1, 64) + case bool: + return strconv.FormatBool(typed) + default: + return fmt.Sprint(typed) + } +} + +func traceDurationMillis(value any, format string) float64 { + var number float64 + switch typed := unwrapTraceValue(value).(type) { + case float64: + number = typed + case int: + number = float64(typed) + case int64: + number = float64(typed) + case string: + number, _ = strconv.ParseFloat(typed, 64) + } + if format == "jaeger" { + return math.Round(number/10) / 100 + } + if number > 1_000_000 { + return math.Round(number/10_000) / 100 + } + return number +} + +func normalizeTraceTimestamp(raw string) string { + if raw == "" { + return "" + } + if number, err := strconv.ParseInt(raw, 10, 64); err == nil { + if number > 10_000_000_000_000 { + return time.Unix(0, number).Format(time.RFC3339Nano) + } + if number > 10_000_000_000 { + return time.UnixMilli(number).Format(time.RFC3339Nano) + } + return time.Unix(number, 0).Format(time.RFC3339Nano) + } + if timestamp, err := time.Parse(time.RFC3339Nano, raw); err == nil { + return timestamp.Format(time.RFC3339Nano) + } + return raw +} diff --git a/query/schema/connection.go b/query/schema/connection.go index 2654f8c..1967f6c 100644 --- a/query/schema/connection.go +++ b/query/schema/connection.go @@ -3,6 +3,7 @@ package schema import ( "encoding/json" "fmt" + "slices" "github.com/flanksource/clicky/rpc" "github.com/flanksource/commons-db/types" @@ -130,18 +131,35 @@ func tailoredBranch(typ string, cfg any) Schema { props := Schema{} propProps := Schema{} + propertyRequired := []string{} + required := stringSlice(flat["required"]) for name, raw := range flat["properties"].(map[string]any) { fs := Schema(raw.(map[string]any)) if key, ok := fs["x-clicky-property"].(string); ok { delete(fs, "x-clicky-property") + if typ == "opentelemetry" && key == "connection" { + fs["x-clicky-lookup"] = Schema{ + "url": "/api/v1/connection", "filter": "connection", "searchParam": "__lookup_q", "multi": false, + "scope": Schema{"param": "types", "from": "type", "map": map[string][]string{"opentelemetry": {"opensearch"}}}, + } + } propProps[key] = fs + if slices.Contains(required, name) { + propertyRequired = append(propertyRequired, key) + required = slices.DeleteFunc(required, func(value string) bool { return value == name }) + } continue } props[name] = fs } if len(propProps) > 0 { - props["properties"] = Schema{"type": "object", "title": "Properties", "properties": propProps} + properties := Schema{"type": "object", "title": "Properties", "properties": propProps} + if len(propertyRequired) > 0 { + properties["required"] = propertyRequired + required = append(required, "properties") + } + props["properties"] = properties } if isHTTPConnectionType(typ) { props["properties"] = httpAuthenticationSchema() @@ -151,8 +169,8 @@ func tailoredBranch(typ string, cfg any) Schema { if len(props) > 0 { then["properties"] = props } - if req := stringSlice(flat["required"]); len(req) > 0 { - then["required"] = req + if len(required) > 0 { + then["required"] = required } return Schema{ diff --git a/query/schema/connection_providers.go b/query/schema/connection_providers.go index 5d8fe16..a084daf 100644 --- a/query/schema/connection_providers.go +++ b/query/schema/connection_providers.go @@ -35,6 +35,12 @@ type HTTPProvider struct{ httpConnection } // OpenSearchProvider extends the HTTP form for an OpenSearch endpoint. type OpenSearchProvider struct{ httpConnection } +// OpenTelemetryProvider delegates trace storage to a nested OpenSearch +// connection while retaining its own first-class connection identity. +type OpenTelemetryProvider struct { + Connection string `json:"connection" clicky:"property=connection,title=OpenSearch Connection,required,order=2"` +} + // PrometheusProvider extends the HTTP form for a Prometheus endpoint. type PrometheusProvider struct{ httpConnection } @@ -115,21 +121,22 @@ type GitProvider struct { // (HTTP-family, SQL, cert/cloud). Kept in sync with allConnectionTypes by the // drift-guard test. var tailoredProviders = map[string]any{ - models.ConnectionTypeHTTP: HTTPProvider{}, - models.ConnectionTypeOpenSearch: OpenSearchProvider{}, - models.ConnectionTypePrometheus: PrometheusProvider{}, - models.ConnectionTypeLoki: LokiProvider{}, - models.ConnectionTypeJaeger: JaegerProvider{}, - models.ConnectionTypePostgres: PostgresProvider{}, - models.ConnectionTypeMySQL: MySQLProvider{}, - models.ConnectionTypeSQLServer: SQLServerProvider{}, - models.ConnectionTypeClickHouse: ClickHouseProvider{}, - models.ConnectionTypeRedis: RedisProvider{}, - models.ConnectionTypeKubernetes: KubernetesProvider{}, - models.ConnectionTypeGCP: GCPProvider{}, - models.ConnectionTypeGCS: GCSProvider{}, - models.ConnectionTypeGCPKMS: GCPKMSProvider{}, - models.ConnectionTypeGit: GitProvider{}, + models.ConnectionTypeHTTP: HTTPProvider{}, + models.ConnectionTypeOpenSearch: OpenSearchProvider{}, + models.ConnectionTypeOpenTelemetry: OpenTelemetryProvider{}, + models.ConnectionTypePrometheus: PrometheusProvider{}, + models.ConnectionTypeLoki: LokiProvider{}, + models.ConnectionTypeJaeger: JaegerProvider{}, + models.ConnectionTypePostgres: PostgresProvider{}, + models.ConnectionTypeMySQL: MySQLProvider{}, + models.ConnectionTypeSQLServer: SQLServerProvider{}, + models.ConnectionTypeClickHouse: ClickHouseProvider{}, + models.ConnectionTypeRedis: RedisProvider{}, + models.ConnectionTypeKubernetes: KubernetesProvider{}, + models.ConnectionTypeGCP: GCPProvider{}, + models.ConnectionTypeGCS: GCSProvider{}, + models.ConnectionTypeGCPKMS: GCPKMSProvider{}, + models.ConnectionTypeGit: GitProvider{}, } // TailoredProviderTypes returns the set of connection types that get a tailored diff --git a/query/schema/connection_types.go b/query/schema/connection_types.go index fd9f4b5..497b778 100644 --- a/query/schema/connection_types.go +++ b/query/schema/connection_types.go @@ -19,7 +19,7 @@ var allConnectionTypes = []string{ models.ConnectionTypeLoki, models.ConnectionTypeMatrix, models.ConnectionTypeMattermost, models.ConnectionTypeMongo, models.ConnectionTypeMySQL, models.ConnectionTypeNtfy, models.ConnectionTypeOllama, - models.ConnectionTypeOpenAI, models.ConnectionTypeOpenSearch, models.ConnectionTypeOpsGenie, + models.ConnectionTypeOpenAI, models.ConnectionTypeOpenTelemetry, models.ConnectionTypeOpenSearch, models.ConnectionTypeOpsGenie, models.ConnectionTypePostgres, models.ConnectionTypePrometheus, models.ConnectionTypePushbullet, models.ConnectionTypePushover, models.ConnectionTypeRedis, models.ConnectionTypeRestic, models.ConnectionTypeRocketchat, models.ConnectionTypeS3, models.ConnectionTypeSFTP, @@ -70,6 +70,7 @@ var connectionTypeIcons = map[string]string{ models.ConnectionTypeNtfy: "ntfy", models.ConnectionTypeOllama: "ollama", models.ConnectionTypeOpenAI: "openai", + models.ConnectionTypeOpenTelemetry: "opentelemetry", models.ConnectionTypeOpenSearch: "opensearch", models.ConnectionTypeOpsGenie: "opsgenie", models.ConnectionTypePostgres: "postgres", diff --git a/schemas/src/connection.json b/schemas/src/connection.json index b4bf9ac..38deb86 100644 --- a/schemas/src/connection.json +++ b/schemas/src/connection.json @@ -526,6 +526,21 @@ "$ref": "connections/openai.json" } }, + { + "if": { + "properties": { + "type": { + "const": "opentelemetry" + } + }, + "required": [ + "type" + ] + }, + "then": { + "$ref": "connections/opentelemetry.json" + } + }, { "if": { "properties": { @@ -884,6 +899,7 @@ "ntfy", "ollama", "openai", + "opentelemetry", "opensearch", "opsgenie", "postgres", @@ -945,6 +961,7 @@ "ollama": "ollama", "openai": "openai", "opensearch": "opensearch", + "opentelemetry": "opentelemetry", "opsgenie": "opsgenie", "postgres": "postgres", "prometheus": "prometheus", diff --git a/schemas/src/connections/opentelemetry.json b/schemas/src/connections/opentelemetry.json new file mode 100644 index 0000000..cdccf00 --- /dev/null +++ b/schemas/src/connections/opentelemetry.json @@ -0,0 +1,58 @@ +{ + "$id": "opentelemetry.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "properties": { + "name": { + "title": "Name", + "type": "string", + "x-clicky-order": 0 + }, + "namespace": { + "title": "Namespace", + "type": "string", + "x-clicky-component": "k8s-namespace-selector", + "x-clicky-order": 1 + }, + "properties": { + "properties": { + "connection": { + "title": "OpenSearch Connection", + "type": "string", + "x-clicky-lookup": { + "filter": "connection", + "multi": false, + "scope": { + "from": "type", + "map": { + "opentelemetry": [ + "opensearch" + ] + }, + "param": "types" + }, + "searchParam": "__lookup_q", + "url": "/api/v1/connection" + }, + "x-clicky-order": 2 + } + }, + "required": [ + "connection" + ], + "title": "Properties", + "type": "object" + }, + "type": { + "const": "opentelemetry", + "title": "Type", + "type": "string" + } + }, + "required": [ + "name", + "type", + "properties" + ], + "title": "Connection: opentelemetry", + "type": "object" +} From 4c9e63379560d08fe8a82799c126f8ec8d946757 Mon Sep 17 00:00:00 2001 From: Moshe Immerman Date: Fri, 17 Jul 2026 17:46:25 +0300 Subject: [PATCH 04/12] feat(query): Add profile inheritance and OpenTelemetry trace support Enable reusable profile composition with cycle detection, ordered row aliases, ignored fields, and derived columns. Add OpenTelemetry provider/schema support and migrate legacy trace profiles while preserving connections and imports. Pass validated parameters to providers and persist query application properties. --- cmd/query/internal/app/migrations/query.hcl | 39 +++ cmd/query/internal/app/migrations_test.go | 6 +- cmd/query/internal/app/schema_handler.go | 4 +- cmd/query/profiles/icons.go | 2 + cmd/query/profiles/legacy_migration.go | 302 ++++++++++++++++++++ cmd/query/profiles/legacy_migration_test.go | 41 +++ cmd/query/profiles/legacy_provider.go | 23 ++ cmd/query/profiles/openapi.go | 6 +- cmd/query/profiles/resolver.go | 169 +++++++++++ cmd/query/profiles/resolver_test.go | 78 +++++ cmd/query/profiles/service.go | 10 +- cmd/query/profiles/store.go | 26 +- cmd/query/profiles/store_test.go | 165 +++++++++++ cmd/query/profiles/suite_test.go | 13 + cmd/query/sessions/service.go | 7 +- cmd/query/sessions/top.go | 4 +- cmd/query/sessions/trace.go | 4 +- go.mod | 2 +- query/cel.go | 112 ++++++-- query/engine.go | 4 +- query/engine_test.go | 30 ++ query/iterator.go | 10 +- query/profile.go | 17 ++ query/provider.go | 4 + query/sample.go | 1 + query/schema/profile.go | 31 +- query/schema/profile_components.go | 31 +- query/schema/schema_test.go | 27 +- query/stream.go | 3 +- schemas/src/profile.json | 60 +++- schemas/src/profiles/clickhouse.json | 3 + schemas/src/profiles/http.json | 3 + schemas/src/profiles/jaeger.json | 3 + schemas/src/profiles/loki.json | 3 + schemas/src/profiles/mysql.json | 3 + schemas/src/profiles/opensearch.json | 3 + schemas/src/profiles/opentelemetry.json | 139 +++++++++ schemas/src/profiles/postgres.json | 3 + schemas/src/profiles/postgrest.json | 3 + schemas/src/profiles/prometheus.json | 3 + schemas/src/profiles/sql.json | 3 + schemas/src/profiles/sqlserver.json | 3 + 42 files changed, 1320 insertions(+), 83 deletions(-) create mode 100644 cmd/query/profiles/legacy_migration.go create mode 100644 cmd/query/profiles/legacy_migration_test.go create mode 100644 cmd/query/profiles/legacy_provider.go create mode 100644 cmd/query/profiles/resolver.go create mode 100644 cmd/query/profiles/resolver_test.go create mode 100644 cmd/query/profiles/suite_test.go create mode 100644 schemas/src/profiles/opentelemetry.json diff --git a/cmd/query/internal/app/migrations/query.hcl b/cmd/query/internal/app/migrations/query.hcl index b78bd7c..d287f8a 100644 --- a/cmd/query/internal/app/migrations/query.hcl +++ b/cmd/query/internal/app/migrations/query.hcl @@ -1,6 +1,45 @@ schema "public" { } +table "properties" { + schema = schema.public + + column "name" { + null = false + type = text + } + column "value" { + null = false + type = text + } + column "created_by" { + null = true + type = uuid + } + column "created_at" { + null = true + type = timestamptz + default = sql("now()") + } + column "updated_at" { + null = true + type = timestamptz + default = sql("now()") + } + column "deleted_at" { + null = true + type = timestamptz + } + + primary_key { + columns = [column.name] + } + + index "properties_created_by_idx" { + columns = [column.created_by] + } +} + table "connections" { schema = schema.public diff --git a/cmd/query/internal/app/migrations_test.go b/cmd/query/internal/app/migrations_test.go index 3c51ca4..e83646e 100644 --- a/cmd/query/internal/app/migrations_test.go +++ b/cmd/query/internal/app/migrations_test.go @@ -36,7 +36,7 @@ func TestMigrateSchemaAndProfileStore(t *testing.T) { require.NoError(t, migrateSchema(t.Context(), dsn)) require.NoError(t, migrateSchema(t.Context(), dsn), "migration must be idempotent") - for _, table := range []string{"connections", "profiles", "migration_api_unmanaged"} { + for _, table := range []string{"connections", "profiles", "properties", "migration_api_unmanaged"} { var exists bool require.NoError(t, gdb.Raw(`SELECT EXISTS ( SELECT 1 FROM information_schema.tables @@ -47,6 +47,10 @@ func TestMigrateSchemaAndProfileStore(t *testing.T) { var existing models.Connection require.NoError(t, gdb.Where("name = ?", "existing").First(&existing).Error) require.Equal(t, models.ConnectionTypePostgres, existing.Type) + require.NoError(t, gdb.Create(&models.AppProperty{Name: "query.page-size", Value: "100"}).Error) + var property models.AppProperty + require.NoError(t, gdb.First(&property, "name = ?", "query.page-size").Error) + require.Equal(t, "100", property.Value) dir := t.TempDir() files, err := profiles.NewFileStore(dir) diff --git a/cmd/query/internal/app/schema_handler.go b/cmd/query/internal/app/schema_handler.go index 1f5fe2f..4f0b171 100644 --- a/cmd/query/internal/app/schema_handler.go +++ b/cmd/query/internal/app/schema_handler.go @@ -77,11 +77,11 @@ func (h *schemaHandler) resolve(path string) (schema.Schema, bool, error) { if name == "" || strings.Contains(name, "/") { return nil, false, nil } - p, err := h.store.Get(context.Background(), name) + resolved, err := profiles.Resolve(context.Background(), h.store, name) if err != nil { return nil, true, err } - return schema.ProfileInstance(p), true, nil + return schema.ProfileInstance(resolved.Profile), true, nil default: return nil, false, nil } diff --git a/cmd/query/profiles/icons.go b/cmd/query/profiles/icons.go index 12059cc..736f9e3 100644 --- a/cmd/query/profiles/icons.go +++ b/cmd/query/profiles/icons.go @@ -16,6 +16,8 @@ func providerIcon(providerType string) string { return "activity" case "opensearch": return "globe" + case "opentelemetry": + return "activity" default: return "table" } diff --git a/cmd/query/profiles/legacy_migration.go b/cmd/query/profiles/legacy_migration.go new file mode 100644 index 0000000..668b702 --- /dev/null +++ b/cmd/query/profiles/legacy_migration.go @@ -0,0 +1,302 @@ +package profiles + +import ( + "fmt" + "os" + "path/filepath" + "slices" + "strings" + + "github.com/flanksource/commons-db/query" + yamlv3 "go.yaml.in/yaml/v3" + "sigs.k8s.io/yaml" +) + +const legacyTraceProvider = "legacy-trace" + +type legacyTraceParam struct { + Field string `json:"field,omitempty" yaml:"field,omitempty"` + Operator string `json:"operator,omitempty" yaml:"operator,omitempty"` + Format string `json:"format,omitempty" yaml:"format,omitempty"` + Template string `json:"template,omitempty" yaml:"template,omitempty"` + Clause string `json:"clause,omitempty" yaml:"clause,omitempty"` + Description string `json:"description,omitempty" yaml:"description,omitempty"` + Internal bool `json:"internal,omitempty" yaml:"internal,omitempty"` + Required bool `json:"required,omitempty" yaml:"required,omitempty"` +} + +type legacyTraceColumn struct { + Name string `json:"name" yaml:"name"` + Field string `json:"field" yaml:"field"` + Detail bool `json:"detail,omitempty" yaml:"detail,omitempty"` +} + +type legacyTraceProfile struct { + Name string `json:"name" yaml:"name"` + Kind string `json:"kind,omitempty" yaml:"kind,omitempty"` + Format string `json:"format,omitempty" yaml:"format,omitempty"` + Index string `json:"index,omitempty" yaml:"index,omitempty"` + DateField string `json:"dateField,omitempty" yaml:"dateField,omitempty"` + TraceIDField string `json:"traceIdField,omitempty" yaml:"traceIdField,omitempty"` + SpanIDField string `json:"spanIdField,omitempty" yaml:"spanIdField,omitempty"` + ParentIDField string `json:"parentIdField,omitempty" yaml:"parentIdField,omitempty"` + ParentRefType string `json:"parentRefType,omitempty" yaml:"parentRefType,omitempty"` + ServiceField string `json:"serviceField,omitempty" yaml:"serviceField,omitempty"` + OperationField string `json:"operationField,omitempty" yaml:"operationField,omitempty"` + StatusFields []string `json:"statusFields,omitempty" yaml:"statusFields,omitempty"` + SelectFields []string `json:"selectFields,omitempty" yaml:"selectFields,omitempty"` + SourceExcludes []string `json:"sourceExcludes,omitempty" yaml:"sourceExcludes,omitempty"` + Imports []string `json:"imports,omitempty" yaml:"imports,omitempty"` + Defaults map[string]any `json:"defaults,omitempty" yaml:"defaults,omitempty"` + Params map[string]legacyTraceParam `json:"params,omitempty" yaml:"params,omitempty"` + Aliases orderedLegacyAliases `json:"-" yaml:"aliases,omitempty"` + Ignore []string `json:"ignore,omitempty" yaml:"ignore,omitempty"` + Columns []legacyTraceColumn `json:"columns,omitempty" yaml:"columns,omitempty"` + SQL map[string]any `json:"sql,omitempty" yaml:"sql,omitempty"` + Kubernetes map[string]any `json:"kubernetes,omitempty" yaml:"kubernetes,omitempty"` + Arthas map[string]any `json:"arthas,omitempty" yaml:"arthas,omitempty"` + Replay map[string]any `json:"replay,omitempty" yaml:"replay,omitempty"` +} + +type orderedLegacyAliases []query.AliasDef + +func (a *orderedLegacyAliases) UnmarshalYAML(node *yamlv3.Node) error { + if node.Kind != yamlv3.MappingNode { + return fmt.Errorf("aliases must be a mapping") + } + for index := 0; index < len(node.Content); index += 2 { + var value struct { + CEL string `yaml:"cel"` + } + if err := node.Content[index+1].Decode(&value); err != nil { + return err + } + *a = append(*a, query.AliasDef{Name: node.Content[index].Value, CEL: value.CEL}) + } + return nil +} + +type profileMigration struct { + path string + data []byte +} + +func (s *FileStore) migrateLegacyTraceProfiles() error { + entries, err := os.ReadDir(s.Dir) + if err != nil { + return fmt.Errorf("read profiles dir %q for migration: %w", s.Dir, err) + } + migrations := make([]profileMigration, 0) + for _, entry := range entries { + if entry.IsDir() || !isYAML(entry.Name()) { + continue + } + path := filepath.Join(s.Dir, entry.Name()) + migration, ok, err := legacyProfileMigration(path) + if err != nil { + return err + } + if ok { + migrations = append(migrations, migration) + } + } + for _, migration := range migrations { + if err := replaceProfileFile(migration); err != nil { + return err + } + } + return nil +} + +func legacyProfileMigration(path string) (profileMigration, bool, error) { + data, err := os.ReadFile(path) + if err != nil { + return profileMigration{}, false, fmt.Errorf("read profile %q for migration: %w", path, err) + } + var identity struct { + Profile string `json:"profile"` + Name string `json:"name"` + Provider query.ProviderConfig `json:"provider"` + } + if err := yaml.Unmarshal(data, &identity); err != nil { + return profileMigration{}, false, fmt.Errorf("parse profile %q for migration: %w", path, err) + } + if identity.Profile != "" { + if identity.Provider.Type != legacyTraceProvider { + return profileMigration{}, false, nil + } + var current query.Profile + if err := yaml.Unmarshal(data, ¤t); err != nil { + return profileMigration{}, false, fmt.Errorf("parse interim trace profile %q: %w", path, err) + } + source, _ := current.Provider.Options["source"].(string) + kind, _ := current.Provider.Options["kind"].(string) + if source == "" || kind != "opensearch" && kind != "import" { + return profileMigration{}, false, nil + } + converted, err := convertLegacyTraceSource(source) + if err != nil { + return profileMigration{}, false, fmt.Errorf("convert interim trace profile %q: %w", path, err) + } + converted.Provider.Connection = current.Provider.Connection + data, err := yaml.Marshal(converted) + if err != nil { + return profileMigration{}, false, fmt.Errorf("marshal migrated trace profile %q: %w", path, err) + } + return profileMigration{path: path, data: data}, true, nil + } + if identity.Name == "" { + return profileMigration{}, false, nil + } + profile, err := convertLegacyTraceSource(string(data)) + if err != nil { + return profileMigration{}, false, fmt.Errorf("convert legacy trace profile %q: %w", path, err) + } + converted, err := yaml.Marshal(profile) + if err != nil { + return profileMigration{}, false, fmt.Errorf("marshal migrated trace profile %q: %w", path, err) + } + return profileMigration{path: path, data: converted}, true, nil +} + +func convertLegacyTraceSource(source string) (query.Profile, error) { + var legacy legacyTraceProfile + if err := yamlv3.Unmarshal([]byte(source), &legacy); err != nil { + return query.Profile{}, fmt.Errorf("parse legacy trace source: %w", err) + } + return convertLegacyTraceProfile(legacy, source) +} + +func convertLegacyTraceProfile(legacy legacyTraceProfile, source string) (query.Profile, error) { + if strings.TrimSpace(legacy.Name) == "" { + return query.Profile{}, fmt.Errorf("name is required") + } + params := make([]query.ParamDef, 0, len(legacy.Params)) + for _, name := range sortedKeys(legacy.Params) { + param := legacy.Params[name] + def := query.ParamDef{ + Name: name, Description: param.Description, Required: param.Required, Template: param.Template, + } + if value, ok := legacy.Defaults[name]; ok { + def.Default = value + def.Type = legacyParamType(value) + } + params = append(params, def) + } + columns := make([]query.ColumnDef, len(legacy.Columns)) + for i, column := range legacy.Columns { + columns[i] = query.ColumnDef{ + Name: column.Name, CEL: column.Field, Hidden: column.Detail, + } + } + provider := query.ProviderConfig{Type: legacyTraceProvider, Options: map[string]any{ + "kind": legacyTraceKind(legacy), "source": source, + }} + if legacyTraceKind(legacy) == "opensearch" { + provider = query.ProviderConfig{Type: "opentelemetry", Options: legacyOpenTelemetryOptions(legacy)} + } else if legacyTraceKind(legacy) == "import" { + provider = query.ProviderConfig{Type: "opentelemetry", Options: map[string]any{"params": legacyProviderParams(legacy.Params)}} + } + return query.Profile{ + Name: legacy.Name, Imports: legacy.Imports, Provider: provider, + Params: params, Columns: columns, Aliases: []query.AliasDef(legacy.Aliases), Ignore: legacy.Ignore, + }, nil +} + +func legacyOpenTelemetryOptions(legacy legacyTraceProfile) map[string]any { + options := map[string]any{ + "format": legacy.Format, "index": legacy.Index, "dateField": legacy.DateField, + "traceIdField": legacy.TraceIDField, "spanIdField": legacy.SpanIDField, + "parentIdField": legacy.ParentIDField, "parentRefType": legacy.ParentRefType, + "serviceField": legacy.ServiceField, "operationField": legacy.OperationField, + "statusFields": legacy.StatusFields, "selectFields": legacy.SelectFields, + "sourceExcludes": legacy.SourceExcludes, "params": legacyProviderParams(legacy.Params), + } + for key, value := range options { + switch typed := value.(type) { + case string: + if typed == "" { + delete(options, key) + } + case []string: + if len(typed) == 0 { + delete(options, key) + } + case map[string]any: + if len(typed) == 0 { + delete(options, key) + } + } + } + return options +} + +func legacyProviderParams(params map[string]legacyTraceParam) map[string]any { + converted := make(map[string]any, len(params)) + for name, param := range params { + converted[name] = map[string]any{ + "field": param.Field, "operator": param.Operator, "format": param.Format, + "template": param.Template, "clause": param.Clause, "internal": param.Internal, + } + for key, value := range converted[name].(map[string]any) { + if value == "" || value == false { + delete(converted[name].(map[string]any), key) + } + } + } + return converted +} + +func legacyTraceKind(profile legacyTraceProfile) string { + if profile.Kind != "" { + return profile.Kind + } + switch { + case profile.SQL != nil: + return "sql" + case profile.Kubernetes != nil: + return "kubernetes" + case profile.Arthas != nil: + return "arthas" + case profile.Index != "" || profile.Format != "": + return "opensearch" + case profile.Replay != nil: + return "replay" + case len(profile.Imports) > 0: + return "import" + default: + return "unknown" + } +} + +func legacyParamType(value any) query.ParamType { + switch value.(type) { + case bool: + return query.ParamTypeBoolean + case float32, float64, int, int32, int64: + return query.ParamTypeNumber + default: + return query.ParamTypeString + } +} + +func replaceProfileFile(migration profileMigration) error { + temporary := migration.path + ".migrating" + if err := os.WriteFile(temporary, migration.data, 0o600); err != nil { + return fmt.Errorf("write migrated profile %q: %w", migration.path, err) + } + if err := os.Rename(temporary, migration.path); err != nil { + _ = os.Remove(temporary) + return fmt.Errorf("replace migrated profile %q: %w", migration.path, err) + } + return nil +} + +func sortedKeys[V any](values map[string]V) []string { + keys := make([]string, 0, len(values)) + for key := range values { + keys = append(keys, key) + } + slices.Sort(keys) + return keys +} diff --git a/cmd/query/profiles/legacy_migration_test.go b/cmd/query/profiles/legacy_migration_test.go new file mode 100644 index 0000000..1ea2254 --- /dev/null +++ b/cmd/query/profiles/legacy_migration_test.go @@ -0,0 +1,41 @@ +package profiles + +import ( + "testing" + + dbcontext "github.com/flanksource/commons-db/context" + "github.com/flanksource/commons-db/query" +) + +func TestLegacyTraceKind(t *testing.T) { + tests := []struct { + name string + profile legacyTraceProfile + want string + }{ + {name: "explicit", profile: legacyTraceProfile{Kind: "watch"}, want: "watch"}, + {name: "sql", profile: legacyTraceProfile{SQL: map[string]any{}}, want: "sql"}, + {name: "kubernetes", profile: legacyTraceProfile{Kubernetes: map[string]any{}}, want: "kubernetes"}, + {name: "arthas", profile: legacyTraceProfile{Arthas: map[string]any{}}, want: "arthas"}, + {name: "opensearch", profile: legacyTraceProfile{Index: "traces-*"}, want: "opensearch"}, + {name: "replay", profile: legacyTraceProfile{Replay: map[string]any{}}, want: "replay"}, + {name: "import", profile: legacyTraceProfile{Imports: []string{"base"}}, want: "import"}, + {name: "unknown", profile: legacyTraceProfile{}, want: "unknown"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := legacyTraceKind(tt.profile); got != tt.want { + t.Fatalf("legacyTraceKind() = %q, want %q", got, tt.want) + } + }) + } +} + +func TestLegacyTraceProviderFailsLoudly(t *testing.T) { + _, err := legacyTraceProfileProvider{}.Execute(dbcontext.New(), query.ProviderRequest{ + Options: map[string]any{"kind": "sql"}, + }) + if err == nil || err.Error() != `legacy trace profile kind "sql" is catalog-compatible but is not executable by the query engine` { + t.Fatalf("unexpected legacy provider error: %v", err) + } +} diff --git a/cmd/query/profiles/legacy_provider.go b/cmd/query/profiles/legacy_provider.go new file mode 100644 index 0000000..39ba4eb --- /dev/null +++ b/cmd/query/profiles/legacy_provider.go @@ -0,0 +1,23 @@ +package profiles + +import ( + "fmt" + + dbcontext "github.com/flanksource/commons-db/context" + "github.com/flanksource/commons-db/query" +) + +type legacyTraceProfileProvider struct{} + +func (legacyTraceProfileProvider) Type() string { return legacyTraceProvider } + +func (legacyTraceProfileProvider) Execute(_ dbcontext.Context, request query.ProviderRequest) ([]query.Row, error) { + return nil, fmt.Errorf( + "legacy trace profile kind %q is catalog-compatible but is not executable by the query engine", + request.Options["kind"], + ) +} + +func init() { + query.RegisterProvider(legacyTraceProfileProvider{}) +} diff --git a/cmd/query/profiles/openapi.go b/cmd/query/profiles/openapi.go index 9a2e6bc..be6be1f 100644 --- a/cmd/query/profiles/openapi.go +++ b/cmd/query/profiles/openapi.go @@ -83,7 +83,11 @@ func mergeStoredProfiles(spec *rpc.OpenAPISpec, store Store) error { return fmt.Errorf("load profile surfaces: %w", err) } for _, profile := range profiles { - addProfileToSpec(spec, profile) + resolved, err := Resolve(context.Background(), store, profile.Name) + if err != nil { + return fmt.Errorf("resolve profile surface %q: %w", profile.Name, err) + } + addProfileToSpec(spec, resolved.Profile) } return nil } diff --git a/cmd/query/profiles/resolver.go b/cmd/query/profiles/resolver.go new file mode 100644 index 0000000..6dcbb01 --- /dev/null +++ b/cmd/query/profiles/resolver.go @@ -0,0 +1,169 @@ +package profiles + +import ( + "context" + "fmt" + "slices" + "strings" + + "github.com/flanksource/commons-db/query" +) + +type ResolvedProfile struct { + Profile query.Profile + ConnectionProfile string +} + +func Resolve(ctx context.Context, store Store, name string) (ResolvedProfile, error) { + return resolve(ctx, store, name, nil) +} + +func resolve(ctx context.Context, store Store, name string, path []string) (ResolvedProfile, error) { + if index := slices.Index(path, name); index >= 0 { + return ResolvedProfile{}, fmt.Errorf("profile import cycle: %s", strings.Join(append(path[index:], name), " -> ")) + } + current, err := store.Get(ctx, name) + if err != nil { + return ResolvedProfile{}, fmt.Errorf("resolve profile %q: %w", name, err) + } + if current.Name == "" { + return ResolvedProfile{}, fmt.Errorf("resolve profile %q: profile not found", name) + } + + var result ResolvedProfile + for _, importedName := range current.Imports { + imported, err := resolve(ctx, store, importedName, append(path, name)) + if err != nil { + return ResolvedProfile{}, fmt.Errorf("profile %q imports %q: %w", current.Name, importedName, err) + } + result.Profile = mergeProfile(result.Profile, imported.Profile) + if imported.Profile.Provider.Type != "" { + result.ConnectionProfile = imported.ConnectionProfile + } + } + + previousType := result.Profile.Provider.Type + result.Profile = mergeProfile(result.Profile, current) + result.Profile.Name = current.Name + result.Profile.Imports = nil + if current.Provider.Connection != "" || current.Provider.Type != "" && current.Provider.Type != previousType { + result.ConnectionProfile = current.Name + } + if result.ConnectionProfile == "" && result.Profile.Provider.Type != "" { + result.ConnectionProfile = current.Name + } + return result, nil +} + +func mergeProfile(base, overlay query.Profile) query.Profile { + merged := base + if overlay.Name != "" { + merged.Name = overlay.Name + } + if overlay.Namespace != "" { + merged.Namespace = overlay.Namespace + } + if overlay.Provider.Type != "" { + merged.Provider.Type = overlay.Provider.Type + } + if overlay.Provider.Connection != "" { + merged.Provider.Connection = overlay.Provider.Connection + } + merged.Provider.Options = mergeMap(merged.Provider.Options, overlay.Provider.Options) + if overlay.Query != "" { + merged.Query = overlay.Query + } + merged.Params = mergeParams(merged.Params, overlay.Params) + if len(overlay.Columns) > 0 { + merged.Columns = slices.Clone(overlay.Columns) + } + if len(overlay.Aliases) > 0 { + merged.Aliases = slices.Clone(overlay.Aliases) + } + if len(overlay.Ignore) > 0 { + merged.Ignore = slices.Clone(overlay.Ignore) + } + if len(overlay.Processors) > 0 { + merged.Processors = slices.Clone(overlay.Processors) + } + if len(overlay.Context) > 0 { + if merged.Context == nil { + merged.Context = map[string]query.SubQuery{} + } + for name, subquery := range overlay.Context { + merged.Context[name] = subquery + } + } + if len(overlay.Output) > 0 { + merged.Output = slices.Clone(overlay.Output) + } + if overlay.Render != "" { + merged.Render = overlay.Render + } + if overlay.Trace != nil { + merged.Trace = overlay.Trace + } + if overlay.Top != nil { + merged.Top = overlay.Top + } + return merged +} + +func mergeMap(base, overlay map[string]any) map[string]any { + if len(base) == 0 && len(overlay) == 0 { + return nil + } + merged := make(map[string]any, len(base)+len(overlay)) + for key, value := range base { + merged[key] = value + } + for key, value := range overlay { + baseMap, baseOK := merged[key].(map[string]any) + overlayMap, overlayOK := value.(map[string]any) + if baseOK && overlayOK { + merged[key] = mergeMap(baseMap, overlayMap) + continue + } + merged[key] = value + } + return merged +} + +func mergeParams(base, overlay []query.ParamDef) []query.ParamDef { + merged := slices.Clone(base) + for _, incoming := range overlay { + index := slices.IndexFunc(merged, func(existing query.ParamDef) bool { return existing.Name == incoming.Name }) + if index < 0 { + merged = append(merged, incoming) + continue + } + merged[index] = mergeParam(merged[index], incoming) + } + return merged +} + +func mergeParam(base, overlay query.ParamDef) query.ParamDef { + if overlay.Label != "" { + base.Label = overlay.Label + } + if overlay.Type != "" { + base.Type = overlay.Type + } + if overlay.Role != "" { + base.Role = overlay.Role + } + if overlay.Default != nil { + base.Default = overlay.Default + } + if len(overlay.Options) > 0 { + base.Options = slices.Clone(overlay.Options) + } + base.Required = base.Required || overlay.Required + if overlay.Description != "" { + base.Description = overlay.Description + } + if overlay.Template != "" { + base.Template = overlay.Template + } + return base +} diff --git a/cmd/query/profiles/resolver_test.go b/cmd/query/profiles/resolver_test.go new file mode 100644 index 0000000..96b14c6 --- /dev/null +++ b/cmd/query/profiles/resolver_test.go @@ -0,0 +1,78 @@ +package profiles + +import ( + "context" + + "github.com/flanksource/commons-db/query" + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +type resolverStore map[string]query.Profile + +func (s resolverStore) List(context.Context) ([]query.Profile, error) { + profiles := make([]query.Profile, 0, len(s)) + for _, profile := range s { + profiles = append(profiles, profile) + } + return profiles, nil +} + +func (s resolverStore) Get(_ context.Context, name string) (query.Profile, error) { + return s[name], nil +} + +func (s resolverStore) Save(_ context.Context, profile query.Profile) error { + s[profile.Name] = profile + return nil +} + +func (s resolverStore) Delete(_ context.Context, name string) error { + delete(s, name) + return nil +} + +var _ = Describe("Resolve", func() { + It("merges imports left to right and reports the profile that owns the connection", func() { + store := resolverStore{ + "jaeger": { + Name: "jaeger", + Provider: query.ProviderConfig{Type: "opentelemetry", Options: map[string]any{ + "format": "jaeger", "params": map[string]any{"namespace": map[string]any{"field": "namespace"}}, + }}, + Params: []query.ParamDef{{Name: "namespace", Description: "base namespace"}}, + Aliases: []query.AliasDef{{Name: "service", CEL: `span["service.name"]`}}, + }, + "jms": { + Name: "jms", + Imports: []string{"jaeger"}, + Provider: query.ProviderConfig{Options: map[string]any{ + "params": map[string]any{"namespace": map[string]any{"template": "{value}-api"}}, + }}, + Params: []query.ParamDef{{Name: "namespace", Required: true}}, + Ignore: []string{"internal"}, + }, + } + + resolved, err := Resolve(context.Background(), store, "jms") + Expect(err).ToNot(HaveOccurred()) + Expect(resolved.ConnectionProfile).To(Equal("jaeger")) + Expect(resolved.Profile.Name).To(Equal("jms")) + Expect(resolved.Profile.Imports).To(BeEmpty()) + Expect(resolved.Profile.Provider.Type).To(Equal("opentelemetry")) + Expect(resolved.Profile.Provider.Options).To(HaveKey("params")) + Expect(resolved.Profile.Params).To(Equal([]query.ParamDef{{Name: "namespace", Description: "base namespace", Required: true}})) + Expect(resolved.Profile.Aliases).To(HaveLen(1)) + Expect(resolved.Profile.Ignore).To(Equal([]string{"internal"})) + }) + + It("rejects cycles with the complete import path", func() { + store := resolverStore{ + "a": {Name: "a", Imports: []string{"b"}}, + "b": {Name: "b", Imports: []string{"a"}}, + } + + _, err := Resolve(context.Background(), store, "a") + Expect(err).To(MatchError(ContainSubstring("a -> b -> a"))) + }) +}) diff --git a/cmd/query/profiles/service.go b/cmd/query/profiles/service.go index 4cdef84..a16566c 100644 --- a/cmd/query/profiles/service.go +++ b/cmd/query/profiles/service.go @@ -191,7 +191,11 @@ func (s *Service) RegisterDynamic(ctx context.Context) error { } for _, p := range profiles { name := p.Name - schemaJSON, err := profileEntitySchema(p) + resolved, err := Resolve(ctx, store, name) + if err != nil { + return err + } + schemaJSON, err := profileEntitySchema(resolved.Profile) if err != nil { return fmt.Errorf("build entity schema for profile %q: %w", name, err) } @@ -204,14 +208,14 @@ func (s *Service) RegisterDynamic(ctx context.Context) error { if err != nil { return nil, err } - live, err := store.Get(context.Background(), name) + live, err := Resolve(context.Background(), store, name) if err != nil { return nil, err } // The base profile flow needs no database; only postgres/sqlite // processors do. The context provider supplies the DB-backed // context under `serve` and a DB-less one on the CLI. - res, err := query.Execute(s.context(), live, toParams(opts)) + res, err := query.Execute(s.context(), live.Profile, toParams(opts)) if err != nil { return nil, err } diff --git a/cmd/query/profiles/store.go b/cmd/query/profiles/store.go index 72f43f2..78895f5 100644 --- a/cmd/query/profiles/store.go +++ b/cmd/query/profiles/store.go @@ -3,6 +3,7 @@ package profiles import ( "context" "encoding/json" + "errors" "fmt" "os" "path/filepath" @@ -51,7 +52,11 @@ func NewFileStore(dir string) (*FileStore, error) { if err := ensurePrivateDir(abs); err != nil { return nil, fmt.Errorf("create profiles dir %q: %w", abs, err) } - return &FileStore{Dir: abs}, nil + store := &FileStore{Dir: abs} + if err := store.migrateLegacyTraceProfiles(); err != nil { + return nil, err + } + return store, nil } func NewDBStore(db *gorm.DB) (*DBStore, error) { @@ -73,7 +78,24 @@ func Import(ctx context.Context, source Store, target *DBStore) error { return err } for _, profile := range items { - if err := target.save(ctx, profile, false); err != nil { + update := false + var existing profileRecord + err := target.db.WithContext(ctx).Where("name = ?", profile.Name).First(&existing).Error + if err == nil { + stored, decodeErr := existing.profile() + if decodeErr != nil { + return decodeErr + } + if stored.Provider.Type == legacyTraceProvider && profile.Provider.Type != legacyTraceProvider { + if stored.Provider.Connection != "" { + profile.Provider.Connection = stored.Provider.Connection + } + update = true + } + } else if !errors.Is(err, gorm.ErrRecordNotFound) { + return fmt.Errorf("inspect imported profile %q: %w", profile.Name, err) + } + if err := target.save(ctx, profile, update); err != nil { return fmt.Errorf("import YAML profile %q: %w", profile.Name, err) } } diff --git a/cmd/query/profiles/store_test.go b/cmd/query/profiles/store_test.go index 5feae67..7dedcb8 100644 --- a/cmd/query/profiles/store_test.go +++ b/cmd/query/profiles/store_test.go @@ -7,6 +7,7 @@ import ( "testing" "github.com/flanksource/commons-db/query" + "sigs.k8s.io/yaml" ) func sampleProfile(name string) query.Profile { @@ -84,6 +85,170 @@ func TestProfileStoreListAndDelete(t *testing.T) { } } +func TestProfileStoreMigratesLegacyTraceProfilesOnStartup(t *testing.T) { + store, legacy := migratedActivityProfileStore(t) + list, err := store.List(context.Background()) + if err != nil { + t.Fatalf("List: %v", err) + } + if got := names(list); len(got) != 2 || got[0] != "Current Profile" || got[1] != "activity.client" { + t.Fatalf("expected migrated and current query profiles, got %v", got) + } + migrated := list[1] + if migrated.Provider.Type != "legacy-trace" || migrated.Provider.Options["kind"] != "sql" { + t.Fatalf("unexpected migrated provider: %+v", migrated.Provider) + } + if source, ok := migrated.Provider.Options["source"].(string); !ok || source != string(legacy) { + t.Fatalf("legacy source was not preserved: %#v", migrated.Provider.Options["source"]) + } + if len(migrated.Params) != 1 || migrated.Params[0].Name != "activity" || !migrated.Params[0].Required { + t.Fatalf("legacy params were not converted: %+v", migrated.Params) + } + if len(migrated.Columns) != 1 || migrated.Columns[0].Name != "ClientGUID" || migrated.Columns[0].CEL != "span.ClientGUID" { + t.Fatalf("legacy columns were not converted: %+v", migrated.Columns) + } +} + +func TestProfileStoreUpgradesInterimTraceImportsToExecutableProfiles(t *testing.T) { + dir := t.TempDir() + jaeger := []byte(`profile: jaeger +provider: + type: legacy-trace + options: + kind: opensearch + source: | + name: jaeger + format: jaeger + index: jaeger-span* + params: + namespace: + field: process.serviceName + operator: term +`) + jms := []byte(`profile: jms +provider: + type: legacy-trace + options: + kind: import + source: | + name: jms + imports: [jaeger] + params: + namespace: + field: process.serviceName + template: "{value}-api" + aliases: + request.xml: + cel: span["input.xml"] + request.copy: + cel: request.xml + ignore: [input.xml] +`) + for name, data := range map[string][]byte{"jaeger.yaml": jaeger, "jms.yaml": jms} { + if err := os.WriteFile(filepath.Join(dir, name), data, 0o600); err != nil { + t.Fatal(err) + } + } + store, err := NewFileStore(dir) + if err != nil { + t.Fatal(err) + } + resolved, err := Resolve(context.Background(), store, "jms") + if err != nil { + t.Fatal(err) + } + if resolved.ConnectionProfile != "jaeger" || resolved.Profile.Provider.Type != "opentelemetry" { + t.Fatalf("unexpected resolved profile: %+v", resolved) + } + if resolved.Profile.Provider.Options["index"] != "jaeger-span*" { + t.Fatalf("opentelemetry options not inherited: %#v", resolved.Profile.Provider.Options) + } + if len(resolved.Profile.Aliases) != 2 || resolved.Profile.Aliases[0].Name != "request.xml" || resolved.Profile.Aliases[1].Name != "request.copy" { + t.Fatalf("alias order not preserved: %+v", resolved.Profile.Aliases) + } +} + +func TestProfileStoreLegacyMigrationRewritesOnce(t *testing.T) { + store, _ := migratedActivityProfileStore(t) + path := filepath.Join(store.Dir, "activity.client.yaml") + data, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read migrated profile: %v", err) + } + var identity struct { + Profile string `json:"profile"` + Name string `json:"name"` + } + if err := yaml.Unmarshal(data, &identity); err != nil { + t.Fatalf("parse migrated profile: %v", err) + } + if identity.Profile != "activity.client" || identity.Name != "" { + t.Fatalf("profile was not rewritten to the current identity: %+v", identity) + } + if _, err := NewFileStore(store.Dir); err != nil { + t.Fatalf("second restart should leave migrated profiles unchanged: %v", err) + } + after, err := os.ReadFile(path) + if err != nil { + t.Fatalf("read profile after second restart: %v", err) + } + if string(after) != string(data) { + t.Fatal("second startup rewrote an already migrated profile") + } +} + +func migratedActivityProfileStore(t *testing.T) (*FileStore, []byte) { + t.Helper() + dir := t.TempDir() + store, err := NewFileStore(dir) + if err != nil { + t.Fatalf("NewProfileStore: %v", err) + } + if err := store.Save(context.Background(), sampleProfile("Current Profile")); err != nil { + t.Fatalf("Save: %v", err) + } + legacy := []byte(`name: activity.client +sql: + root: AsClientActivity +params: + activity: + column: AsClientActivity.ActivityGuid + operator: terms + required: true +columns: + - name: ClientGUID + field: span.ClientGUID +`) + if err := os.WriteFile(filepath.Join(dir, "activity.client.yaml"), legacy, 0o600); err != nil { + t.Fatalf("write legacy trace profile: %v", err) + } + store, err = NewFileStore(dir) + if err != nil { + t.Fatalf("restart profile store: %v", err) + } + return store, legacy +} + +func TestProfileStoreListRejectsMalformedQueryProfile(t *testing.T) { + dir := t.TempDir() + store, err := NewFileStore(dir) + if err != nil { + t.Fatalf("NewProfileStore: %v", err) + } + malformed := []byte(`profile: Broken +params: + region: + required: true +`) + if err := os.WriteFile(filepath.Join(dir, "broken.yaml"), malformed, 0o600); err != nil { + t.Fatalf("write malformed query profile: %v", err) + } + + if _, err := store.List(context.Background()); err == nil { + t.Fatal("expected malformed current query profile to fail listing") + } +} + func TestProfileStoreSaveRequiresName(t *testing.T) { store, _ := NewFileStore(t.TempDir()) if err := store.Save(context.Background(), query.Profile{}); err == nil { diff --git a/cmd/query/profiles/suite_test.go b/cmd/query/profiles/suite_test.go new file mode 100644 index 0000000..c008519 --- /dev/null +++ b/cmd/query/profiles/suite_test.go @@ -0,0 +1,13 @@ +package profiles + +import ( + "testing" + + . "github.com/onsi/ginkgo/v2" + . "github.com/onsi/gomega" +) + +func TestProfiles(t *testing.T) { + RegisterFailHandler(Fail) + RunSpecs(t, "Profiles Suite") +} diff --git a/cmd/query/sessions/service.go b/cmd/query/sessions/service.go index bee74f8..6c89851 100644 --- a/cmd/query/sessions/service.go +++ b/cmd/query/sessions/service.go @@ -115,11 +115,12 @@ func (h *sessionHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { } func (h *sessionHandler) start(w http.ResponseWriter, r *http.Request, name string) { - p, err := h.store.Get(r.Context(), name) + resolved, err := profiles.Resolve(r.Context(), h.store, name) if err != nil { http.Error(w, err.Error(), http.StatusNotFound) return } + p := resolved.Profile if p, err = applySessionSpecOverrides(p, r.URL.Query().Get("interval"), r.URL.Query().Get("duration")); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return @@ -322,7 +323,7 @@ func (h *sessionHandler) result(w http.ResponseWriter, r *http.Request, id strin if !ok { return } - p, err := h.store.Get(r.Context(), info.Profile) + resolved, err := profiles.Resolve(r.Context(), h.store, info.Profile) if err != nil { http.Error(w, fmt.Sprintf("session %q: %v", id, err), http.StatusNotFound) return @@ -332,7 +333,7 @@ func (h *sessionHandler) result(w http.ResponseWriter, r *http.Request, id strin http.Error(w, err.Error(), http.StatusInternalServerError) return } - if result, err = query.MaterializeEvents(h.ctx, p, events); err != nil { + if result, err = query.MaterializeEvents(h.ctx, resolved.Profile, events); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } diff --git a/cmd/query/sessions/top.go b/cmd/query/sessions/top.go index a1beb83..e931405 100644 --- a/cmd/query/sessions/top.go +++ b/cmd/query/sessions/top.go @@ -8,6 +8,7 @@ import ( "github.com/flanksource/clicky/api" "github.com/flanksource/clicky/task" + "github.com/flanksource/commons-db/cmd/query/profiles" "github.com/flanksource/commons-db/query" commonsContext "github.com/flanksource/commons/context" ) @@ -23,10 +24,11 @@ func (r *Runner) RunTop(ctx context.Context, name string, options TopOptions) er if err != nil { return err } - p, err := store.Get(ctx, name) + resolved, err := profiles.Resolve(ctx, store, name) if err != nil { return err } + p := resolved.Profile if p.Kind() == query.KindTrace { return fmt.Errorf("profile %q is a trace; use `query trace` to stream it", p.Name) } diff --git a/cmd/query/sessions/trace.go b/cmd/query/sessions/trace.go index 47ecd60..471b0c5 100644 --- a/cmd/query/sessions/trace.go +++ b/cmd/query/sessions/trace.go @@ -13,6 +13,7 @@ import ( "syscall" "time" + "github.com/flanksource/commons-db/cmd/query/profiles" "github.com/flanksource/commons-db/query" ) @@ -54,10 +55,11 @@ func (r *Runner) RunTrace(ctx context.Context, name string, options TraceOptions if err != nil { return err } - p, err := store.Get(ctx, name) + resolved, err := profiles.Resolve(ctx, store, name) if err != nil { return err } + p := resolved.Profile if p.Kind() != query.KindTrace { return fmt.Errorf("profile %q is not a trace; use `query top` to sample it", p.Name) } diff --git a/go.mod b/go.mod index aa38786..6c70e84 100644 --- a/go.mod +++ b/go.mod @@ -75,6 +75,7 @@ require ( github.com/stretchr/testify v1.11.1 github.com/timberio/go-datemath v0.1.0 github.com/zclconf/go-cty v1.14.4 + go.yaml.in/yaml/v3 v3.0.4 go.opentelemetry.io/otel v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.44.0 go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.44.0 @@ -352,7 +353,6 @@ require ( go.opentelemetry.io/otel/sdk/metric v1.44.0 // indirect go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.yaml.in/yaml/v2 v2.4.4 // indirect - go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/mod v0.36.0 // indirect golang.org/x/net v0.56.0 // indirect golang.org/x/sys v0.46.0 // indirect diff --git a/query/cel.go b/query/cel.go index b643435..82e599d 100644 --- a/query/cel.go +++ b/query/cel.go @@ -2,50 +2,102 @@ package query import ( "fmt" + "regexp" + "strings" "github.com/flanksource/gomplate/v3" "github.com/flanksource/commons-db/context" ) -// applyColumns evaluates each column's CEL expression (when set) against every -// row, writing the computed value back to the row under the column name. Columns -// without a CEL expression are left untouched (the provider already populated -// them). -// -// The current row is exposed to CEL as `row`. -func applyColumns(ctx context.Context, columns []ColumnDef, rows []Row) error { - celColumns := make([]ColumnDef, 0, len(columns)) - for _, col := range columns { - if col.CEL != "" { - celColumns = append(celColumns, col) - } - } - if len(celColumns) == 0 { - return nil - } +var celIdentifier = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`) - for i, row := range rows { - for _, col := range celColumns { - val, err := evalColumnCEL(ctx, col.CEL, row) +func applyRowTransforms(ctx context.Context, profile Profile, rows []Row) error { + for index, row := range rows { + aliasRoots := map[string]bool{} + for _, alias := range profile.Aliases { + if alias.Name == "" || alias.CEL == "" { + return fmt.Errorf("row %d: alias name and cel are required", index) + } + value, err := evalRowCEL(ctx, alias.CEL, row) + if err != nil { + return fmt.Errorf("row %d: alias %q: %w", index, alias.Name, err) + } + setRowPath(row, alias.Name, value) + if strings.Contains(alias.Name, ".") { + aliasRoots[strings.Split(alias.Name, ".")[0]] = true + } + } + for _, ignored := range profile.Ignore { + if aliasRoots[ignored] { + continue + } + delete(row, ignored) + deleteRowPath(row, ignored) + } + for _, column := range profile.Columns { + if column.CEL == "" { + continue + } + value, err := evalRowCEL(ctx, column.CEL, row) if err != nil { - return fmt.Errorf("column %q: row %d: %w", col.Name, i, err) + return fmt.Errorf("row %d: column %q: %w", index, column.Name, err) } - row[col.Name] = val + row[column.Name] = value } } - return nil } -// evalColumnCEL evaluates a single CEL expression with the row bound to `row`, -// returning the typed result. The commons-db CEL environment functions are -// injected so expressions can use the same helpers as the rest of the platform. -func evalColumnCEL(ctx context.Context, expr string, row Row) (any, error) { - t := gomplate.Template{Expression: expr} - for _, f := range context.CelEnvFuncs { - t.CelEnvs = append(t.CelEnvs, f(ctx)) +func applyColumns(ctx context.Context, columns []ColumnDef, rows []Row) error { + return applyRowTransforms(ctx, Profile{Columns: columns}, rows) +} + +func evalRowCEL(ctx context.Context, expression string, row Row) (any, error) { + template := gomplate.Template{Expression: expression} + for _, function := range context.CelEnvFuncs { + template.CelEnvs = append(template.CelEnvs, function(ctx)) + } + environment := map[string]any{"row": row, "span": row} + for name, value := range row { + if celIdentifier.MatchString(name) { + environment[name] = value + } } + return gomplate.RunExpressionContext(ctx.Context, environment, template) +} - return gomplate.RunExpressionContext(ctx.Context, map[string]any{"row": row}, t) +func setRowPath(row Row, path string, value any) { + parts := strings.Split(path, ".") + if len(parts) == 1 { + row[path] = value + return + } + current := map[string]any(row) + for _, part := range parts[:len(parts)-1] { + next, ok := current[part].(map[string]any) + if !ok { + next = map[string]any{} + current[part] = next + } + current = next + } + current[parts[len(parts)-1]] = value +} + +func deleteRowPath(row Row, path string) { + parts := strings.Split(path, ".") + if len(parts) == 1 { + delete(row, path) + return + } + current := map[string]any(row) + for _, part := range parts[:len(parts)-1] { + next, ok := current[part].(map[string]any) + if !ok { + return + } + current = next + } + delete(current, parts[len(parts)-1]) } diff --git a/query/engine.go b/query/engine.go index d28f540..7d65e13 100644 --- a/query/engine.go +++ b/query/engine.go @@ -57,12 +57,13 @@ func executeResolved(ctx context.Context, p Profile, resolved map[string]any) (* Connection: p.Provider.Connection, Query: query, Options: p.Provider.Options, + Params: resolved, }) if err != nil { return nil, fmt.Errorf("profile %q: provider %q failed: %w", p.Name, p.Provider.Type, err) } - if err := applyColumns(ctx, p.Columns, rows); err != nil { + if err := applyRowTransforms(ctx, p, rows); err != nil { return nil, fmt.Errorf("profile %q: %w", p.Name, err) } @@ -156,6 +157,7 @@ func executeSubQuery(ctx context.Context, sub SubQuery, params map[string]any) ( Connection: sub.Provider.Connection, Query: query, Options: sub.Provider.Options, + Params: params, }) } diff --git a/query/engine_test.go b/query/engine_test.go index a8dfa9d..1e6890f 100644 --- a/query/engine_test.go +++ b/query/engine_test.go @@ -102,6 +102,36 @@ output: [table, html] }) var _ = Describe("Execute", func() { + It("passes resolved params to providers and applies ordered aliases before ignore and columns", func() { + provider := &mockProvider{typ: "exec-row-pipeline", rows: []query.Row{{ + "input.xml": "P-7", + "obsolete": "remove-me", + }}} + query.RegisterProvider(provider) + + result, err := query.Execute(context.New(), query.Profile{ + Name: "ordered aliases", + Provider: query.ProviderConfig{Type: provider.typ}, + Params: []query.ParamDef{{Name: "namespace", Default: "prod"}}, + Aliases: []query.AliasDef{ + {Name: "request.xml", CEL: `span["input.xml"]`}, + {Name: "request.copy", CEL: `request.xml`}, + {Name: "ignoredAlias", CEL: `request.copy`}, + }, + Ignore: []string{"input.xml", "obsolete", "ignoredAlias"}, + Columns: []query.ColumnDef{{Name: "Copied", CEL: `request.copy`}}, + }) + Expect(err).ToNot(HaveOccurred()) + Expect(provider.last.Params).To(HaveKeyWithValue("namespace", "prod")) + Expect(result.Rows).To(Equal([]query.Row{{ + "request": map[string]any{ + "xml": "P-7", + "copy": "P-7", + }, + "Copied": "P-7", + }})) + }) + It("dispatches to the provider and returns its rows", func() { rows := []query.Row{{"id": 1}, {"id": 2}} query.RegisterProvider(&mockProvider{typ: "exec-primary", rows: rows}) diff --git a/query/iterator.go b/query/iterator.go index f01450d..e5ff925 100644 --- a/query/iterator.go +++ b/query/iterator.go @@ -82,18 +82,18 @@ func executeRows(ctx context.Context, p Profile, maxRows int, params ...map[stri Connection: p.Provider.Connection, Query: rendered, Options: p.Provider.Options, + Params: resolved, MaxRows: maxRows, }) if err != nil { return nil, fmt.Errorf("profile %q: provider %q failed: %w", p.Name, p.Provider.Type, err) } - return &columnRowIterator{ctx: ctx, profile: p.Name, columns: p.Columns, rows: rows}, nil + return &columnRowIterator{ctx: ctx, profile: p, rows: rows}, nil } type columnRowIterator struct { ctx context.Context - profile string - columns []ColumnDef + profile Profile rows RowIterator row Row err error @@ -105,8 +105,8 @@ func (i *columnRowIterator) Next() bool { return false } i.row = i.rows.Row() - if err := applyColumns(i.ctx, i.columns, []Row{i.row}); err != nil { - i.err = fmt.Errorf("profile %q: row %d: %w", i.profile, i.index, err) + if err := applyRowTransforms(i.ctx, i.profile, []Row{i.row}); err != nil { + i.err = fmt.Errorf("profile %q: row %d: %w", i.profile.Name, i.index, err) return false } i.index++ diff --git a/query/profile.go b/query/profile.go index 6805cae..7a83bda 100644 --- a/query/profile.go +++ b/query/profile.go @@ -10,6 +10,10 @@ type Profile struct { // Name identifies the Profile (e.g. "SQL Server trace"). Name string `json:"profile" yaml:"profile"` + // Imports names profiles to merge from left to right before this profile is + // executed. The authored profile remains unchanged in the profile store. + Imports []string `json:"imports,omitempty" yaml:"imports,omitempty"` + // Namespace scopes Kubernetes secret/configmap lookups and workload URLs used // by inline provider connections. When empty, the caller's namespace is used. Namespace string `json:"namespace,omitempty" yaml:"namespace,omitempty"` @@ -31,6 +35,13 @@ type Profile struct { // provider's raw row keys are used. Columns []ColumnDef `json:"columns,omitempty" yaml:"columns,omitempty"` + // Aliases are ordered CEL projections applied to each provider row. Later + // aliases can reference values produced by earlier aliases. + Aliases []AliasDef `json:"aliases,omitempty" yaml:"aliases,omitempty"` + + // Ignore removes provider fields after aliases have been evaluated. + Ignore []string `json:"ignore,omitempty" yaml:"ignore,omitempty"` + // Processors are post-query steps (e.g. sqlite merge/recon) applied in order. Processors []ProcessorSpec `json:"processors,omitempty" yaml:"processors,omitempty"` @@ -59,6 +70,12 @@ type Profile struct { Top *TopSpec `json:"top,omitempty" yaml:"top,omitempty"` } +// AliasDef is an ordered, named CEL projection over a provider row. +type AliasDef struct { + Name string `json:"name" yaml:"name"` + CEL string `json:"cel" yaml:"cel"` +} + // Render values the frontend keys presentation off (x-clicky-render): // RenderLogs selects the canonical LogsTable; RenderTrace and RenderTop select // the session-backed live views and are derived from the profile kind when diff --git a/query/provider.go b/query/provider.go index b6d180a..a26bea1 100644 --- a/query/provider.go +++ b/query/provider.go @@ -31,6 +31,10 @@ type ProviderRequest struct { // Options carries provider-specific knobs from ProviderConfig.Options. Options map[string]any + // Params contains the validated profile parameters. Providers use this for + // native query builders that cannot be expressed as a query template. + Params map[string]any + // MaxRows is an execution hint for bounded callers such as an interactive // page. Streaming providers may use it to avoid opening a backend cursor // when one finite request can satisfy the caller. Zero means unbounded. diff --git a/query/sample.go b/query/sample.go index 8057b50..cb1d4e3 100644 --- a/query/sample.go +++ b/query/sample.go @@ -57,6 +57,7 @@ func Sample(ctx context.Context, p Profile, params map[string]any, limit int) (* Connection: p.Provider.Connection, Query: rendered, Options: p.Provider.Options, + Params: resolved, }) duration := time.Since(started) if err != nil { diff --git a/query/schema/profile.go b/query/schema/profile.go index dcb374c..cd273be 100644 --- a/query/schema/profile.go +++ b/query/schema/profile.go @@ -10,6 +10,7 @@ import ( var providerTypes = []string{ "sql", "postgres", "mysql", "sqlserver", "clickhouse", "http", "prometheus", "postgrest", "loki", "opensearch", "jaeger", + "opentelemetry", } // providerConnectionTypes maps each profile provider type to the connection @@ -20,17 +21,18 @@ var providerTypes = []string{ // the eligible types as a scope filter. Note ConnectionTypeSQLServer is // "sql_server" — the value the connection list filters on. var providerConnectionTypes = map[string][]string{ - "sql": {models.ConnectionTypePostgres, models.ConnectionTypeMySQL, models.ConnectionTypeSQLServer, models.ConnectionTypeClickHouse}, - "postgres": {models.ConnectionTypePostgres}, - "mysql": {models.ConnectionTypeMySQL}, - "sqlserver": {models.ConnectionTypeSQLServer}, - "clickhouse": {models.ConnectionTypeClickHouse}, - "http": {models.ConnectionTypeHTTP}, - "postgrest": {models.ConnectionTypeHTTP}, - "prometheus": {models.ConnectionTypePrometheus}, - "loki": {models.ConnectionTypeLoki}, - "opensearch": {models.ConnectionTypeOpenSearch}, - "jaeger": {models.ConnectionTypeJaeger}, + "sql": {models.ConnectionTypePostgres, models.ConnectionTypeMySQL, models.ConnectionTypeSQLServer, models.ConnectionTypeClickHouse}, + "postgres": {models.ConnectionTypePostgres}, + "mysql": {models.ConnectionTypeMySQL}, + "sqlserver": {models.ConnectionTypeSQLServer}, + "clickhouse": {models.ConnectionTypeClickHouse}, + "http": {models.ConnectionTypeHTTP}, + "postgrest": {models.ConnectionTypeHTTP}, + "prometheus": {models.ConnectionTypePrometheus}, + "loki": {models.ConnectionTypeLoki}, + "opensearch": {models.ConnectionTypeOpenSearch}, + "opentelemetry": {models.ConnectionTypeOpenTelemetry}, + "jaeger": {models.ConnectionTypeJaeger}, } // ProfileSource returns the externally referenced profile form schema. Each @@ -76,6 +78,10 @@ func ProfileSource() Schema { "hidden": Schema{"type": "boolean", "title": "Hidden"}, }, } + aliasDef := Schema{ + "type": "object", "required": []string{"name", "cel"}, + "properties": Schema{"name": strProp("Name", "Dotted output path"), "cel": strProp("CEL", "Ordered row projection")}, + } provider := Schema{ "type": "object", @@ -111,6 +117,7 @@ func ProfileSource() Schema { "required": []string{"profile", "provider"}, "properties": Schema{ "profile": strProp("Name", "Profile name"), + "imports": Schema{"type": "array", "title": "Imports", "items": Schema{"type": "string"}}, "namespace": Schema{ "type": "string", "title": "Namespace", @@ -127,6 +134,8 @@ func ProfileSource() Schema { }, "params": Schema{"type": "array", "title": "Params", "items": paramDef}, "columns": Schema{"type": "array", "title": "Columns", "x-layout": "table", "items": columnDef}, + "aliases": Schema{"type": "array", "title": "Aliases", "x-layout": "table", "items": aliasDef}, + "ignore": Schema{"type": "array", "title": "Ignore", "items": Schema{"type": "string"}}, "output": Schema{"type": "array", "title": "Output", "items": Schema{"type": "string"}}, "render": Schema{"type": "string", "title": "Render", "enum": []string{"table", "logs"}, "description": "Presentation mode: table (default) or logs (canonical LogsTable view for trace/log profiles)"}, }, diff --git a/query/schema/profile_components.go b/query/schema/profile_components.go index 6b6202c..3a0b733 100644 --- a/query/schema/profile_components.go +++ b/query/schema/profile_components.go @@ -3,17 +3,18 @@ package schema // providerTypeIcons are runtime icon names resolved by clicky-ui's fallback // icon provider. They intentionally mirror the profile surface icon families. var providerTypeIcons = map[string]string{ - "sql": "database", - "postgres": "postgres", - "mysql": "mysql", - "sqlserver": "sqlserver", - "clickhouse": "clickhouse", - "http": "globe", - "prometheus": "prometheus", - "postgrest": "globe", - "loki": "grafana", - "opensearch": "opensearch", - "jaeger": "activity", + "sql": "database", + "postgres": "postgres", + "mysql": "mysql", + "sqlserver": "sqlserver", + "clickhouse": "clickhouse", + "http": "globe", + "prometheus": "prometheus", + "postgrest": "globe", + "loki": "grafana", + "opensearch": "opensearch", + "opentelemetry": "opentelemetry", + "jaeger": "activity", } // ProfileComponents returns one standalone provider-form component per @@ -87,6 +88,14 @@ func providerOptions(typ string) Schema { props["address"] = inlineURLProp("Address", "Inline OpenSearch address used instead of a saved connection") props["index"] = strProp("Index", "Index or index pattern") props["limit"] = strProp("Limit", "Maximum number of hits") + case "opentelemetry": + for _, field := range []string{"format", "index", "dateField", "traceIdField", "spanIdField", "parentIdField", "parentRefType", "serviceField", "operationField"} { + props[field] = strProp(titleCase(field), "") + } + for _, field := range []string{"statusFields", "selectFields", "sourceExcludes"} { + props[field] = Schema{"type": "array", "title": titleCase(field), "items": Schema{"type": "string"}} + } + props["params"] = Schema{"type": "object", "title": "Provider Params"} case "jaeger": props["url"] = inlineURLProp("URL", "Inline Jaeger query URL used instead of a saved connection") for _, field := range []string{"service", "operation", "lookback", "start", "end", "limit", "minDuration", "maxDuration", "tags"} { diff --git a/query/schema/schema_test.go b/query/schema/schema_test.go index 001ce42..36b8277 100644 --- a/query/schema/schema_test.go +++ b/query/schema/schema_test.go @@ -62,7 +62,8 @@ var _ = Describe("Connection schema", func() { models.ConnectionTypeKubernetes, models.ConnectionTypeZulipChat, )) // guards against drift from the models.ConnectionType* constant set - Expect(enum).To(HaveLen(55)) + Expect(enum).To(ContainElement(models.ConnectionTypeOpenTelemetry)) + Expect(enum).To(HaveLen(56)) }) It("keeps the base form to name/namespace/type/properties", func() { @@ -84,7 +85,7 @@ var _ = Describe("Connection schema", func() { typeProp := s["properties"].(schema.Schema)["type"].(schema.Schema) Expect(typeProp["x-enum-display"]).To(Equal("combobox")) icons := typeProp["x-enum-icons"].(map[string]string) - Expect(icons).To(HaveLen(55)) + Expect(icons).To(HaveLen(56)) Expect(icons[models.ConnectionTypePostgres]).To(Equal("postgres")) }) @@ -146,6 +147,16 @@ var _ = Describe("Connection schema", func() { Expect(props["properties"].(schema.Schema)["properties"].(schema.Schema)).To(HaveKey("authType")) }) + It("scopes OpenTelemetry to a required nested OpenSearch connection", func() { + then := branchFor(s, models.ConnectionTypeOpenTelemetry) + properties := then["properties"].(schema.Schema)["properties"].(schema.Schema) + Expect(properties["required"]).To(ContainElement("connection")) + connection := properties["properties"].(schema.Schema)["connection"].(schema.Schema) + lookup := connection["x-clicky-lookup"].(schema.Schema) + scope := lookup["scope"].(schema.Schema) + Expect(scope["map"].(map[string][]string)[models.ConnectionTypeOpenTelemetry]).To(Equal([]string{models.ConnectionTypeOpenSearch})) + }) + It("surfaces certificate per type: optional for kubernetes, required for GCP", func() { k8s := branchFor(s, models.ConnectionTypeKubernetes) Expect(k8s["properties"].(schema.Schema)).To(HaveKey("certificate")) @@ -171,14 +182,14 @@ var _ = Describe("Connection schema", func() { } }) - It("emits external source refs and a local-ref bundle for all 55 components", func() { - Expect(schema.ConnectionComponents()).To(HaveLen(55)) + It("emits external source refs and a local-ref bundle for all 56 components", func() { + Expect(schema.ConnectionComponents()).To(HaveLen(56)) source := schema.ConnectionSource() firstSourceBranch := source["allOf"].([]any)[0].(schema.Schema) Expect(firstSourceBranch["then"].(schema.Schema)["$ref"]).To(HavePrefix("connections/")) bundled := schema.Connection() - Expect(bundled["$defs"].(schema.Schema)).To(HaveLen(55)) + Expect(bundled["$defs"].(schema.Schema)).To(HaveLen(56)) firstBundledBranch := bundled["allOf"].([]any)[0].(schema.Schema) Expect(firstBundledBranch["then"].(schema.Schema)["$ref"]).To(HavePrefix("#/$defs/")) }) @@ -218,18 +229,18 @@ var _ = Describe("Profile schema", func() { Expect(provider["x-discriminator"]).To(Equal("type")) typeProp := provider["properties"].(schema.Schema)["type"].(schema.Schema) Expect(typeProp["x-enum-display"]).To(Equal("combobox")) - Expect(typeProp["x-enum-icons"].(map[string]string)).To(HaveLen(11)) + Expect(typeProp["x-enum-icons"].(map[string]string)).To(HaveLen(12)) }) It("bundles every provider component and enriches inline URLs", func() { - Expect(schema.ProfileComponents()).To(HaveLen(11)) + Expect(schema.ProfileComponents()).To(HaveLen(12)) source := schema.ProfileSource() provider := source["properties"].(schema.Schema)["provider"].(schema.Schema) firstSourceBranch := provider["allOf"].([]any)[0].(schema.Schema) Expect(firstSourceBranch["then"].(schema.Schema)["$ref"]).To(HavePrefix("profiles/")) bundled := schema.Profile() - Expect(bundled["$defs"].(schema.Schema)).To(HaveLen(11)) + Expect(bundled["$defs"].(schema.Schema)).To(HaveLen(12)) http := bundled["$defs"].(schema.Schema)["http"].(schema.Schema) options := http["properties"].(schema.Schema)["options"].(schema.Schema) url := options["properties"].(schema.Schema)["url"].(schema.Schema) diff --git a/query/stream.go b/query/stream.go index f1b699e..4eef303 100644 --- a/query/stream.go +++ b/query/stream.go @@ -75,6 +75,7 @@ func startTrace(ctx context.Context, reg *SessionRegistry, p Profile, resolved m Connection: p.Provider.Connection, Query: rendered, Options: p.Provider.Options, + Params: resolved, }) return session, nil } @@ -87,7 +88,7 @@ func runTrace(ctx context.Context, cancel stdcontext.CancelFunc, sp StreamProvid var once sync.Once err := sp.Stream(ctx, req, func(row Row) { rows := []Row{row} - if cerr := applyColumns(ctx, p.Columns, rows); cerr != nil { + if cerr := applyRowTransforms(ctx, p, rows); cerr != nil { once.Do(func() { emitErr = fmt.Errorf("profile %q: %w", p.Name, cerr) cancel() diff --git a/schemas/src/profile.json b/schemas/src/profile.json index 947c9e2..35b77d1 100644 --- a/schemas/src/profile.json +++ b/schemas/src/profile.json @@ -1,6 +1,30 @@ { "$schema": "https://json-schema.org/draft/2020-12/schema", "properties": { + "aliases": { + "items": { + "properties": { + "cel": { + "description": "Ordered row projection", + "title": "CEL", + "type": "string" + }, + "name": { + "description": "Dotted output path", + "title": "Name", + "type": "string" + } + }, + "required": [ + "name", + "cel" + ], + "type": "object" + }, + "title": "Aliases", + "type": "array", + "x-layout": "table" + }, "columns": { "items": { "properties": { @@ -75,6 +99,20 @@ "type": "array", "x-layout": "table" }, + "ignore": { + "items": { + "type": "string" + }, + "title": "Ignore", + "type": "array" + }, + "imports": { + "items": { + "type": "string" + }, + "title": "Imports", + "type": "array" + }, "namespace": { "title": "Namespace", "type": "string", @@ -326,6 +364,21 @@ "then": { "$ref": "profiles/jaeger.json" } + }, + { + "if": { + "properties": { + "type": { + "const": "opentelemetry" + } + }, + "required": [ + "type" + ] + }, + "then": { + "$ref": "profiles/opentelemetry.json" + } } ], "properties": { @@ -357,6 +410,9 @@ "opensearch": [ "opensearch" ], + "opentelemetry": [ + "opentelemetry" + ], "postgres": [ "postgres" ], @@ -398,7 +454,8 @@ "postgrest", "loki", "opensearch", - "jaeger" + "jaeger", + "opentelemetry" ], "title": "Type", "type": "string", @@ -410,6 +467,7 @@ "loki": "grafana", "mysql": "mysql", "opensearch": "opensearch", + "opentelemetry": "opentelemetry", "postgres": "postgres", "postgrest": "globe", "prometheus": "prometheus", diff --git a/schemas/src/profiles/clickhouse.json b/schemas/src/profiles/clickhouse.json index 9063f87..d12fcea 100644 --- a/schemas/src/profiles/clickhouse.json +++ b/schemas/src/profiles/clickhouse.json @@ -30,6 +30,9 @@ "opensearch": [ "opensearch" ], + "opentelemetry": [ + "opentelemetry" + ], "postgres": [ "postgres" ], diff --git a/schemas/src/profiles/http.json b/schemas/src/profiles/http.json index 1b16907..760ac6c 100644 --- a/schemas/src/profiles/http.json +++ b/schemas/src/profiles/http.json @@ -30,6 +30,9 @@ "opensearch": [ "opensearch" ], + "opentelemetry": [ + "opentelemetry" + ], "postgres": [ "postgres" ], diff --git a/schemas/src/profiles/jaeger.json b/schemas/src/profiles/jaeger.json index ec89379..ae509b7 100644 --- a/schemas/src/profiles/jaeger.json +++ b/schemas/src/profiles/jaeger.json @@ -30,6 +30,9 @@ "opensearch": [ "opensearch" ], + "opentelemetry": [ + "opentelemetry" + ], "postgres": [ "postgres" ], diff --git a/schemas/src/profiles/loki.json b/schemas/src/profiles/loki.json index b0977f8..526e886 100644 --- a/schemas/src/profiles/loki.json +++ b/schemas/src/profiles/loki.json @@ -30,6 +30,9 @@ "opensearch": [ "opensearch" ], + "opentelemetry": [ + "opentelemetry" + ], "postgres": [ "postgres" ], diff --git a/schemas/src/profiles/mysql.json b/schemas/src/profiles/mysql.json index d64e14e..ac4de4c 100644 --- a/schemas/src/profiles/mysql.json +++ b/schemas/src/profiles/mysql.json @@ -30,6 +30,9 @@ "opensearch": [ "opensearch" ], + "opentelemetry": [ + "opentelemetry" + ], "postgres": [ "postgres" ], diff --git a/schemas/src/profiles/opensearch.json b/schemas/src/profiles/opensearch.json index a80a5ab..1ada925 100644 --- a/schemas/src/profiles/opensearch.json +++ b/schemas/src/profiles/opensearch.json @@ -30,6 +30,9 @@ "opensearch": [ "opensearch" ], + "opentelemetry": [ + "opentelemetry" + ], "postgres": [ "postgres" ], diff --git a/schemas/src/profiles/opentelemetry.json b/schemas/src/profiles/opentelemetry.json new file mode 100644 index 0000000..95242cb --- /dev/null +++ b/schemas/src/profiles/opentelemetry.json @@ -0,0 +1,139 @@ +{ + "$id": "opentelemetry.json", + "$schema": "https://json-schema.org/draft/2020-12/schema", + "properties": { + "connection": { + "description": "Pick a saved connection or type an inline DSN/URL", + "title": "Connection", + "type": "string", + "x-clicky-lookup": { + "filter": "connection", + "multi": false, + "scope": { + "from": "provider.type", + "map": { + "clickhouse": [ + "clickhouse" + ], + "http": [ + "http" + ], + "jaeger": [ + "jaeger" + ], + "loki": [ + "loki" + ], + "mysql": [ + "mysql" + ], + "opensearch": [ + "opensearch" + ], + "opentelemetry": [ + "opentelemetry" + ], + "postgres": [ + "postgres" + ], + "postgrest": [ + "http" + ], + "prometheus": [ + "prometheus" + ], + "sql": [ + "postgres", + "mysql", + "sql_server", + "clickhouse" + ], + "sqlserver": [ + "sql_server" + ] + }, + "param": "types" + }, + "searchParam": "__lookup_q", + "url": "/api/v1/connection" + } + }, + "options": { + "properties": { + "dateField": { + "title": "date Field", + "type": "string" + }, + "format": { + "title": "Format", + "type": "string" + }, + "index": { + "title": "Index", + "type": "string" + }, + "operationField": { + "title": "operation Field", + "type": "string" + }, + "params": { + "title": "Provider Params", + "type": "object" + }, + "parentIdField": { + "title": "parent IdField", + "type": "string" + }, + "parentRefType": { + "title": "parent RefType", + "type": "string" + }, + "selectFields": { + "items": { + "type": "string" + }, + "title": "select Fields", + "type": "array" + }, + "serviceField": { + "title": "service Field", + "type": "string" + }, + "sourceExcludes": { + "items": { + "type": "string" + }, + "title": "source Excludes", + "type": "array" + }, + "spanIdField": { + "title": "span IdField", + "type": "string" + }, + "statusFields": { + "items": { + "type": "string" + }, + "title": "status Fields", + "type": "array" + }, + "traceIdField": { + "title": "trace IdField", + "type": "string" + } + }, + "title": "Options", + "type": "object" + }, + "type": { + "const": "opentelemetry", + "title": "Type", + "type": "string" + } + }, + "required": [ + "type" + ], + "title": "Query provider: opentelemetry", + "type": "object" +} diff --git a/schemas/src/profiles/postgres.json b/schemas/src/profiles/postgres.json index 9344aab..c7a8521 100644 --- a/schemas/src/profiles/postgres.json +++ b/schemas/src/profiles/postgres.json @@ -30,6 +30,9 @@ "opensearch": [ "opensearch" ], + "opentelemetry": [ + "opentelemetry" + ], "postgres": [ "postgres" ], diff --git a/schemas/src/profiles/postgrest.json b/schemas/src/profiles/postgrest.json index 7dff4da..64dd935 100644 --- a/schemas/src/profiles/postgrest.json +++ b/schemas/src/profiles/postgrest.json @@ -30,6 +30,9 @@ "opensearch": [ "opensearch" ], + "opentelemetry": [ + "opentelemetry" + ], "postgres": [ "postgres" ], diff --git a/schemas/src/profiles/prometheus.json b/schemas/src/profiles/prometheus.json index 93266d8..270ff4c 100644 --- a/schemas/src/profiles/prometheus.json +++ b/schemas/src/profiles/prometheus.json @@ -30,6 +30,9 @@ "opensearch": [ "opensearch" ], + "opentelemetry": [ + "opentelemetry" + ], "postgres": [ "postgres" ], diff --git a/schemas/src/profiles/sql.json b/schemas/src/profiles/sql.json index e6090c9..18dc460 100644 --- a/schemas/src/profiles/sql.json +++ b/schemas/src/profiles/sql.json @@ -30,6 +30,9 @@ "opensearch": [ "opensearch" ], + "opentelemetry": [ + "opentelemetry" + ], "postgres": [ "postgres" ], diff --git a/schemas/src/profiles/sqlserver.json b/schemas/src/profiles/sqlserver.json index 0d4ffe1..262e483 100644 --- a/schemas/src/profiles/sqlserver.json +++ b/schemas/src/profiles/sqlserver.json @@ -30,6 +30,9 @@ "opensearch": [ "opensearch" ], + "opentelemetry": [ + "opentelemetry" + ], "postgres": [ "postgres" ], From a8e144b3d557ac68fb0b598894c446e2d569989f Mon Sep 17 00:00:00 2001 From: Moshe Immerman Date: Fri, 17 Jul 2026 17:46:36 +0300 Subject: [PATCH 05/12] feat(query): support OpenTelemetry connection mapping Prompt for a missing OpenTelemetry connection, persist the selected mapping, and retry profile execution. Add an inline connection form when no compatible connection exists. --- cmd/query/profiles/execution.go | 86 +++++- cmd/query/profiles/execution_test.go | 82 ++++++ cmd/query/www/src/App.tsx | 25 +- .../www/src/profileConnectionMapping.test.ts | 37 +++ .../www/src/profileConnectionMapping.tsx | 268 ++++++++++++++++++ 5 files changed, 490 insertions(+), 8 deletions(-) create mode 100644 cmd/query/www/src/profileConnectionMapping.test.ts create mode 100644 cmd/query/www/src/profileConnectionMapping.tsx diff --git a/cmd/query/profiles/execution.go b/cmd/query/profiles/execution.go index fe13a4f..8e6b8d1 100644 --- a/cmd/query/profiles/execution.go +++ b/cmd/query/profiles/execution.go @@ -3,6 +3,7 @@ package profiles import ( "bytes" stdcontext "context" + "encoding/json" "fmt" "net/http" "strconv" @@ -12,6 +13,7 @@ import ( "github.com/flanksource/clicky/api" "github.com/flanksource/clicky/formatters" dbcontext "github.com/flanksource/commons-db/context" + "github.com/flanksource/commons-db/models" "github.com/flanksource/commons-db/query" ) @@ -36,6 +38,12 @@ func newExecHandler(prefix string, ctx dbcontext.Context, store Store, next http } func (h *execHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method == http.MethodPut { + if name, ok := h.connectionProfileName(r.URL.Path); ok { + h.mapConnection(w, r, name) + return + } + } if r.Method == http.MethodGet && !wantsSchema(r) { if name, ok := h.profileName(r.URL.Path); ok { h.execute(w, r, name) @@ -45,6 +53,15 @@ func (h *execHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { h.next.ServeHTTP(w, r) } +func (h *execHandler) connectionProfileName(path string) (string, bool) { + rel := strings.Trim(strings.TrimPrefix(strings.TrimSuffix(path, "/"), h.prefix), "/") + parts := strings.Split(rel, "/") + if len(parts) != 3 || parts[0] != "profile" || parts[1] == "" || parts[2] != "connection" { + return "", false + } + return parts[1], true +} + // profileName returns the {name} segment of {prefix}/profile/{name}, or false. func (h *execHandler) profileName(path string) (string, bool) { rel := strings.Trim(strings.TrimPrefix(strings.TrimSuffix(path, "/"), h.prefix), "/") @@ -68,13 +85,18 @@ func (h *execHandler) execute(w http.ResponseWriter, r *http.Request, name strin case <-base.Done(): } }() - execCtx := dbcontext.NewContext(base) + execCtx := h.ctx.Wrap(base) - p, err := h.store.Get(r.Context(), name) + resolved, err := Resolve(r.Context(), h.store, name) if err != nil { http.Error(w, err.Error(), http.StatusNotFound) return } + p := resolved.Profile + if p.Provider.Type == "opentelemetry" && p.Provider.Connection == "" { + h.writeConnectionRequired(w, name, resolved.ConnectionProfile) + return + } params := map[string]any{} for k, vs := range r.URL.Query() { @@ -144,6 +166,66 @@ func (h *execHandler) execute(w http.ResponseWriter, r *http.Request, name strin } } +type connectionMappingRequest struct { + Connection string `json:"connection"` +} + +func (h *execHandler) mapConnection(w http.ResponseWriter, r *http.Request, name string) { + resolved, err := Resolve(r.Context(), h.store, name) + if err != nil { + http.Error(w, err.Error(), http.StatusNotFound) + return + } + var request connectionMappingRequest + if err := json.NewDecoder(r.Body).Decode(&request); err != nil { + http.Error(w, fmt.Sprintf("decode connection mapping: %v", err), http.StatusBadRequest) + return + } + request.Connection = strings.TrimSpace(request.Connection) + if request.Connection == "" { + http.Error(w, "connection is required", http.StatusBadRequest) + return + } + selected, err := dbcontext.FindConnectionByURL(h.ctx, request.Connection) + if err != nil { + http.Error(w, err.Error(), http.StatusBadRequest) + return + } + if selected == nil { + http.Error(w, fmt.Sprintf("connection %q not found", request.Connection), http.StatusBadRequest) + return + } + if selected.Type != models.ConnectionTypeOpenTelemetry { + http.Error(w, fmt.Sprintf("connection %q has type %q, expected %q", selected.Name, selected.Type, models.ConnectionTypeOpenTelemetry), http.StatusBadRequest) + return + } + owner, err := h.store.Get(r.Context(), resolved.ConnectionProfile) + if err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + owner.Provider.Connection = request.Connection + if err := h.store.Save(r.Context(), owner); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _ = json.NewEncoder(w).Encode(map[string]any{ + "profile": name, "mappingProfile": resolved.ConnectionProfile, "connection": request.Connection, + }) +} + +func (h *execHandler) writeConnectionRequired(w http.ResponseWriter, profile, mappingProfile string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusConflict) + _ = json.NewEncoder(w).Encode(map[string]any{ + "code": "profile_connection_required", "profile": profile, + "mappingProfile": mappingProfile, "connectionType": models.ConnectionTypeOpenTelemetry, + "mappingUrl": h.prefix + "/profile/" + profile + "/connection", + }) +} + type exportRequest struct { format string scope string diff --git a/cmd/query/profiles/execution_test.go b/cmd/query/profiles/execution_test.go index 32a82df..3b1eeb8 100644 --- a/cmd/query/profiles/execution_test.go +++ b/cmd/query/profiles/execution_test.go @@ -1,6 +1,7 @@ package profiles import ( + "bytes" "context" "encoding/csv" "encoding/json" @@ -11,7 +12,11 @@ import ( "testing" dbcontext "github.com/flanksource/commons-db/context" + "github.com/flanksource/commons-db/models" "github.com/flanksource/commons-db/query" + "github.com/google/uuid" + "gorm.io/driver/sqlite" + "gorm.io/gorm" ) type nextMarker struct{ hit bool } @@ -123,6 +128,83 @@ func TestExecHandlerMissingProfile(t *testing.T) { } } +func TestExecHandlerRequestsOpenTelemetryMappingForImportRoot(t *testing.T) { + store, err := NewFileStore(t.TempDir()) + if err != nil { + t.Fatal(err) + } + for _, profile := range []query.Profile{ + {Name: "jaeger", Provider: query.ProviderConfig{Type: "opentelemetry"}}, + {Name: "jms", Imports: []string{"jaeger"}, Provider: query.ProviderConfig{Type: "opentelemetry"}}, + } { + if err := store.Save(context.Background(), profile); err != nil { + t.Fatal(err) + } + } + handler := newExecHandler("/api/v1", dbcontext.New(), store, &nextMarker{}) + response := get(handler, "/api/v1/profile/jms", "") + if response.Code != http.StatusConflict { + t.Fatalf("status=%d body=%s", response.Code, response.Body.String()) + } + var body map[string]any + if err := json.Unmarshal(response.Body.Bytes(), &body); err != nil { + t.Fatal(err) + } + if body["code"] != "profile_connection_required" || body["mappingProfile"] != "jaeger" || body["connectionType"] != "opentelemetry" { + t.Fatalf("unexpected mapping response: %#v", body) + } +} + +func TestExecHandlerPersistsOpenTelemetryMappingOnImportRoot(t *testing.T) { + store, err := NewFileStore(t.TempDir()) + if err != nil { + t.Fatal(err) + } + for _, profile := range []query.Profile{ + {Name: "jaeger", Provider: query.ProviderConfig{Type: "opentelemetry"}}, + {Name: "jms", Imports: []string{"jaeger"}, Provider: query.ProviderConfig{Type: "opentelemetry"}}, + } { + if err := store.Save(context.Background(), profile); err != nil { + t.Fatal(err) + } + } + database, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{}) + if err != nil { + t.Fatal(err) + } + if err := database.Exec(`CREATE TABLE connections ( +id TEXT PRIMARY KEY, name TEXT, namespace TEXT, source TEXT, type TEXT, +url TEXT, username TEXT, password TEXT, properties TEXT, certificate TEXT, +insecure_tls NUMERIC, created_at DATETIME, updated_at DATETIME, created_by TEXT +)`).Error; err != nil { + t.Fatal(err) + } + if err := database.Create(&models.Connection{ID: uuid.New(), Name: "traces", Type: models.ConnectionTypeOpenTelemetry}).Error; err != nil { + t.Fatal(err) + } + handler := newExecHandler("/api/v1", dbcontext.New().WithDB(database, nil), store, &nextMarker{}) + request := httptest.NewRequest(http.MethodPut, "/api/v1/profile/jms/connection", bytes.NewBufferString(`{"connection":"connection://traces"}`)) + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + if response.Code != http.StatusOK { + t.Fatalf("status=%d body=%s", response.Code, response.Body.String()) + } + jaeger, err := store.Get(context.Background(), "jaeger") + if err != nil { + t.Fatal(err) + } + if jaeger.Provider.Connection != "connection://traces" { + t.Fatalf("mapping persisted to %q", jaeger.Provider.Connection) + } + jms, err := store.Get(context.Background(), "jms") + if err != nil { + t.Fatal(err) + } + if jms.Provider.Connection != "" { + t.Fatalf("child profile was modified: %+v", jms.Provider) + } +} + type execStreamMock struct { rows []query.Row last query.ProviderRequest diff --git a/cmd/query/www/src/App.tsx b/cmd/query/www/src/App.tsx index f48956c..9019902 100644 --- a/cmd/query/www/src/App.tsx +++ b/cmd/query/www/src/App.tsx @@ -18,6 +18,10 @@ import { getMonacoWorker } from "./monacoWorkers"; import { ChatWidget } from "./chatWidget"; import { profileBuilderFormExtensions } from "./profileBuilder"; import { BuildProfileButton } from "./buildProfileAction"; +import { + configureProfileConnectionForm, + useProfileConnectionMapping, +} from "./profileConnectionMapping"; // Compose the form extensions: the namespace picker, plus the secret/workload // url selector (which reads the selected namespace from the form's root value). @@ -29,6 +33,11 @@ const formExtensions = { ], }; +configureProfileConnectionForm({ + formPost: formExtensions.post, + footerActions: connectionFormActions, +}); + // The Go server (query serve) exposes: // - the OpenAPI spec + executor under /api (entity discovery, list/get), // - mutations at POST/PUT/DELETE /api/v1/{connection,profile}, @@ -40,7 +49,7 @@ const formExtensions = { // The EntityExplorerApp drives list/detail/filter UI from the OpenAPI spec. The // schema-by-convention endpoints power the create/edit forms and the per-profile // FilterBar; see cmd/query/README.md for the contract. -const client = createOperationsApiClient({ +const baseClient = createOperationsApiClient({ baseUrl: "", openApiPath: "/api/openapi.json", }); @@ -53,6 +62,7 @@ const queryClient = new QueryClient(); // Explorer reads the logs-surface set (needs the QueryClient context) and wires // the result renderer so `render: logs` profiles present via clicky-ui LogsTable. function Explorer() { + const { client, dialog } = useProfileConnectionMapping(baseClient); const logsEntityNames = useLogsEntityNames(); const renderLogsResult = logsResultRenderer(logsEntityNames); const renderResult = (context: ResultRenderContext) => { @@ -68,8 +78,9 @@ function Explorer() { ); }; return ( - + + entityDetailHeaderRenderer={connectionDetailHeaderRenderer} + /> + + {dialog} + ); } @@ -105,7 +119,6 @@ export function App() {
-
diff --git a/cmd/query/www/src/profileConnectionMapping.test.ts b/cmd/query/www/src/profileConnectionMapping.test.ts new file mode 100644 index 0000000..c6e01cc --- /dev/null +++ b/cmd/query/www/src/profileConnectionMapping.test.ts @@ -0,0 +1,37 @@ +import { describe, expect, it } from "vitest"; +import type { ResolvedOperation } from "@flanksource/clicky-ui"; +import { + findConnectionCreateOperation, + isProfileConnectionRequired, + profileConnectionOptions, +} from "./profileConnectionMapping"; + +describe("profile connection mapping", () => { + it("recognizes only structured profile mapping conflicts", () => { + expect( + isProfileConnectionRequired({ + code: "profile_connection_required", + profile: "jms", + mappingProfile: "jaeger", + connectionType: "opentelemetry", + mappingUrl: "/api/v1/profile/jms/connection", + }), + ).toBe(true); + expect(isProfileConnectionRequired({ code: "profile_connection_required" })).toBe(false); + }); + + it("finds the connection create action and formats saved references", () => { + const create = { + path: "/api/v1/connection", + method: "post", + operation: { + responses: {}, + "x-clicky": { surface: "connection", scope: "collection", verb: "create" }, + }, + } satisfies ResolvedOperation; + expect(findConnectionCreateOperation([create])).toBe(create); + expect(profileConnectionOptions({ options: { "connection://traces": {} } })).toEqual([ + { value: "connection://traces", label: "traces" }, + ]); + }); +}); diff --git a/cmd/query/www/src/profileConnectionMapping.tsx b/cmd/query/www/src/profileConnectionMapping.tsx new file mode 100644 index 0000000..c607391 --- /dev/null +++ b/cmd/query/www/src/profileConnectionMapping.tsx @@ -0,0 +1,268 @@ +import { + Button, + Modal, + OperationsApiClientError, + SchemaActionForm, + Select, + useOperations, + type ExecutionResponse, + type FormActionsRenderer, + type PostExtension, + type ResolvedOperation, + type SharedOperationsApiClient, +} from "@flanksource/clicky-ui"; +import { useQuery, useQueryClient } from "@tanstack/react-query"; +import { useCallback, useMemo, useState, type ReactNode } from "react"; + +export type ProfileConnectionRequired = { + code: "profile_connection_required"; + profile: string; + mappingProfile: string; + connectionType: "opentelemetry"; + mappingUrl: string; +}; + +type PendingMapping = { + required: ProfileConnectionRequired; + retry: () => Promise; + resolve: (response: ExecutionResponse) => void; + reject: (error: unknown) => void; + error: unknown; +}; + +export function isProfileConnectionRequired(value: unknown): value is ProfileConnectionRequired { + if (value == null || typeof value !== "object") return false; + const payload = value as Record; + return ( + payload.code === "profile_connection_required" && + typeof payload.profile === "string" && + typeof payload.mappingProfile === "string" && + payload.connectionType === "opentelemetry" && + typeof payload.mappingUrl === "string" + ); +} + +export function findConnectionCreateOperation( + operations: ResolvedOperation[], +): ResolvedOperation | undefined { + return operations.find((operation) => { + const metadata = operation.operation["x-clicky"]; + return ( + metadata?.surface === "connection" && + metadata.scope === "collection" && + metadata.verb === "create" + ); + }); +} + +export function profileConnectionOptions(filter: { + options?: Record; +}): Array<{ value: string; label: string }> { + return Object.keys(filter.options ?? {}).map((value) => ({ + value, + label: value.replace(/^connection:\/\//, ""), + })); +} + +export function useProfileConnectionMapping(base: SharedOperationsApiClient): { + client: SharedOperationsApiClient; + dialog: ReactNode; +} { + const [pending, setPending] = useState(null); + const [createOpen, setCreateOpen] = useState(false); + const waitForMapping = useCallback( + (required: ProfileConnectionRequired, retry: () => Promise, error: unknown) => + new Promise((resolve, reject) => { + setPending({ required, retry, resolve, reject, error }); + }), + [], + ); + const client = useMemo( + () => ({ + ...base, + async executeCommand(path, method, params, headers) { + try { + return await base.executeCommand(path, method, params, headers); + } catch (error) { + if ( + error instanceof OperationsApiClientError && + error.status === 409 && + isProfileConnectionRequired(error.responseData) + ) { + return waitForMapping( + error.responseData, + () => base.executeCommand(path, method, params, headers), + error, + ); + } + throw error; + } + }, + }), + [base, waitForMapping], + ); + return { + client, + dialog: ( + { + pending?.reject(pending.error); + setPending(null); + }} + onMapped={(response) => { + pending?.resolve(response); + setPending(null); + }} + onCreate={() => { + pending?.reject(pending.error); + setPending(null); + setCreateOpen(true); + }} + onCreateClose={() => setCreateOpen(false)} + /> + ), + }; +} + +function ProfileConnectionMappingDialogs({ + client, + pending, + createOpen, + onClose, + onMapped, + onCreate, + onCreateClose, +}: { + client: SharedOperationsApiClient; + pending: PendingMapping | null; + createOpen: boolean; + onClose: () => void; + onMapped: (response: ExecutionResponse) => void; + onCreate: () => void; + onCreateClose: () => void; +}) { + const [selected, setSelected] = useState(""); + const [saving, setSaving] = useState(false); + const [error, setError] = useState(""); + const queryClient = useQueryClient(); + const { operations } = useOperations(client); + const createAction = findConnectionCreateOperation(operations); + const connections = useQuery({ + queryKey: ["profile-opentelemetry-connections"], + enabled: pending != null, + queryFn: () => + client.lookupFilterOptions( + "/api/v1/connection", + "GET", + "connection", + "", + { types: "opentelemetry" }, + ), + }); + const options = profileConnectionOptions(connections.data ?? {}); + + const mapAndRetry = async () => { + if (!pending || !selected) return; + setSaving(true); + setError(""); + try { + await client.submitForm(pending.required.mappingUrl, "PUT", { + connection: selected, + }); + onMapped(await pending.retry()); + } catch (mappingError) { + setError(mappingError instanceof Error ? mappingError.message : String(mappingError)); + } finally { + setSaving(false); + } + }; + + return ( + <> + + {error ? {error} : null} + + {options.length === 0 && !connections.isLoading ? ( + + ) : ( + + )} + + } + > + {connections.isLoading ? ( +
Loading OpenTelemetry connections…
+ ) : options.length > 0 ? ( + + setFilter((current) => ({ + ...current, + query: event.target.value, + })) + } + placeholder={`Search ${discovered.length} fields`} + aria-label="Search fields" + className={inputClassName} + /> +
+ + +
+ +
+ {visibleFields.map((field) => { + const selected = selectedNames.has(field.name); + const active = activeField?.name === field.name; + return ( +
+ + setFieldSelection(field, event.target.checked) + } + /> + +
+ ); + })} + {visibleFields.length === 0 ? ( +

+ No fields match these filters. +

+ ) : null} +
+ + + { + if (activeField) setFieldSelection(activeField, selected); + }} + onChange={updateActiveField} + /> + + ); +} + +function FieldEditor({ + field, + selected, + onSelectedChange, + onChange, +}: { + field?: ProfileColumn; + selected: boolean; + onSelectedChange: (selected: boolean) => void; + onChange: (patch: Partial) => void; +}) { + if (!field) { + return ( +
+ Run a sample to discover and configure fields. +
+ ); + } + + return ( +
+
+
+

+ Field editor +

+

+ {field.name} +

+
+ +
+
+ + onChange({ label: event.target.value || undefined })} + /> + + + + + + + + + onChange({ format: event.target.value || undefined })} + /> + + + onChange({ unit: event.target.value || undefined })} + /> + + + + onChange({ + width: event.target.value + ? Number(event.target.value) + : undefined, + }) + } + /> + +
+ +