diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index aad9647..bf021bd 100644 --- a/.github/workflows/e2e.yml +++ b/.github/workflows/e2e.yml @@ -37,6 +37,11 @@ jobs: go-version: ["1.26"] timeout-minutes: 30 + # Exercises the embedded-postgres path on both platforms. The external-DB + # path is covered by e2e-external, which needs a Linux-only service container. + env: + COMMONS_DB_EMBEDDED_TEST: "1" + steps: - name: Checkout code uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 @@ -116,11 +121,30 @@ jobs: go fmt ./e2e/... continue-on-error: false - e2e-linux: - name: E2E Tests (Linux - Detailed) + e2e-external: + name: E2E and DB Tests (external postgres) runs-on: ubuntu-latest timeout-minutes: 30 + services: + postgres: + image: postgres:16 + env: + POSTGRES_PASSWORD: postgres + ports: + - 5432:5432 + options: >- + --health-cmd pg_isready + --health-interval 10s + --health-timeout 5s + --health-retries 5 + + # COMMONS_DB_URL names the maintenance database; COMMONS_DB_CREATE defaults + # to true, so each test carves out and drops its own database on this server. + env: + COMMONS_DB_URL: postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable + CGO_ENABLED: 1 + steps: - name: Checkout code uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 @@ -135,19 +159,27 @@ jobs: - name: Run E2E tests with verbose output run: | go run github.com/onsi/ginkgo/v2/ginkgo@v2.28.0 run -v -r --timeout=10m --poll-progress-after=1m ./e2e - env: - CGO_ENABLED: 1 + + - name: Run DB integration tests + run: go test ./migrate/... ./query/... ./db/... -timeout 15m + + - name: Assert no scratch databases leaked + run: | + leaked=$(psql "$COMMONS_DB_URL" -tAc \ + "SELECT count(*) FROM pg_database WHERE datname LIKE '%\_%\_%' AND datname NOT IN ('postgres','template0','template1')") + echo "leftover databases: $leaked" + test "$leaked" -eq 0 quality-gate: name: Quality Gate runs-on: ubuntu-latest - needs: [e2e-tests] + needs: [e2e-tests, e2e-external] if: always() steps: - name: Check test results run: | - if [ "${{ needs.e2e-tests.result }}" = "failure" ]; then + if [ "${{ needs.e2e-tests.result }}" = "failure" ] || [ "${{ needs.e2e-external.result }}" = "failure" ]; then echo "E2E tests failed" exit 1 fi diff --git a/.github/workflows/query.yml b/.github/workflows/query.yml index 6038ac5..8617c10 100644 --- a/.github/workflows/query.yml +++ b/.github/workflows/query.yml @@ -23,6 +23,25 @@ permissions: read-all jobs: test: runs-on: ubuntu-latest + + services: + postgres: + image: postgres:16 + env: + POSTGRES_PASSWORD: postgres + ports: + - 5432:5432 + options: >- + --health-cmd pg_isready + --health-interval 10s + --health-timeout 5s + --health-retries 5 + + # Without this the module's DB integration tests skip silently. + # COMMONS_DB_CREATE defaults to true, so each test gets its own database. + env: + COMMONS_DB_URL: postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable + steps: - name: Checkout repository uses: actions/checkout@34e114876b0b11c390a56381ad16ebd13914f8d5 # v4 diff --git a/Taskfile.yaml b/Taskfile.yaml index 466bd0a..c0c4081 100644 --- a/Taskfile.yaml +++ b/Taskfile.yaml @@ -32,6 +32,16 @@ tasks: cmds: - go test -v ./... + test:integration: + desc: Run tests including DB integration tests (needs COMMONS_DB_URL, or set COMMONS_DB_EMBEDDED_TEST=1) + cmds: + - | + if [ -z "$COMMONS_DB_URL" ] && [ -z "$COMMONS_DB_EMBEDDED_TEST" ]; then + echo "set COMMONS_DB_URL to an external postgres, or COMMONS_DB_EMBEDDED_TEST=1 to embed one" >&2 + exit 1 + fi + - go test ./... -timeout 15m + tidy: desc: Tidy go modules cmds: 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..899886c 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,27 @@ 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 + } + nested, err := dbcontext.HydrateConnectionByURL(dbcontext.New().WithDB(db, nil), openTelemetry.Connection) + if err != nil { + return fmt.Errorf("resolve nested OpenSearch connection: %w", err) + } + if nested == nil { + return fmt.Errorf("nested OpenSearch connection %q not found", openTelemetry.Connection) + } + 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..cb5ce8a 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,33 @@ func TestConnectionFromBodyRequiresNameAndType(t *testing.T) { } } +func TestValidateOpenTelemetryRequiresNestedOpenSearchConnection(t *testing.T) { + database := connectionInfoTestDB(t) + wrong := models.Connection{ID: uuid.New(), Name: "OS", Type: models.ConnectionTypeHTTP} + search := models.Connection{ID: uuid.New(), Name: "OS", Namespace: "team", Type: models.ConnectionTypeOpenSearch} + if err := database.Create(&wrong).Error; err != nil { + t.Fatal(err) + } + if err := database.Create(&search).Error; err != nil { + t.Fatal(err) + } + candidate := &models.Connection{ + Name: "traces", Type: models.ConnectionTypeOpenTelemetry, + Properties: types.JSONStringMap{"connection": "connection://team/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") + } + candidate.Properties["connection"] = "https://example.test/OS" + if err := validateNestedConnection(database, candidate); err == nil { + t.Fatal("expected an HTTP URL ending in the unrelated connection name 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/go.mod b/cmd/query/go.mod index c0d61ac..1837da7 100644 --- a/cmd/query/go.mod +++ b/cmd/query/go.mod @@ -3,8 +3,9 @@ module github.com/flanksource/commons-db/cmd/query go 1.26.1 require ( - github.com/flanksource/clicky v1.21.37-0.20260715094201-d847a9108f43 - github.com/flanksource/clicky/aichat v1.21.36 + github.com/flanksource/captain v0.0.12-0.20260715131932-f441c766f3fd + github.com/flanksource/clicky v1.21.38-0.20260716031053-78eecd7e914b + github.com/flanksource/clicky/aichat v1.21.38-0.20260716031053-78eecd7e914b github.com/flanksource/clicky/valkey v1.21.34 github.com/flanksource/commons v1.53.1 github.com/flanksource/commons-db v0.1.15 @@ -15,6 +16,7 @@ require ( github.com/spf13/cobra v1.10.2 github.com/stretchr/testify v1.11.1 github.com/valkey-io/valkey-go v1.0.75 + go.yaml.in/yaml/v3 v3.0.4 gorm.io/gorm v1.31.0 k8s.io/api v0.36.1 k8s.io/apimachinery v0.36.1 @@ -127,8 +129,7 @@ require ( github.com/fatih/color v1.18.0 // indirect github.com/felixge/httpsnoop v1.1.0 // indirect github.com/fergusstrange/embedded-postgres v1.34.0 // indirect - github.com/firebase/genkit/go v1.8.0 // indirect - github.com/flanksource/captain v0.0.9 // indirect + github.com/firebase/genkit/go v1.10.0 // indirect github.com/flanksource/gomplate/v3 v3.24.82 // indirect github.com/flanksource/is-healthy v1.0.88 // indirect github.com/flanksource/kubectl-neat v1.0.4 // indirect @@ -197,6 +198,7 @@ require ( github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect github.com/jmespath/go-jmespath v0.4.1-0.20220621161143-b0104c826a24 // indirect + github.com/joho/godotenv v1.5.1 // indirect github.com/josharian/intern v1.0.0 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/kevinburke/ssh_config v1.2.0 // indirect @@ -272,6 +274,7 @@ require ( github.com/samber/lo v1.53.0 // indirect github.com/samber/oops v1.21.0 // indirect github.com/segmentio/asm v1.2.1 // indirect + github.com/segmentio/encoding v0.5.4 // indirect github.com/sergi/go-diff v1.4.0 // indirect github.com/shirou/gopsutil/v3 v3.24.5 // indirect github.com/shoenig/go-m1cpu v0.1.7 // indirect @@ -329,7 +332,6 @@ require ( go.opentelemetry.io/otel/trace 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 gocloud.dev v0.43.0 // indirect golang.org/x/crypto v0.53.0 // indirect golang.org/x/exp v0.0.0-20260410095643-746e56fc9e2f // indirect diff --git a/cmd/query/go.sum b/cmd/query/go.sum index 647152d..6f58266 100644 --- a/cmd/query/go.sum +++ b/cmd/query/go.sum @@ -501,6 +501,8 @@ github.com/jmespath/go-jmespath v0.4.1-0.20220621161143-b0104c826a24 h1:liMMTbpW github.com/jmespath/go-jmespath v0.4.1-0.20220621161143-b0104c826a24/go.mod h1:T8mJZnbsbmF+m6zOOFylbeCJqk5+pHWvzYPziyZiYoo= github.com/jmespath/go-jmespath/internal/testify v1.5.1 h1:shLQSRRSCCPj3f2gpwzGwWFoC7ycTf1rcQZHOlsJ6N8= github.com/jmespath/go-jmespath/internal/testify v1.5.1/go.mod h1:L3OGu8Wl2/fWfCI6z80xFu9LTZmf1ZRjMHUOPmWr69U= +github.com/joho/godotenv v1.5.1 h1:7eLL/+HRGLY0ldzfGMeQkb7vMd0as4CfYvUVzLqw0N0= +github.com/joho/godotenv v1.5.1/go.mod h1:f4LDr5Voq0i2e/R5DDNOoa2zzDfwtkZa6DnEwAbqwq4= github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY= github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y= github.com/joshdk/go-junit v1.0.0 h1:S86cUKIdwBHWwA6xCmFlf3RTLfVXYQfvanM5Uh+K6GE= @@ -686,6 +688,8 @@ github.com/samber/oops v1.21.0 h1:18atcO4oEigNFuGXqr3NZWZ6P0XOSEXyBSAMXdQRxTc= github.com/samber/oops v1.21.0/go.mod h1:Hsm/sKPxtCfPh0w/cE3xVoRfSiE1joDRiStPAsmG9bo= github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0= github.com/segmentio/asm v1.2.1/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs= +github.com/segmentio/encoding v0.5.4 h1:OW1VRern8Nw6ITAtwSZ7Idrl3MXCFwXHPgqESYfvNt0= +github.com/segmentio/encoding v0.5.4/go.mod h1:HS1ZKa3kSN32ZHVZ7ZLPLXWvOVIiZtyJnO1gPH1sKt0= github.com/sergi/go-diff v1.4.0 h1:n/SP9D5ad1fORl+llWyN+D6qoUETXNZARKjyY2/KVCw= github.com/sergi/go-diff v1.4.0/go.mod h1:A0bzQcvG0E7Rwjx0REVgAGH58e96+X0MeOfepqsbeW4= github.com/shirou/gopsutil/v3 v3.24.5 h1:i0t8kL+kQTvpAYToeuiVk3TgDeKOFioZO3Ztz/iZ9pI= diff --git a/cmd/query/internal/app/chat.go b/cmd/query/internal/app/chat.go index 151d48d..d7fa4bb 100644 --- a/cmd/query/internal/app/chat.go +++ b/cmd/query/internal/app/chat.go @@ -1,9 +1,13 @@ package app import ( + "context" "strings" - "github.com/flanksource/clicky/aichat" + captools "github.com/flanksource/captain/pkg/ai/tools" + capchat "github.com/flanksource/captain/pkg/aichat" + "github.com/flanksource/captain/pkg/api" + clickyaichat "github.com/flanksource/clicky/aichat" "github.com/spf13/cobra" ) @@ -11,16 +15,24 @@ import ( // explorer as in-process AI tools. Provider initialization remains lazy: the // server can start without an API key and reports the configuration error only // when chat is used. -func newQueryChatServer(root *cobra.Command) *aichat.Server { - return aichat.NewServer(aichat.Options{ - RootCmd: root, - System: "You are a database operations assistant. Use the available tools " + - "to inspect connections, query profiles, and profile results. Prefer tools " + - "over guessing, never invent connection details, and summarize results clearly.", - Threads: aichat.NewMemThreadStore(), - ToolFilter: isQueryChatTool, - ToolApprovalPolicy: queryToolRequiresApproval, +func newQueryChatServer(root *cobra.Command) (*capchat.Service, error) { + provider, err := clickyaichat.NewCobraToolProvider(clickyaichat.CobraToolProviderOptions{ + Root: root, Filter: isQueryChatTool, Permission: queryToolPermission, }) + if err != nil { + return nil, err + } + return capchat.NewService(capchat.ServiceOptions{ + Settings: capchat.RuntimeSettingsProviderFunc(func(context.Context) (capchat.RuntimeSettings, error) { + return capchat.RuntimeSettings{ + Spec: api.Spec{Model: api.Model{Name: "api:sonnet-5"}}, + System: "You are a database operations assistant. Use the available tools " + + "to inspect connections, query profiles, and profile results. Prefer tools " + + "over guessing, never invent connection details, and summarize results clearly.", + }, nil + }), + Tools: provider, Threads: capchat.NewMemoryThreadStore(), + }), nil } // isQueryChatTool removes query's process-management and long-running @@ -28,8 +40,8 @@ func newQueryChatServer(root *cobra.Command) *aichat.Server { // printing schemas, or blocking on a live trace/top stream is useful on the // CLI but is not a meaningful operation for the in-app assistant (sessions are // managed via the REST API instead). -func isQueryChatTool(tool aichat.ToolInfo) bool { - name := strings.ToLower(strings.TrimSpace(tool.OperationName)) +func isQueryChatTool(tool captools.ToolInfo) bool { + name := strings.ToLower(strings.TrimSpace(tool.Annotation("clicky/operation"))) if name == "" { name = strings.ToLower(strings.TrimSpace(tool.Name)) } @@ -44,15 +56,15 @@ func isQueryChatTool(tool aichat.ToolInfo) bool { // queryToolRequiresApproval auto-runs safe HTTP/read verbs and gates everything // else. The UI may still make a deliberate per-tool On/Ask/Off choice for an // individual request. -func queryToolRequiresApproval(tool aichat.ToolInfo, _ any) bool { - switch strings.ToUpper(strings.TrimSpace(tool.Method)) { +func queryToolPermission(tool captools.ToolInfo) api.ToolMode { + switch strings.ToUpper(strings.TrimSpace(tool.Annotation("clicky/method"))) { case "GET", "HEAD", "OPTIONS": - return false + return api.ToolModeOn } - switch strings.ToLower(strings.TrimSpace(tool.ClickyVerb)) { + switch strings.ToLower(strings.TrimSpace(tool.Annotation("clicky/verb"))) { case "get", "list": - return false + return api.ToolModeOn default: - return true + return api.ToolModeAsk } } diff --git a/cmd/query/internal/app/chat_test.go b/cmd/query/internal/app/chat_test.go index 04499e0..e67c9f0 100644 --- a/cmd/query/internal/app/chat_test.go +++ b/cmd/query/internal/app/chat_test.go @@ -6,20 +6,21 @@ import ( "net/http/httptest" "testing" - "github.com/flanksource/clicky/aichat" + captools "github.com/flanksource/captain/pkg/ai/tools" + "github.com/flanksource/captain/pkg/api" "github.com/spf13/cobra" ) func TestIsQueryChatTool(t *testing.T) { tests := []struct { name string - tool aichat.ToolInfo + tool captools.ToolInfo want bool }{ - {name: "connection list", tool: aichat.ToolInfo{OperationName: "connection"}, want: true}, - {name: "dynamic profile", tool: aichat.ToolInfo{Name: "profile-orders"}, want: true}, - {name: "serve", tool: aichat.ToolInfo{OperationName: "serve"}, want: false}, - {name: "schema fallback name", tool: aichat.ToolInfo{Name: "schema"}, want: false}, + {name: "connection list", tool: captools.ToolInfo{Annotations: map[string]string{"clicky/operation": "connection"}}, want: true}, + {name: "dynamic profile", tool: captools.ToolInfo{Name: "profile-orders"}, want: true}, + {name: "serve", tool: captools.ToolInfo{Annotations: map[string]string{"clicky/operation": "serve"}}, want: false}, + {name: "schema fallback name", tool: captools.ToolInfo{Name: "schema"}, want: false}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { @@ -30,22 +31,22 @@ func TestIsQueryChatTool(t *testing.T) { } } -func TestQueryToolRequiresApproval(t *testing.T) { +func TestQueryToolPermission(t *testing.T) { tests := []struct { name string - tool aichat.ToolInfo - want bool + tool captools.ToolInfo + want api.ToolMode }{ - {name: "get method", tool: aichat.ToolInfo{Method: "GET"}, want: false}, - {name: "head method", tool: aichat.ToolInfo{Method: "head"}, want: false}, - {name: "list verb", tool: aichat.ToolInfo{ClickyVerb: "list"}, want: false}, - {name: "post method", tool: aichat.ToolInfo{Method: "POST"}, want: true}, - {name: "unknown defaults safe", tool: aichat.ToolInfo{}, want: true}, + {name: "get method", tool: captools.ToolInfo{Annotations: map[string]string{"clicky/method": "GET"}}, want: api.ToolModeOn}, + {name: "head method", tool: captools.ToolInfo{Annotations: map[string]string{"clicky/method": "head"}}, want: api.ToolModeOn}, + {name: "list verb", tool: captools.ToolInfo{Annotations: map[string]string{"clicky/verb": "list"}}, want: api.ToolModeOn}, + {name: "post method", tool: captools.ToolInfo{Annotations: map[string]string{"clicky/method": "POST"}}, want: api.ToolModeAsk}, + {name: "unknown defaults safe", tool: captools.ToolInfo{}, want: api.ToolModeAsk}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - if got := queryToolRequiresApproval(tt.tool, nil); got != tt.want { - t.Fatalf("queryToolRequiresApproval(%+v) = %v, want %v", tt.tool, got, tt.want) + if got := queryToolPermission(tt.tool); got != tt.want { + t.Fatalf("queryToolPermission(%+v) = %v, want %v", tt.tool, got, tt.want) } }) } @@ -63,8 +64,10 @@ func TestQueryChatServerCatalogFiltersProcessCommands(t *testing.T) { &cobra.Command{Use: "serve", Run: func(*cobra.Command, []string) {}}, &cobra.Command{Use: "schema", Run: func(*cobra.Command, []string) {}}, ) - chat := newQueryChatServer(root) - t.Cleanup(func() { _ = chat.Close() }) + chat, err := newQueryChatServer(root) + if err != nil { + t.Fatalf("newQueryChatServer: %v", err) + } req := httptest.NewRequest(http.MethodGet, "/api/chat/tools", nil) res := httptest.NewRecorder() 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..efa9a0c 100644 --- a/cmd/query/internal/app/migrations_test.go +++ b/cmd/query/internal/app/migrations_test.go @@ -2,31 +2,21 @@ package app import ( "context" - "os" - "path/filepath" "testing" "github.com/flanksource/commons-db/cmd/query/profiles" - commonsdb "github.com/flanksource/commons-db/db" + "github.com/flanksource/commons-db/dbtest" "github.com/flanksource/commons-db/models" "github.com/flanksource/commons-db/query" "github.com/stretchr/testify/require" ) func TestMigrateSchemaAndProfileStore(t *testing.T) { - if os.Getenv("COMMONS_DB_EMBEDDED_TEST") == "" { - t.Skip("set COMMONS_DB_EMBEDDED_TEST=1 to run embedded-postgres integration tests") - } - dsn, stop, err := commonsdb.StartEmbedded(commonsdb.EmbeddedConfig{ - DataDir: filepath.Join(t.TempDir(), "postgres"), - Database: "query_migrate", - }) - require.NoError(t, err) - t.Cleanup(func() { require.NoError(t, stop()) }) + // A fresh, un-migrated database is the premise of this test: it fabricates a + // pre-HCL schema and asserts the migrator upgrades it in place. + handle := dbtest.ForT(t, dbtest.Options{Name: "query_migrate", LogName: "query-migrate-test"}) + dsn, gdb := handle.DSN(), handle.Gorm() - gdb, pool, err := commonsdb.SetupDB(dsn, "query-migrate-test") - require.NoError(t, err) - t.Cleanup(pool.Close) pgulidSQL, err := querySchema.ReadFile("migrations/001_generate_ulid.sql") require.NoError(t, err) require.NoError(t, gdb.Exec(string(pgulidSQL)).Error) @@ -36,7 +26,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 +37,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/internal/app/server.go b/cmd/query/internal/app/server.go index c7bd77b..229a5b4 100644 --- a/cmd/query/internal/app/server.go +++ b/cmd/query/internal/app/server.go @@ -119,8 +119,10 @@ func (a *App) Serve(parent context.Context, root *cobra.Command, configDir strin return err } mux.Handle("/api/openapi.json", openAPI) - chat := newQueryChatServer(root) - defer func() { _ = chat.Close() }() + chat, err := newQueryChatServer(root) + if err != nil { + return err + } mux.Handle("/api/chat", chat.Handler()) mux.Handle("/api/chat/", chat.Handler()) mux.Handle("/api/", serverMux) diff --git a/cmd/query/profiles/execution.go b/cmd/query/profiles/execution.go index fe13a4f..a5abd25 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,70 @@ 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 + } + if resolved.Profile.Provider.Type != models.ConnectionTypeOpenTelemetry { + http.Error(w, fmt.Sprintf("profile %q has provider type %q, expected %q", name, resolved.Profile.Provider.Type, models.ConnectionTypeOpenTelemetry), http.StatusBadRequest) + 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..f6a7fb2 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,106 @@ 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 TestExecHandlerRejectsMappingForNonOpenTelemetryProfile(t *testing.T) { + store, err := NewFileStore(t.TempDir()) + if err != nil { + t.Fatal(err) + } + if err := store.Save(context.Background(), query.Profile{ + Name: "sql", + Provider: query.ProviderConfig{Type: "postgres"}, + }); err != nil { + t.Fatal(err) + } + handler := newExecHandler("/api/v1", dbcontext.New(), store, &nextMarker{}) + request := httptest.NewRequest(http.MethodPut, "/api/v1/profile/sql/connection", bytes.NewBufferString(`{"connection":"connection://traces"}`)) + response := httptest.NewRecorder() + handler.ServeHTTP(response, request) + if response.Code != http.StatusBadRequest { + t.Fatalf("status=%d body=%s", response.Code, response.Body.String()) + } + if !strings.Contains(response.Body.String(), `expected "opentelemetry"`) { + t.Fatalf("unexpected response: %s", response.Body.String()) + } +} + +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/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..d46a67a --- /dev/null +++ b/cmd/query/profiles/resolver.go @@ -0,0 +1,174 @@ +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) + } + previousType := result.Profile.Provider.Type + previousConnection := result.Profile.Provider.Connection + result.Profile = mergeProfile(result.Profile, imported.Profile) + if imported.Profile.Provider.Type != "" && + (result.ConnectionProfile == "" || result.Profile.Provider.Type != previousType || result.Profile.Provider.Connection != previousConnection) { + 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.Top = nil + merged.Trace = overlay.Trace + } + if overlay.Top != nil { + merged.Trace = 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..d1dae47 --- /dev/null +++ b/cmd/query/profiles/resolver_test.go @@ -0,0 +1,106 @@ +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"))) + }) + + It("keeps the first connection owner when a later import has no connection", func() { + store := resolverStore{ + "owner": {Name: "owner", Provider: query.ProviderConfig{Type: "opentelemetry", Connection: "connection://traces"}}, + "overlay": {Name: "overlay", Provider: query.ProviderConfig{Type: "opentelemetry"}}, + "profile": {Name: "profile", Imports: []string{"owner", "overlay"}}, + } + + resolved, err := Resolve(context.Background(), store, "profile") + Expect(err).ToNot(HaveOccurred()) + Expect(resolved.ConnectionProfile).To(Equal("owner")) + }) + + It("clears the inherited session kind when an overlay selects the other kind", func() { + merged := mergeProfile( + query.Profile{Trace: &query.TraceSpec{}}, + query.Profile{Top: &query.TopSpec{}}, + ) + Expect(merged.Trace).To(BeNil()) + Expect(merged.Top).ToNot(BeNil()) + + merged = mergeProfile( + query.Profile{Top: &query.TopSpec{}}, + query.Profile{Trace: &query.TraceSpec{}}, + ) + Expect(merged.Top).To(BeNil()) + Expect(merged.Trace).ToNot(BeNil()) + }) +}) 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/store_test.go b/cmd/query/sessions/store_test.go index 6c4a24d..385d708 100644 --- a/cmd/query/sessions/store_test.go +++ b/cmd/query/sessions/store_test.go @@ -1,12 +1,10 @@ package sessions import ( - "os" - "path/filepath" "testing" "time" - commonsdb "github.com/flanksource/commons-db/db" + "github.com/flanksource/commons-db/dbtest" "github.com/flanksource/commons-db/query" "github.com/stretchr/testify/require" "gorm.io/gorm" @@ -14,19 +12,7 @@ import ( func startSessionStoreDB(t *testing.T) *gorm.DB { t.Helper() - if os.Getenv("COMMONS_DB_EMBEDDED_TEST") == "" { - t.Skip("set COMMONS_DB_EMBEDDED_TEST=1 to run embedded-postgres integration tests") - } - dsn, stop, err := commonsdb.StartEmbedded(commonsdb.EmbeddedConfig{ - DataDir: filepath.Join(t.TempDir(), "postgres"), - Database: "query_sessions", - }) - require.NoError(t, err) - t.Cleanup(func() { require.NoError(t, stop()) }) - - gdb, pool, err := commonsdb.SetupDB(dsn, "query-sessions-test") - require.NoError(t, err) - t.Cleanup(pool.Close) + gdb := dbtest.ForT(t, dbtest.Options{Name: "query_sessions", LogName: "query-sessions-test"}).Gorm() require.NoError(t, gdb.WithContext(t.Context()).AutoMigrate(&sessionRecord{}, &sessionEventRecord{})) return gdb } 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/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/buildProfileAction.tsx b/cmd/query/www/src/buildProfileAction.tsx index f7db95e..f5c63a6 100644 --- a/cmd/query/www/src/buildProfileAction.tsx +++ b/cmd/query/www/src/buildProfileAction.tsx @@ -1,8 +1,6 @@ import { Button, Icon, - Modal, - SchemaActionForm, useOperations, type OperationsApiClient, type ResolvedOperation, @@ -10,10 +8,7 @@ import { import { UiMagicWand } from "@flanksource/clicky-ui/icons"; import { useQueryClient } from "@tanstack/react-query"; import { useMemo, useState } from "react"; -import { - ProfileBuilderAutoOpen, - profileBuilderFormExtensions, -} from "./profileBuilder"; +import { ProfileWizard } from "./profileWizard"; type BuildProfileButtonProps = { client: OperationsApiClient; @@ -86,43 +81,20 @@ export function BuildProfileButton({ {open && createAction ? ( - - { - setOpen(false); - await Promise.all([ - queryClient.invalidateQueries({ queryKey: ["openapi-spec"] }), - queryClient.invalidateQueries({ - queryKey: ["operation-list"], - }), - queryClient.invalidateQueries({ - queryKey: ["logs-entity-names"], - }), - ]); - }} - renderLayout={({ body, footer }) => ( - setOpen(false)} - title="Build Profile" - size="lg" - {...(footer ? { footer } : {})} - > - {body} - - )} - fallback={ -
- The profile form is unavailable. -
- } - /> - + setOpen(false)} + onSuccess={async () => { + setOpen(false); + await Promise.all([ + queryClient.invalidateQueries({ queryKey: ["openapi-spec"] }), + queryClient.invalidateQueries({ queryKey: ["operation-list"] }), + queryClient.invalidateQueries({ queryKey: ["logs-entity-names"] }), + ]); + }} + /> ) : null} ); diff --git a/cmd/query/www/src/chatWidget.tsx b/cmd/query/www/src/chatWidget.tsx index 8caab33..4517738 100644 --- a/cmd/query/www/src/chatWidget.tsx +++ b/cmd/query/www/src/chatWidget.tsx @@ -37,7 +37,6 @@ export function ChatWidget({ client }: { client: OperationsApiClient }) { chat={{ api: "/api/chat", modelsApi: "/api/chat/models", - toolApproval: "manual", suggestions: [ "List configured connections", "Show available query profiles", 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/cmd/query/www/src/profileConnectionMapping.test.ts b/cmd/query/www/src/profileConnectionMapping.test.ts new file mode 100644 index 0000000..2188101 --- /dev/null +++ b/cmd/query/www/src/profileConnectionMapping.test.ts @@ -0,0 +1,45 @@ +import { describe, expect, it } from "vitest"; +import type { ResolvedOperation } from "@flanksource/clicky-ui"; +import { + findConnectionCreateOperation, + isProfileConnectionRequired, + profileConnectionOptions, + rejectPendingMapping, +} 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" }, + ]); + }); + + it("rejects an existing mapping request before replacing it", () => { + const error = new Error("mapping conflict"); + let rejectedWith: unknown; + rejectPendingMapping({ error, reject: (value) => { rejectedWith = value; } }); + expect(rejectedWith).toBe(error); + }); +}); diff --git a/cmd/query/www/src/profileConnectionMapping.tsx b/cmd/query/www/src/profileConnectionMapping.tsx new file mode 100644 index 0000000..2735d78 --- /dev/null +++ b/cmd/query/www/src/profileConnectionMapping.tsx @@ -0,0 +1,282 @@ +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 rejectPendingMapping(current: Pick | null): void { + current?.reject(current.error); +} + +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((current) => { + rejectPendingMapping(current); + return { 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} + + {connections.isError ? ( + + ) : options.length === 0 && !connections.isLoading ? ( + + ) : ( + + )} + + } + > + {connections.isLoading ? ( +
Loading OpenTelemetry connections…
+ ) : connections.isError ? ( +
+ Unable to load OpenTelemetry connections:{" "} + {connections.error instanceof Error ? connections.error.message : String(connections.error)} +
+ ) : 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, + }) + } + /> + +
+ +