diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 0000000..b4e018b --- /dev/null +++ b/.dockerignore @@ -0,0 +1,11 @@ +# Keep the Docker build context small for the Go-service images +# (config-manager, datasource-gateway). These builds only need the +# Makefile, Go sources, and configs — not JS deps, build artifacts, or VCS. +.git +.gitignore +**/node_modules +**/dist +bin +data +temp_path +*.log diff --git a/.gitmodules b/.gitmodules deleted file mode 100644 index 9500d07..0000000 --- a/.gitmodules +++ /dev/null @@ -1,4 +0,0 @@ -[submodule "mfork-grafana-plugin"] - path = mfork-grafana-plugin - url = https://github.com/couchbaselabs/mfork-grafana-plugin.git - ignore = dirty diff --git a/Makefile b/Makefile index d19580a..6de4299 100644 --- a/Makefile +++ b/Makefile @@ -1,13 +1,23 @@ .PHONY: build clean test lint help # Default target: Build all services -build: build-cm +build: build-cm build-gateway # Build the config-manager service build-cm: @echo "Building config-manager service..." @cd config-manager && go build -o ../bin/config-manager . +# Build the datasource-gateway service +build-gateway: + @echo "Building datasource-gateway service..." + @cd datasource-gateway && go build -o ../bin/datasource-gateway . + +# Build the datasource-gateway service docker image (standalone compose project) +build-gateway-docker: build-gateway + @echo "Building datasource-gateway service docker image..." + @docker compose -f deployments/docker/compose.datasource-gateway.yml up --build -d + # Build the config-manager service docker image build-cm-docker: build-cm @echo "Building config-manager service docker image..." @@ -27,28 +37,15 @@ build-plugin: mage # Build the grafana-app plugin docker image -build-plugin-docker: build-plugin move-datasource-build-artifacts +build-plugin-docker: build-plugin @echo "Building cbmonitor grafana-app plugin docker image..." @cd cbmonitor && npm run server # Rebuild plugin locally and restart container to pick up changes -reload-plugin: build-plugin move-datasource-build-artifacts +reload-plugin: build-plugin @echo "Restarting cbmonitor container to load plugin changes..." @docker restart cbmonitor -# Build the datasource plugin -build-couchbase-datasource: - @cd mfork-grafana-plugin && cd couchbase-datasource && \ - yarn install && \ - yarn build && \ - mage -v - -# Move the datasource build artifacts to the cbmonitor dist directory -move-datasource-build-artifacts: - @echo "Moving datasource build artifacts..." - @mkdir -p cbmonitor/dist/couchbase-datasource/ - @cp -r mfork-grafana-plugin/couchbase-datasource/dist/* cbmonitor/dist/couchbase-datasource/ - # Clean build artifacts clean-cm: @echo "Cleaning config-manager service build artifacts..." @@ -71,6 +68,8 @@ help: @echo " build - Build all services" @echo " build-cm - Build config-manager service" @echo " build-cm-docker - Build config-manager service docker image" + @echo " build-gateway - Build datasource-gateway service" + @echo " build-gateway-docker - Build datasource-gateway service docker image" @echo " build-plugin - Build cbmonitor grafana-app plugin" @echo " build-plugin-docker - Build cbmonitor grafana-app plugin docker image" @echo " reload-plugin - Rebuild plugin and restart container" diff --git a/cbmonitor/.config/docker-compose-base.yaml b/cbmonitor/.config/docker-compose-base.yaml index f18cfe0..17848b6 100644 --- a/cbmonitor/.config/docker-compose-base.yaml +++ b/cbmonitor/.config/docker-compose-base.yaml @@ -25,7 +25,6 @@ services: - ../dist:/var/lib/grafana/plugins/cbmonitor - ../provisioning:/etc/grafana/provisioning - ..:/root/cbmonitor - - ../../couchbase-datasource/couchbase-datasource/dist:/var/lib/grafana/plugins/couchbase-datasource - ../../test-dashboards:/var/lib/grafana/dashboards environment: @@ -43,9 +42,8 @@ services: GF_AUTH_MANAGED_SERVICE_ACCOUNTS_ENABLED: 'true' GF_FEATURE_TOGGLES_ENABLE: ${GF_FEATURE_TOGGLES_ENABLE:-externalServiceAccounts} - # Couchbase credentials are consumed by: - # 1. provisioning/datasources/datasource.yaml (couchbase-datasource plugin) - # 2. provisioning/plugins/apps.yaml + # Couchbase credentials are consumed by provisioning/plugins/apps.yaml + # (the cbmonitor app jsonData). COUCHBASE_CONNECTION_STRING: ${COUCHBASE_CONNECTION_STRING:-couchbase://host.docker.internal:8091} COUCHBASE_USERNAME: ${COUCHBASE_USERNAME:-Administrator} COUCHBASE_PASSWORD: ${COUCHBASE_PASSWORD:-password} diff --git a/cbmonitor/pkg/handlers/prometheus.go b/cbmonitor/pkg/handlers/prometheus.go deleted file mode 100644 index f7ccbd9..0000000 --- a/cbmonitor/pkg/handlers/prometheus.go +++ /dev/null @@ -1,279 +0,0 @@ -package handlers - -import ( - "context" - "encoding/json" - "fmt" - "log" - "net/http" - "time" - - "github.com/couchbase/cbmonitor/pkg/promql" - "github.com/couchbase/cbmonitor/pkg/services" -) - -// couchbaseQuerier captures just the bit of services.CouchbaseService the -// PromQL handler needs. Behind an interface so tests can inject failing -// stubs without standing up a real gocb cluster. -type couchbaseQuerier interface { - ExecuteQuery(ctx context.Context, query string) ([]map[string]interface{}, error) -} - -// PromQLHandler handles Prometheus Query API requests -// https://prometheus.io/docs/prometheus/latest/querying/api/ -type PromQLHandler struct { - couchbaseService couchbaseQuerier -} - -// NewPromQLHandler creates a new PromQL handler. A nil svc is permitted — -// requests will return a clean error rather than panic — so the plugin -// can boot even when the underlying Couchbase service failed to init. -func NewPromQLHandler(svc *services.CouchbaseService) *PromQLHandler { - h := &PromQLHandler{} - if svc != nil { - h.couchbaseService = svc - } - return h -} - -// HandleQuery handles GET /query (instant query) -func (h *PromQLHandler) HandleQuery(w http.ResponseWriter, req *http.Request) { - if req.Method != http.MethodGet { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - - // Parse query parameters - query := req.URL.Query().Get("query") - if query == "" { - h.sendErrorResponse(w, "query parameter is required", http.StatusBadRequest) - return - } - - timeStr := req.URL.Query().Get("time") - if timeStr == "" { - timeStr = fmt.Sprintf("%d", time.Now().Unix()) - } - - // Parse query context (snapshot will come from 'job' label in PromQL) - queryCtx, err := promql.ParseQueryContext(query, timeStr, "", "", "", "") - if err != nil { - h.sendErrorResponse(w, fmt.Sprintf("Invalid query parameters: %v", err), http.StatusBadRequest) - return - } - queryCtx.Context = req.Context() - - // Execute query - result, err := h.executeQuery(queryCtx) - if err != nil { - log.Printf("Query execution error: %v", err) - h.sendErrorResponse(w, fmt.Sprintf("Query execution failed: %v", err), http.StatusInternalServerError) - return - } - - // Send response - h.sendJSONResponse(w, result, http.StatusOK) -} - -// HandleQueryRange handles GET /query_range (range query) -func (h *PromQLHandler) HandleQueryRange(w http.ResponseWriter, req *http.Request) { - if req.Method != http.MethodGet { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - - // Parse query parameters - query := req.URL.Query().Get("query") - if query == "" { - h.sendErrorResponse(w, "query parameter is required", http.StatusBadRequest) - return - } - - startStr := req.URL.Query().Get("start") - endStr := req.URL.Query().Get("end") - stepStr := req.URL.Query().Get("step") - - if startStr == "" || endStr == "" { - h.sendErrorResponse(w, "start and end parameters are required", http.StatusBadRequest) - return - } - - if stepStr == "" { - stepStr = "15s" // Default step - } - - // Parse query context (snapshot will come from 'job' label in PromQL) - queryCtx, err := promql.ParseQueryContext(query, "", startStr, endStr, stepStr, "") - if err != nil { - h.sendErrorResponse(w, fmt.Sprintf("Invalid query parameters: %v", err), http.StatusBadRequest) - return - } - queryCtx.Context = req.Context() - - // Execute query - result, err := h.executeQuery(queryCtx) - if err != nil { - log.Printf("Query range execution error: %v", err) - h.sendErrorResponse(w, fmt.Sprintf("Query execution failed: %v", err), http.StatusInternalServerError) - return - } - - // Send response - h.sendJSONResponse(w, result, http.StatusOK) -} - -// HandleSeries handles GET /api/v1/series (series discovery) -func (h *PromQLHandler) HandleSeries(w http.ResponseWriter, req *http.Request) { - if req.Method != http.MethodGet { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - - // Parse query parameters - match := req.URL.Query()["match[]"] // Can be multiple - _ = req.URL.Query().Get("start") // TODO: Use for series discovery - _ = req.URL.Query().Get("end") // TODO: Use for series discovery - - if len(match) == 0 { - h.sendErrorResponse(w, "match[] parameter is required", http.StatusBadRequest) - return - } - - // For now, return a simple response indicating series discovery is not fully implemented - // This would require querying Couchbase metadata to discover available series - result := promql.PrometheusResult{ - Status: "success", - Data: promql.ResultData{ - ResultType: "series", - Result: []interface{}{}, - }, - } - - h.sendJSONResponse(w, result, http.StatusOK) -} - -// executeQuery executes a PromQL query and returns Prometheus-formatted results -func (h *PromQLHandler) executeQuery(queryCtx *promql.QueryContext) (*promql.PrometheusResult, error) { - if h.couchbaseService == nil { - return nil, fmt.Errorf("couchbase metrics service is unavailable (initialization may have failed; check plugin settings and Couchbase connectivity)") - } - - // Parse PromQL query - expr, err := promql.ParseQuery(queryCtx.Query) - if err != nil { - return nil, fmt.Errorf("failed to parse PromQL: %w", err) - } - - // Create query plan (snapshot will be extracted from 'job' label in PromQL) - plan, err := promql.PlanQuery(expr, "") - if err != nil { - return nil, fmt.Errorf("failed to plan query: %w", err) - } - - log.Printf("Query plan: %s", plan.String()) - - // Build SQL++ queries - sqlBuilder := promql.NewSQLBuilder(plan, queryCtx) - sqlQueries, err := sqlBuilder.Build() - if err != nil { - return nil, fmt.Errorf("failed to build SQL++ queries: %w", err) - } - - log.Printf("Generated %d SQL++ queries", len(sqlQueries)) - - // Execute queries against Couchbase, tracking which sub-queries fail. - // Partial failure produces warnings on the response so the client knows - // the data is incomplete; total failure surfaces as an error. - var ( - allResults []promql.QueryResult - failures []error - ) - for i, sqlQuery := range sqlQueries { - log.Printf("Executing query %d: %s", i+1, sqlQuery) - - results, err := h.couchbaseService.ExecuteQuery(queryCtx.Context, sqlQuery) - if err != nil { - log.Printf("Query execution error: %v", err) - failures = append(failures, fmt.Errorf("sub-query %d: %w", i+1, err)) - continue - } - - // Convert results to QueryResult format - for _, row := range results { - queryResult := promql.QueryResult{ - Time: getStringValue(row, "time"), - Value: getValue(row, "value"), - } - - // Extract labels if present - if labels, ok := row["labels"].(map[string]interface{}); ok { - queryResult.Labels = labels - } - - allResults = append(allResults, queryResult) - } - } - - // All sub-queries failed → surface a real error instead of returning - // a misleading empty success that renders as "no data" in panels. - if len(failures) > 0 && len(failures) == len(sqlQueries) { - return nil, fmt.Errorf("all %d sub-queries failed; first error: %w", len(failures), failures[0]) - } - - // Transform results to Prometheus format - result, err := promql.TransformResults(allResults, plan, queryCtx) - if err != nil { - return nil, fmt.Errorf("failed to transform results: %w", err) - } - - // Partial failure → return data but warn the client some sub-queries - // failed, per Prometheus HTTP API conventions. - for _, f := range failures { - result.Warnings = append(result.Warnings, f.Error()) - } - - return result, nil -} - -// Helper functions for extracting values from query results -func getStringValue(row map[string]interface{}, key string) string { - if val, ok := row[key]; ok { - if str, ok := val.(string); ok { - return str - } - return fmt.Sprintf("%v", val) - } - return "" -} - -func getValue(row map[string]interface{}, key string) interface{} { - if val, ok := row[key]; ok { - return val - } - return nil -} - -// sendJSONResponse sends a JSON response -func (h *PromQLHandler) sendJSONResponse(w http.ResponseWriter, data interface{}, statusCode int) { - w.Header().Set("Content-Type", "application/json") - w.Header().Set("Access-Control-Allow-Origin", "*") - w.Header().Set("Access-Control-Allow-Methods", "GET, OPTIONS") - w.Header().Set("Access-Control-Allow-Headers", "Content-Type") - - w.WriteHeader(statusCode) - - if err := json.NewEncoder(w).Encode(data); err != nil { - log.Printf("Error encoding JSON response: %v", err) - http.Error(w, "Internal server error", http.StatusInternalServerError) - } -} - -// sendErrorResponse sends an error response -func (h *PromQLHandler) sendErrorResponse(w http.ResponseWriter, message string, statusCode int) { - response := promql.PrometheusResult{ - Status: "error", - Error: message, - ErrorType: "bad_data", - } - h.sendJSONResponse(w, response, statusCode) -} diff --git a/cbmonitor/pkg/handlers/prometheus_test.go b/cbmonitor/pkg/handlers/prometheus_test.go deleted file mode 100644 index 038bcd8..0000000 --- a/cbmonitor/pkg/handlers/prometheus_test.go +++ /dev/null @@ -1,89 +0,0 @@ -package handlers - -import ( - "context" - "errors" - "strings" - "testing" - - "github.com/couchbase/cbmonitor/pkg/promql" -) - -// fakeQuerier is a couchbaseQuerier stub: each invocation pops the next -// scripted result from results, with errs supplying the matching errors. -// Used to exercise the PromQL handler's per-sub-query failure tracking -// without standing up a real gocb cluster. -type fakeQuerier struct { - calls int - results [][]map[string]interface{} - errs []error -} - -func (f *fakeQuerier) ExecuteQuery(ctx context.Context, query string) ([]map[string]interface{}, error) { - i := f.calls - f.calls++ - var rows []map[string]interface{} - var err error - if i < len(f.results) { - rows = f.results[i] - } - if i < len(f.errs) { - err = f.errs[i] - } - return rows, err -} - -// runExecuteQuery exercises the handler's executeQuery with a fake -// querier. Returns the result + error verbatim so callers can assert on either path. -func runExecuteQuery(t *testing.T, q couchbaseQuerier, promQL string) (*promql.PrometheusResult, error) { - t.Helper() - h := &PromQLHandler{couchbaseService: q} - // PromQL parsing requires a real metric name + timestamp; use a - // simple instant query. - queryCtx, err := promql.ParseQueryContext(promQL, "1700000000", "", "", "", "") - if err != nil { - t.Fatalf("ParseQueryContext: %v", err) - } - queryCtx.Context = context.Background() - return h.executeQuery(queryCtx) -} - -func TestPromQLExecute_AllSubQueriesFailedReturnsError(t *testing.T) { - // Force the underlying querier to fail every call. The handler must - // not return a misleading "success with empty data" — it should - // surface the failure so panels show an error instead of silently - // rendering "no data". - q := &fakeQuerier{ - errs: []error{ - errors.New("boom1"), - errors.New("boom2"), - errors.New("boom3"), - }, - } - _, err := runExecuteQuery(t, q, `up`) - if err == nil { - t.Fatalf("expected error when all sub-queries fail; got nil") - } - if !strings.Contains(err.Error(), "sub-queries failed") { - t.Errorf("error should mention sub-query failures, got: %v", err) - } -} - -func TestPromQLExecute_NilServiceReturnsError(t *testing.T) { - // A nil-couchbase-service handler (init failed but routes are still - // registered because the feature toggle is on) must return a clear - // error rather than panic with a nil-pointer dereference. - h := &PromQLHandler{couchbaseService: nil} - queryCtx, err := promql.ParseQueryContext(`up`, "1700000000", "", "", "", "") - if err != nil { - t.Fatalf("ParseQueryContext: %v", err) - } - queryCtx.Context = context.Background() - _, err = h.executeQuery(queryCtx) - if err == nil { - t.Fatalf("expected error when couchbaseService is nil") - } - if !strings.Contains(err.Error(), "unavailable") { - t.Errorf("error should explain the service is unavailable, got: %v", err) - } -} diff --git a/cbmonitor/pkg/plugin/datasources.go b/cbmonitor/pkg/plugin/datasources.go index 397a2d4..5a02eac 100644 --- a/cbmonitor/pkg/plugin/datasources.go +++ b/cbmonitor/pkg/plugin/datasources.go @@ -27,6 +27,12 @@ const ( // Any UID in this set that isn't in the current `desired` list during a // reconcile pass gets DELETEd from Grafana — that's how toggling a // feature off in app settings cleans up its datasource. +// +// dsUIDCouchbase is retained here even though desiredDatasources no longer +// produces it: the gateway replaced the standalone couchbase-datasource, so +// the reconciler must DELETE any pre-existing cbdatasource left over from +// before the gateway. Drop it (and the constant + IAM scopes) in a future +// release once deployments have reconciled it away. var appManagedUIDs = []string{dsUIDPrometheus, dsUIDCouchbase} // reconcileTimeout caps a single reconciliation pass. Long enough to handle @@ -364,9 +370,10 @@ func readBodySnippet(body io.Reader) string { // uses this to distinguish "feature disabled — clean up the DS" from // "feature enabled but misconfigured — leave the existing DS alone". func (s *PluginSettings) claimedDatasources() map[string]bool { + // cbdatasource is intentionally absent: it's retired, so it's never + // "claimed" and the reconciler's delete phase always removes any leftover. return map[string]bool{ dsUIDPrometheus: s.PrometheusDatasource.Enabled, - dsUIDCouchbase: s.CouchbaseDatasource.Enabled, } } @@ -376,13 +383,23 @@ func (s *PluginSettings) claimedDatasources() map[string]bool { // as "skipped", not as a hard error. func (s *PluginSettings) desiredDatasources() []DesiredDatasource { out := []DesiredDatasource{} - if s.PrometheusDatasource.Enabled && s.PrometheusDatasource.URL != "" { + + // The single Prometheus-typed datasource points at the gateway sidecar when + // the gateway is enabled (so it can translate Couchbase-backed snapshots and + // serve overlap); otherwise straight at the upstream Prometheus/Mimir + // (pure-Prometheus mode). Either way it's a prometheus-typed datasource + // speaking the Prometheus HTTP API, so the jsonData below is unchanged. + promURL := s.PrometheusDatasource.URL + if s.Gateway.Enabled { + promURL = s.Gateway.URL + } + if s.PrometheusDatasource.Enabled && promURL != "" { out = append(out, DesiredDatasource{ UID: dsUIDPrometheus, Name: "Prometheus", Type: "prometheus", Access: "proxy", - URL: s.PrometheusDatasource.URL, + URL: promURL, IsDefault: s.PrometheusDatasource.IsDefault, JSONData: map[string]any{ "cacheLevel": "High", @@ -392,38 +409,8 @@ func (s *PluginSettings) desiredDatasources() []DesiredDatasource { }, }) } - if s.CouchbaseDatasource.Enabled && s.CouchbaseServer.ConnectionString != "" { - ds := s.CouchbaseDatasource - // The couchbase-datasource plugin (and any UDFs the panels call - // through) read bucket/scope/collection from jsonData. Forward - // every field the app captures so panel queries don't need to - // hardcode location info. - jsonData := map[string]any{ - "host": s.CouchbaseServer.ConnectionString, - "username": s.CouchbaseServer.Username, - "bucket": ds.Bucket, - } - if ds.Scope != "" { - jsonData["scope"] = ds.Scope - } - if ds.Collection != "" { - jsonData["collection"] = ds.Collection - } - out = append(out, DesiredDatasource{ - UID: dsUIDCouchbase, - Name: "cbdatasource", - Type: "couchbase-datasource", - Access: "proxy", - // The couchbase-datasource plugin reads `host` from jsonData, - // not the top-level `url` — but Grafana's API requires `url` - // to be set on all datasources. Send the connection string in - // both so the wire object is well-formed. - URL: s.CouchbaseServer.ConnectionString, - JSONData: jsonData, - SecureJSONData: map[string]string{ - "password": s.CouchbaseServer.Password, - }, - }) - } + // cbdatasource is no longer created: Couchbase-backed snapshots are served + // by the gateway (the prometheus DS above points at it). Any pre-existing + // cbdatasource is orphan-deleted via appManagedUIDs. return out } diff --git a/cbmonitor/pkg/plugin/datasources_test.go b/cbmonitor/pkg/plugin/datasources_test.go index 2e683fc..da84e60 100644 --- a/cbmonitor/pkg/plugin/datasources_test.go +++ b/cbmonitor/pkg/plugin/datasources_test.go @@ -469,42 +469,38 @@ func TestReconcile_PreservesClaimedDatasourceWithMissingConfig(t *testing.T) { } func TestClaimedDatasources_ReflectsEnabledFlags(t *testing.T) { - // claimedDatasources() ignores URL/connection-string presence — it's - // purely about user intent. Verifies the hardening pivot: "did the - // user ask for this feature to be on?" rather than "is it currently - // reconcilable?". + // claimedDatasources() ignores URL presence — it's purely about user + // intent ("did the user ask for this feature on?"). cbdatasource is retired + // and must never be claimed, so its leftover always gets orphan-deleted. cases := []struct { - name string - s PluginSettings - want map[string]bool + name string + s PluginSettings + wantProm bool }{ { - name: "both enabled, both URLs/strings empty", - s: PluginSettings{ - PrometheusDatasource: PrometheusDatasourceSettings{Enabled: true}, - CouchbaseDatasource: CouchbaseDatasourceSettings{Enabled: true}, - }, - want: map[string]bool{dsUIDPrometheus: true, dsUIDCouchbase: true}, + name: "prom enabled, url empty", + s: PluginSettings{PrometheusDatasource: PrometheusDatasourceSettings{Enabled: true}}, + wantProm: true, }, { - name: "prom enabled, couchbase disabled", - s: PluginSettings{ - PrometheusDatasource: PrometheusDatasourceSettings{Enabled: true, URL: "http://x"}, - CouchbaseDatasource: CouchbaseDatasourceSettings{Enabled: false}, - }, - want: map[string]bool{dsUIDPrometheus: true, dsUIDCouchbase: false}, + name: "prom enabled with url", + s: PluginSettings{PrometheusDatasource: PrometheusDatasourceSettings{Enabled: true, URL: "http://x"}}, + wantProm: true, }, { - name: "neither enabled", - s: PluginSettings{}, - want: map[string]bool{dsUIDPrometheus: false, dsUIDCouchbase: false}, + name: "prom disabled", + s: PluginSettings{}, + wantProm: false, }, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { got := tc.s.claimedDatasources() - if got[dsUIDPrometheus] != tc.want[dsUIDPrometheus] || got[dsUIDCouchbase] != tc.want[dsUIDCouchbase] { - t.Errorf("claimedDatasources() = %v, want %v", got, tc.want) + if got[dsUIDPrometheus] != tc.wantProm { + t.Errorf("claimed[prometheus] = %v, want %v", got[dsUIDPrometheus], tc.wantProm) + } + if got[dsUIDCouchbase] { + t.Error("cbdatasource must not be claimed (retired)") } }) } @@ -531,18 +527,19 @@ func TestDesiredDatasources_OnlyIncludesEnabledWithRequiredFields(t *testing.T) wantUIDs: []string{}, }, { - name: "both enabled with required fields", + name: "both features enabled — only the prometheus DS (cbdatasource retired)", s: PluginSettings{ PrometheusDatasource: PrometheusDatasourceSettings{Enabled: true, URL: "http://prom"}, CouchbaseDatasource: CouchbaseDatasourceSettings{Enabled: true, Bucket: "cbmonitor"}, CouchbaseServer: CouchbaseServerSettings{ConnectionString: "couchbase://x"}, }, - wantUIDs: []string{"prometheus", "cbdatasource"}, + wantUIDs: []string{"prometheus"}, }, { - name: "couchbase enabled but connection string missing — skipped", + name: "couchbase enabled, prometheus off — no datasource (cbdatasource retired)", s: PluginSettings{ CouchbaseDatasource: CouchbaseDatasourceSettings{Enabled: true, Bucket: "cbmonitor"}, + CouchbaseServer: CouchbaseServerSettings{ConnectionString: "couchbase://x"}, }, wantUIDs: []string{}, }, @@ -561,11 +558,12 @@ func TestDesiredDatasources_OnlyIncludesEnabledWithRequiredFields(t *testing.T) } } -func TestDesiredDatasources_PropagatesBucketScopeCollectionToCbdatasource(t *testing.T) { - // The whole point of these settings is for panel queries (and any - // UDFs) to know where the timeseries data lives. Verify they actually - // land in the cbdatasource JSONData the reconciler sends to Grafana. +func TestDesiredDatasources_NeverCreatesCbdatasource(t *testing.T) { + // cbdatasource is retired: even with the Couchbase datasource feature fully + // enabled, no couchbase-datasource entry is produced — the gateway serves + // Couchbase-backed snapshots through the single prometheus datasource. s := PluginSettings{ + PrometheusDatasource: PrometheusDatasourceSettings{Enabled: true, URL: "http://prom:9090"}, CouchbaseDatasource: CouchbaseDatasourceSettings{ Enabled: true, Bucket: "metrics-bucket", @@ -577,44 +575,40 @@ func TestDesiredDatasources_PropagatesBucketScopeCollectionToCbdatasource(t *tes Username: "u", }, } - got := s.desiredDatasources() - if len(got) != 1 || got[0].UID != "cbdatasource" { - t.Fatalf("expected exactly one cbdatasource desired entry, got %#v", got) + for _, d := range s.desiredDatasources() { + if d.Type == "couchbase-datasource" || d.UID == dsUIDCouchbase { + t.Errorf("cbdatasource should no longer be produced, got %#v", d) + } } - jd := got[0].JSONData - if jd["bucket"] != "metrics-bucket" { - t.Errorf("bucket not propagated, got %v", jd["bucket"]) +} + +func TestDesiredDatasources_PrometheusTargetsGatewayWhenEnabled(t *testing.T) { + base := PluginSettings{ + PrometheusDatasource: PrometheusDatasourceSettings{Enabled: true, URL: "http://mimir:9009/prometheus", IsDefault: true}, } - if jd["scope"] != "metrics-scope" { - t.Errorf("scope not propagated, got %v", jd["scope"]) + + // Pure-Prometheus mode: the datasource targets the upstream Mimir directly. + got := base.desiredDatasources() + if len(got) != 1 || got[0].UID != dsUIDPrometheus { + t.Fatalf("expected one prometheus DS, got %#v", got) } - if jd["collection"] != "timeseries" { - t.Errorf("collection not propagated, got %v", jd["collection"]) + if got[0].URL != "http://mimir:9009/prometheus" { + t.Errorf("pure-Prometheus URL = %q, want the Mimir URL", got[0].URL) } -} -func TestDesiredDatasources_OmitsEmptyScopeCollection(t *testing.T) { - // When scope/collection are blank (user wants defaults), the - // JSONData shouldn't carry empty strings — the couchbase-datasource - // plugin / UDFs should see them as absent, not as "" (which can - // trigger different code paths). - s := PluginSettings{ - CouchbaseDatasource: CouchbaseDatasourceSettings{ - Enabled: true, - Bucket: "metrics-bucket", - }, - CouchbaseServer: CouchbaseServerSettings{ - ConnectionString: "couchbase://cb", - Username: "u", - }, + // Gateway enabled: the same datasource now targets the gateway URL, still + // as a prometheus-typed datasource. + withGW := base + withGW.Gateway = GatewaySettings{Enabled: true, URL: "http://datasource-gateway:8090"} + got = withGW.desiredDatasources() + if len(got) != 1 || got[0].UID != dsUIDPrometheus { + t.Fatalf("expected one prometheus DS, got %#v", got) } - got := s.desiredDatasources() - jd := got[0].JSONData - if _, ok := jd["scope"]; ok { - t.Errorf("scope should be omitted when blank, got %v", jd["scope"]) + if got[0].URL != "http://datasource-gateway:8090" { + t.Errorf("gateway-mode URL = %q, want the gateway URL", got[0].URL) } - if _, ok := jd["collection"]; ok { - t.Errorf("collection should be omitted when blank, got %v", jd["collection"]) + if got[0].Type != "prometheus" { + t.Errorf("DS type = %q, want prometheus", got[0].Type) } } diff --git a/cbmonitor/pkg/plugin/resources.go b/cbmonitor/pkg/plugin/resources.go index ad8d5e8..2894a82 100644 --- a/cbmonitor/pkg/plugin/resources.go +++ b/cbmonitor/pkg/plugin/resources.go @@ -63,6 +63,9 @@ func (a *App) handleGetDatasourceConfig(w http.ResponseWriter, req *http.Request "defaultDataSource": a.settings.DefaultDataSource(), "prometheusAvailable": a.settings.PrometheusDatasource.Enabled, "couchbaseAvailable": a.settings.CouchbaseDatasource.Enabled, + "gatewayEnabled": a.settings.Gateway.Enabled, + "gatewayUrl": a.settings.Gateway.URL, + "overlapEnabled": a.settings.Gateway.Enabled && a.settings.Gateway.Overlap, "reconciliation": a.getReconcileState(), "settings": settingsBlock, } @@ -127,10 +130,6 @@ func (a *App) registerRoutes(mux *http.ServeMux) { mux.HandleFunc("/admin/reconcile-dashboards", a.handleReconcileDashboards) mux.HandleFunc("/healthcheck/connection", a.handleHealthCheckConnection) - if a.settings.CouchbaseDatasource.Enabled { - a.setupPrometheusRoutes(mux) - } - if a.settings.Snapshots.Enabled || a.settings.PrometheusDatasource.Enabled { a.setupSnapshotRoutes(mux) } @@ -186,19 +185,6 @@ func (a *App) setupSnapshotRoutes(mux *http.ServeMux) { log.Printf("Snapshot routes registered: /snapshots/{id}, /snapshots/{id}/metric-names, /snapshots/{id}/metrics/{metric}, /snapshots/{id}/metrics/{metric}/phases/{phase}, /snapshots/{id}/annotations/sync") } -// setupPrometheusRoutes registers the Prometheus Query API routes backed -// by the Couchbase datasource bucket. Only called when -// CouchbaseDatasource.Enabled is true. Service is owned by App.initServices. -func (a *App) setupPrometheusRoutes(mux *http.ServeMux) { - promQLHandler := handlers.NewPromQLHandler(a.couchbaseService) - - mux.HandleFunc("/query", promQLHandler.HandleQuery) - mux.HandleFunc("/query_range", promQLHandler.HandleQueryRange) - mux.HandleFunc("/series", promQLHandler.HandleSeries) - - log.Printf("PromQL Query API routes registered: /query, /query_range, /series") -} - // handleHealthCheckConnection probes the Couchbase buckets each enabled // feature relies on (Snapshots metadata bucket + CouchbaseDatasource // metrics bucket). Always responds HTTP 200; per-bucket state lives in diff --git a/cbmonitor/pkg/plugin/resources_test.go b/cbmonitor/pkg/plugin/resources_test.go index e2a448e..528ca33 100644 --- a/cbmonitor/pkg/plugin/resources_test.go +++ b/cbmonitor/pkg/plugin/resources_test.go @@ -127,15 +127,19 @@ func TestCallResource_DatasourceConfigReflectsSettings(t *testing.T) { } } -func TestCallResource_DatasourceRoutesGatedByToggle(t *testing.T) { - // Couchbase datasource OFF → /query, /query_range, /series should 404. - app := newAppWithSettings(t, defaultSettings()) +func TestCallResource_QueryRoutesNotServedByPlugin(t *testing.T) { + // The in-plugin PromQL-over-Couchbase query API was removed; those + // queries are served by the standalone datasource-gateway. The routes + // must 404 even when the Couchbase datasource is enabled. + settings := defaultSettings() + settings.CouchbaseDatasource = CouchbaseDatasourceSettings{Enabled: true, Bucket: "cbmonitor"} + app := newAppWithSettings(t, settings) for _, path := range []string{"query", "query_range", "series"} { - t.Run("off/"+path, func(t *testing.T) { + t.Run(path, func(t *testing.T) { resp := call(t, app, http.MethodGet, path, nil) if resp.Status != http.StatusNotFound { - t.Errorf("%s with Couchbase DS off: status = %d, want 404", path, resp.Status) + t.Errorf("%s: status = %d, want 404 (query API moved to gateway)", path, resp.Status) } }) } diff --git a/cbmonitor/pkg/plugin/settings.go b/cbmonitor/pkg/plugin/settings.go index da10aa9..ef321f1 100644 --- a/cbmonitor/pkg/plugin/settings.go +++ b/cbmonitor/pkg/plugin/settings.go @@ -16,6 +16,7 @@ type PluginSettings struct { Snapshots SnapshotsSettings `json:"snapshots"` CouchbaseDatasource CouchbaseDatasourceSettings `json:"couchbaseDatasource"` PrometheusDatasource PrometheusDatasourceSettings `json:"prometheusDatasource"` + Gateway GatewaySettings `json:"gateway"` } type CouchbaseServerSettings struct { @@ -47,6 +48,19 @@ type PrometheusDatasourceSettings struct { URL string `json:"url"` } +// GatewaySettings configures the datasource-gateway sidecar. When Enabled, the +// reconciler points the single Prometheus datasource at URL (instead of at the +// upstream Prometheus/Mimir) so the gateway can translate Couchbase-backed +// snapshots and serve overlap. When disabled, the plugin runs as plain +// Prometheus — the datasource targets PrometheusDatasource.URL directly. +type GatewaySettings struct { + Enabled bool `json:"enabled"` + URL string `json:"url"` + // Overlap reports whether snapshot overlap/time-padding is available via the + // gateway; surfaced to the UI for feature-gating the comparison view. + Overlap bool `json:"overlap"` +} + // secureFieldCouchbasePassword is the secureJsonData key that holds the // Couchbase server password. const secureFieldCouchbasePassword = "couchbasePassword" @@ -96,6 +110,9 @@ func defaultSettings() *PluginSettings { Enabled: true, IsDefault: true, }, + Gateway: GatewaySettings{ + Enabled: false, + }, } } @@ -121,6 +138,15 @@ func (s *PluginSettings) validate() error { return fmt.Errorf("prometheusDatasource.url must be an absolute URL (e.g. http://prometheus:9090)") } } + if s.Gateway.Enabled { + if s.Gateway.URL == "" { + return fmt.Errorf("gateway.url is required when the gateway is enabled") + } + u, err := url.Parse(s.Gateway.URL) + if err != nil || u.Scheme == "" || u.Host == "" { + return fmt.Errorf("gateway.url must be an absolute URL (e.g. http://datasource-gateway:8090)") + } + } return nil } diff --git a/cbmonitor/pkg/plugin/settings_test.go b/cbmonitor/pkg/plugin/settings_test.go index da4b3b3..7d4ec9a 100644 --- a/cbmonitor/pkg/plugin/settings_test.go +++ b/cbmonitor/pkg/plugin/settings_test.go @@ -154,3 +154,47 @@ func TestDefaultDataSource(t *testing.T) { }) } } + +func TestLoadSettings_GatewayParsed(t *testing.T) { + jsonData := []byte(`{ + "prometheusDatasource": {"enabled": true, "url": "http://prometheus:9090"}, + "gateway": {"enabled": true, "url": "http://datasource-gateway:8090", "overlap": true} + }`) + s, err := LoadSettings(backend.AppInstanceSettings{JSONData: jsonData}) + if err != nil { + t.Fatalf("LoadSettings: %v", err) + } + if !s.Gateway.Enabled { + t.Error("gateway.enabled not parsed") + } + if s.Gateway.URL != "http://datasource-gateway:8090" { + t.Errorf("gateway.url = %q", s.Gateway.URL) + } + if !s.Gateway.Overlap { + t.Error("gateway.overlap not parsed") + } +} + +func TestLoadSettings_GatewayDisabledByDefault(t *testing.T) { + s, err := LoadSettings(backend.AppInstanceSettings{}) + if err != nil { + t.Fatalf("LoadSettings: %v", err) + } + if s.Gateway.Enabled { + t.Error("gateway should be disabled by default (pure-Prometheus)") + } +} + +func TestLoadSettings_GatewayRequiresURLWhenEnabled(t *testing.T) { + jsonData := []byte(`{"gateway": {"enabled": true}}`) + if _, err := LoadSettings(backend.AppInstanceSettings{JSONData: jsonData}); err == nil { + t.Fatal("expected error when gateway enabled without url") + } +} + +func TestLoadSettings_GatewayRejectsMalformedURL(t *testing.T) { + jsonData := []byte(`{"gateway": {"enabled": true, "url": "not-a-url"}}`) + if _, err := LoadSettings(backend.AppInstanceSettings{JSONData: jsonData}); err == nil { + t.Fatal("expected error for malformed gateway url") + } +} diff --git a/cbmonitor/pkg/promql/parser.go b/cbmonitor/pkg/promql/parser.go deleted file mode 100644 index 15c769b..0000000 --- a/cbmonitor/pkg/promql/parser.go +++ /dev/null @@ -1,113 +0,0 @@ -package promql - -import ( - "context" - "fmt" - "strconv" - "time" - - "github.com/prometheus/prometheus/promql/parser" -) - -// ParseQuery parses a PromQL query string into an AST -func ParseQuery(query string) (parser.Expr, error) { - expr, err := parser.ParseExpr(query) - if err != nil { - return nil, fmt.Errorf("failed to parse PromQL query: %w", err) - } - return expr, nil -} - -// QueryContext holds context for query execution -type QueryContext struct { - Context context.Context - Query string - Time time.Time - StartTime time.Time - EndTime time.Time - Step time.Duration - IsRange bool - SnapshotID string // Optional snapshot ID from context -} - -// ParseQueryContext parses query parameters into a QueryContext -func ParseQueryContext(query string, timeStr string, startStr, endStr string, stepStr string, snapshotID string) (*QueryContext, error) { - ctx := &QueryContext{ - Query: query, - SnapshotID: snapshotID, - } - - // Parse time (for instant queries) - if timeStr != "" { - t, err := parseTime(timeStr) - if err != nil { - return nil, fmt.Errorf("invalid time parameter: %w", err) - } - ctx.Time = t - ctx.IsRange = false - } - - // Parse time range (for range queries) - if startStr != "" && endStr != "" { - start, err := parseTime(startStr) - if err != nil { - return nil, fmt.Errorf("invalid start time: %w", err) - } - end, err := parseTime(endStr) - if err != nil { - return nil, fmt.Errorf("invalid end time: %w", err) - } - ctx.StartTime = start - ctx.EndTime = end - ctx.IsRange = true - ctx.Time = end // Use end time as the evaluation time - } - - // Parse step (for range queries) - if stepStr != "" { - step, err := parseDuration(stepStr) - if err != nil { - return nil, fmt.Errorf("invalid step parameter: %w", err) - } - ctx.Step = step - } - - // Default to current time if no time specified - if ctx.Time.IsZero() { - ctx.Time = time.Now() - } - - return ctx, nil -} - -// parseTime parses a time string (Unix timestamp or RFC3339) -func parseTime(timeStr string) (time.Time, error) { - // Try RFC3339 first (most common format from Prometheus API) - // This includes formats like: 2025-11-26T14:54:31.661792773Z - if t, err := time.Parse(time.RFC3339, timeStr); err == nil { - return t, nil - } - - // Try RFC3339Nano for nanosecond precision - if t, err := time.Parse(time.RFC3339Nano, timeStr); err == nil { - return t, nil - } - - // Try Unix timestamp (only if the entire string is numeric) - // Use strconv.ParseFloat which requires the entire string to be a number - if timestamp, err := strconv.ParseFloat(timeStr, 64); err == nil { - // Check if it's in seconds or milliseconds - if timestamp > 1e10 { - // Likely milliseconds - return time.Unix(0, int64(timestamp*1e6)), nil - } - return time.Unix(int64(timestamp), 0), nil - } - - return time.Time{}, fmt.Errorf("unable to parse time: %s (expected RFC3339 format or Unix timestamp)", timeStr) -} - -// parseDuration parses a duration string (e.g., "5m", "1h", "30s") -func parseDuration(durationStr string) (time.Duration, error) { - return time.ParseDuration(durationStr) -} diff --git a/cbmonitor/pkg/promql/planner.go b/cbmonitor/pkg/promql/planner.go deleted file mode 100644 index 7233ddd..0000000 --- a/cbmonitor/pkg/promql/planner.go +++ /dev/null @@ -1,210 +0,0 @@ -package promql - -import ( - "fmt" - "strings" - - "github.com/prometheus/prometheus/model/labels" - "github.com/prometheus/prometheus/promql/parser" -) - -// QueryPlan represents an optimized query execution plan -type QueryPlan struct { - BaseExpr parser.Expr - SeriesQueries []SeriesQuery - Aggregation *AggregationPlan - Function *FunctionPlan -} - -// SeriesQuery represents a single series query to Couchbase -type SeriesQuery struct { - MetricName string - Snapshot string // Extracted from 'job' label in PromQL - Labels map[string]string // Additional labels to filter (including node, instance, etc.) -} - -// AggregationPlan represents aggregation operations -type AggregationPlan struct { - Operation string // sum, avg, max, min, count - By []string - Without []string -} - -// FunctionPlan represents function operations -type FunctionPlan struct { - Name string // rate, increase, irate, etc. - Range string // For range functions like rate[5m] - Parameter string // Optional parameter -} - -// PlanQuery creates an optimized query plan from a PromQL expression -func PlanQuery(expr parser.Expr, snapshotID string) (*QueryPlan, error) { - plan := &QueryPlan{ - BaseExpr: expr, - } - - // Walk the AST to extract series selectors and operations - err := parser.Walk(&queryPlanner{ - plan: plan, - snapshotID: snapshotID, - }, expr, nil) - - if err != nil { - return nil, fmt.Errorf("failed to plan query: %w", err) - } - - return plan, nil -} - -// queryPlanner implements parser.Visitor to walk the AST -type queryPlanner struct { - plan *QueryPlan - snapshotID string -} - -func (v *queryPlanner) Visit(node parser.Node, path []parser.Node) (parser.Visitor, error) { - switch n := node.(type) { - case *parser.VectorSelector: - // Extract series selector - seriesQuery, err := v.extractSeriesSelector(n) - if err != nil { - return nil, err - } - v.plan.SeriesQueries = append(v.plan.SeriesQueries, seriesQuery) - - case *parser.AggregateExpr: - // Extract aggregation - v.plan.Aggregation = &AggregationPlan{ - Operation: n.Op.String(), - By: v.extractLabels(n.Grouping), - Without: v.extractLabels(n.Grouping), // Simplified - would need to check Without flag - } - - case *parser.Call: - // Extract function call - v.plan.Function = &FunctionPlan{ - Name: n.Func.Name, - } - // Extract range parameter if it's a range function - if len(n.Args) > 0 { - if matrixSelector, ok := n.Args[0].(*parser.MatrixSelector); ok { - v.plan.Function.Range = matrixSelector.Range.String() - } - } - } - - return v, nil -} - -// extractSeriesSelector extracts series information from a vector selector -func (v *queryPlanner) extractSeriesSelector(selector *parser.VectorSelector) (SeriesQuery, error) { - query := SeriesQuery{ - MetricName: selector.Name, - Snapshot: v.snapshotID, // Will be overridden by 'job' label if present - Labels: make(map[string]string), - } - - // Extract labels from matchers - for _, matcher := range selector.LabelMatchers { - labelName := matcher.Name - labelValue := matcher.Value - - switch matcher.Type { - case labels.MatchEqual: - // Map known labels to query parameters - switch labelName { - case "job", "snapshot", "snapshot_id": - // 'job' label maps to snapshot in get_metric_for function - query.Snapshot = labelValue - default: - // Store as additional label filter (node, instance, bucket, etc.) - query.Labels[labelName] = labelValue - } - - case labels.MatchNotEqual: - // For !=, we'll need to filter in WHERE clause - query.Labels["!"+labelName] = labelValue - - case labels.MatchRegexp: - // For =~, we'll need regex matching in WHERE clause - query.Labels["=~"+labelName] = labelValue - - case labels.MatchNotRegexp: - // For !~, we'll need negative regex matching - query.Labels["!~"+labelName] = labelValue - } - } - - // Default snapshot if not provided (from job label or context) - if query.Snapshot == "" { - query.Snapshot = v.snapshotID - } - - return query, nil -} - -// extractLabels extracts label names from grouping -func (v *queryPlanner) extractLabels(grouping []string) []string { - return grouping -} - -// GetMetricName extracts the metric name from the expression -func GetMetricName(expr parser.Expr) string { - switch e := expr.(type) { - case *parser.VectorSelector: - return e.Name - case *parser.MatrixSelector: - return GetMetricName(e.VectorSelector) - case *parser.Call: - if len(e.Args) > 0 { - return GetMetricName(e.Args[0]) - } - } - return "" -} - -// HasMultipleSeries checks if the query involves multiple series -func (p *QueryPlan) HasMultipleSeries() bool { - return len(p.SeriesQueries) > 1 -} - -// ShouldBatch determines if queries should be batched -func (p *QueryPlan) ShouldBatch() bool { - // Batch if we have multiple series with the same metric name - if len(p.SeriesQueries) <= 1 { - return false - } - - // Check if all series have the same metric name - firstMetric := p.SeriesQueries[0].MetricName - for _, sq := range p.SeriesQueries[1:] { - if sq.MetricName != firstMetric { - return false - } - } - - // Batch if we have 5-20 series (adaptive threshold) - return len(p.SeriesQueries) >= 5 && len(p.SeriesQueries) <= 20 -} - -// GetBatchedQueries groups queries by metric name for batching -func (p *QueryPlan) GetBatchedQueries() map[string][]SeriesQuery { - batched := make(map[string][]SeriesQuery) - for _, sq := range p.SeriesQueries { - batched[sq.MetricName] = append(batched[sq.MetricName], sq) - } - return batched -} - -// String returns a human-readable representation of the plan -func (p *QueryPlan) String() string { - var parts []string - parts = append(parts, fmt.Sprintf("SeriesQueries: %d", len(p.SeriesQueries))) - if p.Aggregation != nil { - parts = append(parts, fmt.Sprintf("Aggregation: %s", p.Aggregation.Operation)) - } - if p.Function != nil { - parts = append(parts, fmt.Sprintf("Function: %s", p.Function.Name)) - } - return strings.Join(parts, ", ") -} diff --git a/cbmonitor/pkg/promql/sqlbuilder.go b/cbmonitor/pkg/promql/sqlbuilder.go deleted file mode 100644 index 995b05b..0000000 --- a/cbmonitor/pkg/promql/sqlbuilder.go +++ /dev/null @@ -1,319 +0,0 @@ -package promql - -import ( - "fmt" - "strings" - "time" - - "github.com/couchbase/cbmonitor/pkg/querybuilder" -) - -// SQLBuilder builds SQL++ queries from query plans -type SQLBuilder struct { - plan *QueryPlan - queryCtx *QueryContext - useBatching bool -} - -// NewSQLBuilder creates a new SQL++ query builder -func NewSQLBuilder(plan *QueryPlan, queryCtx *QueryContext) *SQLBuilder { - return &SQLBuilder{ - plan: plan, - queryCtx: queryCtx, - useBatching: plan.ShouldBatch(), - } -} - -// Build generates SQL++ query string(s) from the plan -func (b *SQLBuilder) Build() ([]string, error) { - if b.useBatching { - return b.buildBatchedQueries() - } - return b.buildIndividualQueries() -} - -// buildIndividualQueries builds separate queries for each series -func (b *SQLBuilder) buildIndividualQueries() ([]string, error) { - var queries []string - - for _, seriesQuery := range b.plan.SeriesQueries { - query, err := b.buildSeriesQuery(seriesQuery) - if err != nil { - return nil, err - } - queries = append(queries, query) - } - - return queries, nil -} - -// buildBatchedQueries builds batched queries using UNION ALL -func (b *SQLBuilder) buildBatchedQueries() ([]string, error) { - batched := b.plan.GetBatchedQueries() - var queries []string - - for _, seriesQueries := range batched { - if len(seriesQueries) == 1 { - // Single series, no need to batch - query, err := b.buildSeriesQuery(seriesQueries[0]) - if err != nil { - return nil, err - } - queries = append(queries, query) - continue - } - - // Build UNION ALL query - var unionParts []string - for _, sq := range seriesQueries { - part := b.buildSeriesQueryPart(sq) - unionParts = append(unionParts, part) - } - - // Combine with UNION ALL and add aggregation if needed - unionQuery := strings.Join(unionParts, " UNION ALL ") - - // If aggregation is needed, wrap in aggregation query - if b.plan.Aggregation != nil { - unionQuery = b.wrapWithAggregation(unionQuery) - } - - queries = append(queries, unionQuery) - } - - return queries, nil -} - -// buildSeriesQuery builds a SQL++ query for a single series -func (b *SQLBuilder) buildSeriesQuery(seriesQuery SeriesQuery) (string, error) { - baseQuery := b.buildSeriesQueryPart(seriesQuery) - - // Apply function transformation if needed - if b.plan.Function != nil { - baseQuery = b.applyFunction(baseQuery) - } - - // Apply aggregation if needed - if b.plan.Aggregation != nil { - baseQuery = b.wrapWithAggregation(baseQuery) - } - - return baseQuery, nil -} - -// buildSeriesQueryPart builds the base SQL++ query part for a series -func (b *SQLBuilder) buildSeriesQueryPart(seriesQuery SeriesQuery) string { - // Build metrics_filter function call - metricFilter := b.buildMetricFilter(seriesQuery) - - // Build time range - fromMillis := b.queryCtx.StartTime.UnixMilli() - toMillis := b.queryCtx.EndTime.UnixMilli() - if !b.queryCtx.IsRange { - // For instant queries, use a small range around the time (~30s) - fromMillis = b.queryCtx.Time.UnixMilli() - 30000 - toMillis = b.queryCtx.Time.UnixMilli() + 1000 - } - - // Build SELECT clause - selectClause := b.buildSelectClause(seriesQuery) - - // Build time range condition - // Note: label filters are already embedded in metricFilter (before UNNEST) - // Time filtering happens after UNNEST but before final output - timeCondition := fmt.Sprintf("d.time_millis >= %d AND d.time_millis <= %d", fromMillis, toMillis) - - query := fmt.Sprintf( - "SELECT %s FROM %s AS d WHERE %s", - selectClause, - metricFilter, - timeCondition, - ) - - return query -} - -// buildMetricFilter builds the query with all label conditions embedded BEFORE UNNEST -// This ensures optimal performance by filtering at the earliest possible point -func (b *SQLBuilder) buildMetricFilter(seriesQuery SeriesQuery) string { - // Build all label conditions including job/snapshot - labelConditions := b.buildLabelWhereClause(seriesQuery) - - // Construct query inline with conditions embedded - filters BEFORE UNNEST - baseQuery := fmt.Sprintf( - "SELECT t._t AS time_millis, MILLIS_TO_STR(t._t) AS time, t._v0 AS `value`, d.labels AS labels FROM cbmonitor._default._default AS d UNNEST _timeseries(d, {'ts_ranges':[0, 9223372036854775807]}) AS t WHERE d.metric_name = '%s'", - seriesQuery.MetricName, - ) - - if labelConditions != "" { - baseQuery = fmt.Sprintf("%s AND %s", baseQuery, labelConditions) - } - - return fmt.Sprintf("(%s)", baseQuery) -} - -// buildSelectClause builds the SELECT clause -func (b *SQLBuilder) buildSelectClause(seriesQuery SeriesQuery) string { - parts := []string{ - "d.time", - "d.`value`", - } - - // Add label fields if needed - if len(seriesQuery.Labels) > 0 || b.plan.Aggregation != nil { - // Include labels for grouping/aggregation - parts = append(parts, "d.labels") - } - - return strings.Join(parts, ", ") -} - -// buildLabelWhereClause builds the WHERE clause for label filters including job/snapshot -func (b *SQLBuilder) buildLabelWhereClause(seriesQuery SeriesQuery) string { - // Convert seriesQuery.Labels to LabelFilter slice for shared utility - var filters []querybuilder.LabelFilter - - // Check if job label is already in Labels (could be with operators like !=, =~, !~) - hasJobInLabels := false - for labelName := range seriesQuery.Labels { - // Remove operator prefixes to get actual label name - actualLabel := labelName - if strings.HasPrefix(labelName, "!") { - actualLabel = strings.TrimPrefix(labelName, "!") - } else if strings.HasPrefix(labelName, "=~") { - actualLabel = strings.TrimPrefix(labelName, "=~") - } else if strings.HasPrefix(labelName, "!~") { - actualLabel = strings.TrimPrefix(labelName, "!~") - } - if actualLabel == "job" || actualLabel == "snapshot" || actualLabel == "snapshot_id" { - hasJobInLabels = true - break - } - } - - // Add job/snapshot filter if not already present in Labels - if !hasJobInLabels { - snapshot := seriesQuery.Snapshot - if snapshot == "" { - snapshot = b.queryCtx.SnapshotID - } - if snapshot != "" { - // Add job label filter (snapshot is stored as job label) - filters = append(filters, querybuilder.LabelFilter{ - Name: "job", - Value: snapshot, - Op: "=", - }) - } - } - - // Process all label filters from Labels - for labelName, labelValue := range seriesQuery.Labels { - if strings.HasPrefix(labelName, "!") { - // NOT EQUAL - actualLabel := strings.TrimPrefix(labelName, "!") - filters = append(filters, querybuilder.LabelFilter{ - Name: actualLabel, - Value: labelValue, - Op: "!=", - }) - } else if strings.HasPrefix(labelName, "=~") { - // REGEX MATCH - actualLabel := strings.TrimPrefix(labelName, "=~") - filters = append(filters, querybuilder.LabelFilter{ - Name: actualLabel, - Value: labelValue, - Op: "=~", - }) - } else if strings.HasPrefix(labelName, "!~") { - // NOT REGEX MATCH - actualLabel := strings.TrimPrefix(labelName, "!~") - filters = append(filters, querybuilder.LabelFilter{ - Name: actualLabel, - Value: labelValue, - Op: "!~", - }) - } else { - // EQUAL - process all labels (job will be handled above if not in Labels) - filters = append(filters, querybuilder.LabelFilter{ - Name: labelName, - Value: labelValue, - Op: "=", - }) - } - } - - return querybuilder.BuildLabelWhereClauseFromFilters(filters) -} - -// applyFunction applies PromQL function transformations -func (b *SQLBuilder) applyFunction(query string) string { - if b.plan.Function == nil { - return query - } - - switch b.plan.Function.Name { - case "rate", "irate", "increase": - // These need to be calculated post-query for now - // We'll mark the query for post-processing - return query - default: - return query - } -} - -// wrapWithAggregation wraps query with aggregation -func (b *SQLBuilder) wrapWithAggregation(query string) string { - if b.plan.Aggregation == nil { - return query - } - - // Build GROUP BY clause - var groupBy []string - if len(b.plan.Aggregation.By) > 0 { - for _, label := range b.plan.Aggregation.By { - groupBy = append(groupBy, fmt.Sprintf("d.labels.%s", querybuilder.EscapeLabel(label))) - } - } - groupBy = append(groupBy, "d.time") // Always group by time - - // Build aggregation function - aggFunc := strings.ToUpper(b.plan.Aggregation.Operation) - if aggFunc == "" { - aggFunc = "SUM" - } - - // Wrap query - wrapped := fmt.Sprintf( - "SELECT time, %s(value) AS value %s FROM (%s) AS subq GROUP BY %s ORDER BY time", - aggFunc, - b.buildLabelSelect(groupBy), - query, - strings.Join(groupBy, ", "), - ) - - return wrapped -} - -// buildLabelSelect builds label selection for GROUP BY -func (b *SQLBuilder) buildLabelSelect(groupBy []string) string { - var labels []string - for _, gb := range groupBy { - if strings.HasPrefix(gb, "d.labels.") { - labels = append(labels, gb) - } - } - if len(labels) > 0 { - return ", " + strings.Join(labels, ", ") - } - return "" -} - -// GetTimeRange returns the time range for the query -func (b *SQLBuilder) GetTimeRange() (time.Time, time.Time) { - if b.queryCtx.IsRange { - return b.queryCtx.StartTime, b.queryCtx.EndTime - } - // For instant queries, return a small range - return b.queryCtx.Time.Add(-time.Second), b.queryCtx.Time.Add(time.Second) -} diff --git a/cbmonitor/pkg/promql/transformer.go b/cbmonitor/pkg/promql/transformer.go deleted file mode 100644 index 3f10c53..0000000 --- a/cbmonitor/pkg/promql/transformer.go +++ /dev/null @@ -1,510 +0,0 @@ -package promql - -import ( - "fmt" - "math" - "sort" - "strconv" - "strings" - "time" -) - -// PrometheusResult represents a Prometheus API result -type PrometheusResult struct { - Status string `json:"status"` - Data ResultData `json:"data"` - Error string `json:"error,omitempty"` - ErrorType string `json:"errorType,omitempty"` - // Warnings carries non-fatal issues encountered while serving the query - // (e.g. some sub-queries failed but enough succeeded to return data). - // Matches the Prometheus HTTP API spec. - Warnings []string `json:"warnings,omitempty"` -} - -// ResultData represents the data portion of Prometheus response -type ResultData struct { - ResultType string `json:"resultType"` - Result []interface{} `json:"result"` -} - -// Sample represents a single time-value pair -type Sample struct { - Timestamp float64 `json:"timestamp"` // Unix timestamp in seconds - Value string `json:"value"` // String representation of float -} - -// Series represents a time series with labels and samples -type Series struct { - Metric map[string]string `json:"metric"` - Values []Sample `json:"values,omitempty"` // For range queries - Value Sample `json:"value,omitempty"` // For instant queries -} - -// QueryResult represents raw query results from Couchbase -type QueryResult struct { - Time string `json:"time"` - Value interface{} `json:"value"` - Labels map[string]interface{} `json:"labels,omitempty"` -} - -// TransformResults transforms Couchbase query results to Prometheus format -func TransformResults(results []QueryResult, plan *QueryPlan, queryCtx *QueryContext) (*PrometheusResult, error) { - if len(results) == 0 { - return &PrometheusResult{ - Status: "success", - Data: ResultData{ - ResultType: getResultType(plan, queryCtx), - Result: []interface{}{}, - }, - }, nil - } - - // Group results by labels to form series - seriesMap := make(map[string]*Series) - - for _, result := range results { - // Extract labels - labels := extractLabels(result, plan) - labelKey := buildLabelKey(labels) - - // Get or create series - series, exists := seriesMap[labelKey] - if !exists { - series = &Series{ - Metric: labels, - } - seriesMap[labelKey] = series - } - - // Parse time and value - timestamp, err := parseTimeFromResult(result.Time) - if err != nil { - return nil, fmt.Errorf("failed to parse time: %w", err) - } - - value, err := parseValue(result.Value) - if err != nil { - return nil, fmt.Errorf("failed to parse value: %w", err) - } - - // Create sample - sample := Sample{ - Timestamp: float64(timestamp.Unix()), - Value: formatValue(value), - } - - // Add to series - if queryCtx.IsRange { - series.Values = append(series.Values, sample) - } else { - series.Value = sample - } - } - - // Convert map to slice and sort - var seriesList []interface{} - for _, series := range seriesMap { - // Sort values by timestamp - if queryCtx.IsRange { - sort.Slice(series.Values, func(i, j int) bool { - return series.Values[i].Timestamp < series.Values[j].Timestamp - }) - } - seriesList = append(seriesList, series) - } - - // Apply function transformations - if plan.Function != nil { - seriesList, err := applyFunction(seriesList, plan.Function, queryCtx) - if err != nil { - return nil, fmt.Errorf("failed to apply function: %w", err) - } - // Update seriesList with transformed results - _ = seriesList // Will be used after implementing function application - } - - // Apply aggregation if needed - if plan.Aggregation != nil { - seriesList = applyAggregation(seriesList, plan.Aggregation) - } - - return &PrometheusResult{ - Status: "success", - Data: ResultData{ - ResultType: getResultType(plan, queryCtx), - Result: seriesList, - }, - }, nil -} - -// extractLabels extracts labels from query result -func extractLabels(result QueryResult, plan *QueryPlan) map[string]string { - labels := make(map[string]string) - - // Add metric name from plan - if len(plan.SeriesQueries) > 0 { - metricName := plan.SeriesQueries[0].MetricName - if metricName != "" { - labels["__name__"] = metricName - } - } - - // Extract labels from result - if result.Labels != nil { - for key, value := range result.Labels { - if strValue, ok := value.(string); ok { - labels[key] = strValue - } else if strValue := fmt.Sprintf("%v", value); strValue != "" { - labels[key] = strValue - } - } - } - - return labels -} - -// buildLabelKey creates a unique key for a label set -func buildLabelKey(labels map[string]string) string { - // Sort keys for consistent ordering - var keys []string - for k := range labels { - keys = append(keys, k) - } - sort.Strings(keys) - - var parts []string - for _, k := range keys { - parts = append(parts, fmt.Sprintf("%s=%s", k, labels[k])) - } - return fmt.Sprintf("{%s}", strings.Join(parts, ",")) -} - -// parseTimeFromResult parses time from result string -func parseTimeFromResult(timeStr string) (time.Time, error) { - // Try RFC3339 first - if t, err := time.Parse(time.RFC3339, timeStr); err == nil { - return t, nil - } - - // Try Unix timestamp - if timestamp, err := strconv.ParseFloat(timeStr, 64); err == nil { - if timestamp > 1e10 { - // Milliseconds - return time.Unix(0, int64(timestamp*1e6)), nil - } - return time.Unix(int64(timestamp), 0), nil - } - - return time.Time{}, fmt.Errorf("unable to parse time: %s", timeStr) -} - -// parseValue parses value from result -func parseValue(value interface{}) (float64, error) { - switch v := value.(type) { - case float64: - return v, nil - case int: - return float64(v), nil - case int64: - return float64(v), nil - case string: - f, err := strconv.ParseFloat(v, 64) - if err != nil { - return 0, fmt.Errorf("invalid value: %s", v) - } - return f, nil - default: - return 0, fmt.Errorf("unsupported value type: %T", value) - } -} - -// formatValue formats a float value as string (Prometheus format) -func formatValue(value float64) string { - // Handle special values - if math.IsNaN(value) { - return "NaN" - } - if math.IsInf(value, 1) { - return "+Inf" - } - if math.IsInf(value, -1) { - return "-Inf" - } - - // Format with appropriate precision - return strconv.FormatFloat(value, 'f', -1, 64) -} - -// getResultType determines the result type based on query -func getResultType(plan *QueryPlan, queryCtx *QueryContext) string { - if queryCtx.IsRange { - return "matrix" - } - return "vector" -} - -// applyFunction applies PromQL function transformations -func applyFunction(seriesList []interface{}, function *FunctionPlan, queryCtx *QueryContext) ([]interface{}, error) { - switch function.Name { - case "rate": - return applyRate(seriesList, function.Range) - case "irate": - return applyIRate(seriesList) - case "increase": - return applyIncrease(seriesList, function.Range) - default: - // Unknown function, return as-is - return seriesList, nil - } -} - -// applyRate calculates rate (per-second average rate) -func applyRate(seriesList []interface{}, rangeStr string) ([]interface{}, error) { - // Parse range duration (for future use in range-based rate calculation) - _, err := time.ParseDuration(rangeStr) - if err != nil && rangeStr != "" { - return nil, fmt.Errorf("invalid range: %w", err) - } - - // Apply rate to each series - var result []interface{} - for _, item := range seriesList { - series, ok := item.(*Series) - if !ok { - continue - } - - if len(series.Values) < 2 { - // Need at least 2 points for rate - continue - } - - // Calculate rate for each interval - newValues := make([]Sample, 0, len(series.Values)-1) - for i := 1; i < len(series.Values); i++ { - prev := series.Values[i-1] - curr := series.Values[i] - - prevVal, _ := strconv.ParseFloat(prev.Value, 64) - currVal, _ := strconv.ParseFloat(curr.Value, 64) - - timeDiff := curr.Timestamp - prev.Timestamp - if timeDiff <= 0 { - continue - } - - // Rate = (value change) / (time change in seconds) - rate := (currVal - prevVal) / timeDiff - - newValues = append(newValues, Sample{ - Timestamp: curr.Timestamp, - Value: formatValue(rate), - }) - } - - // Update series - newSeries := *series - newSeries.Values = newValues - result = append(result, &newSeries) - } - - return result, nil -} - -// applyIRate calculates instant rate (per-second rate from last two points) -func applyIRate(seriesList []interface{}) ([]interface{}, error) { - // Similar to rate but only uses last two points - return applyRate(seriesList, "") -} - -// applyIncrease calculates increase over time range -func applyIncrease(seriesList []interface{}, rangeStr string) ([]interface{}, error) { - // Similar to rate but returns absolute increase - // For now, use rate implementation - return applyRate(seriesList, rangeStr) -} - -// applyAggregation applies aggregation operations -func applyAggregation(seriesList []interface{}, agg *AggregationPlan) []interface{} { - // Group series by labels (excluding aggregation labels) - grouped := make(map[string][]*Series) - - for _, item := range seriesList { - series, ok := item.(*Series) - if !ok { - continue - } - - // Build group key - groupKey := buildAggregationKey(series.Metric, agg) - grouped[groupKey] = append(grouped[groupKey], series) - } - - // Apply aggregation to each group - var result []interface{} - for _, group := range grouped { - aggSeries := aggregateSeries(group, agg) - if aggSeries != nil { - result = append(result, aggSeries) - } - } - - return result -} - -// buildAggregationKey builds key for grouping -func buildAggregationKey(labels map[string]string, agg *AggregationPlan) string { - var keys []string - for k, v := range labels { - // Skip __name__ and aggregation-excluded labels - if k == "__name__" { - continue - } - // Include only labels in "by" or exclude those in "without" - keys = append(keys, fmt.Sprintf("%s=%s", k, v)) - } - sort.Strings(keys) - return strings.Join(keys, ",") -} - -// aggregateSeries aggregates multiple series into one -func aggregateSeries(seriesList []*Series, agg *AggregationPlan) *Series { - if len(seriesList) == 0 { - return nil - } - - // Use first series as base - result := &Series{ - Metric: make(map[string]string), - } - - // Copy labels from first series (excluding aggregation-excluded labels) - for k, v := range seriesList[0].Metric { - if k != "__name__" { - result.Metric[k] = v - } - } - - // Aggregate values - if len(seriesList[0].Values) > 0 { - // Range query - aggregate across series for each timestamp - result.Values = aggregateValues(seriesList, agg) - } else { - // Instant query - aggregate single values - result.Value = aggregateValue(seriesList, agg) - } - - return result -} - -// aggregateValues aggregates values across series for range queries -func aggregateValues(seriesList []*Series, agg *AggregationPlan) []Sample { - // Find all unique timestamps - timestampMap := make(map[float64]bool) - for _, series := range seriesList { - for _, sample := range series.Values { - timestampMap[sample.Timestamp] = true - } - } - - // Sort timestamps - var timestamps []float64 - for ts := range timestampMap { - timestamps = append(timestamps, ts) - } - sort.Float64s(timestamps) - - // Aggregate for each timestamp - var result []Sample - for _, ts := range timestamps { - var values []float64 - for _, series := range seriesList { - for _, sample := range series.Values { - if sample.Timestamp == ts { - if val, err := strconv.ParseFloat(sample.Value, 64); err == nil { - values = append(values, val) - } - break - } - } - } - - if len(values) > 0 { - aggValue := aggregateFloat(values, agg.Operation) - result = append(result, Sample{ - Timestamp: ts, - Value: formatValue(aggValue), - }) - } - } - - return result -} - -// aggregateValue aggregates single values for instant queries -func aggregateValue(seriesList []*Series, agg *AggregationPlan) Sample { - var values []float64 - var timestamp float64 - - for _, series := range seriesList { - if val, err := strconv.ParseFloat(series.Value.Value, 64); err == nil { - values = append(values, val) - if timestamp == 0 { - timestamp = series.Value.Timestamp - } - } - } - - aggValue := aggregateFloat(values, agg.Operation) - return Sample{ - Timestamp: timestamp, - Value: formatValue(aggValue), - } -} - -// aggregateFloat aggregates float values based on operation -func aggregateFloat(values []float64, operation string) float64 { - if len(values) == 0 { - return 0 - } - - switch strings.ToUpper(operation) { - case "SUM": - sum := 0.0 - for _, v := range values { - sum += v - } - return sum - case "AVG", "AVERAGE": - sum := 0.0 - for _, v := range values { - sum += v - } - return sum / float64(len(values)) - case "MAX": - max := values[0] - for _, v := range values { - if v > max { - max = v - } - } - return max - case "MIN": - min := values[0] - for _, v := range values { - if v < min { - min = v - } - } - return min - case "COUNT": - return float64(len(values)) - default: - // Default to sum - sum := 0.0 - for _, v := range values { - sum += v - } - return sum - } -} diff --git a/cbmonitor/provisioning/datasources/datasource.yaml b/cbmonitor/provisioning/datasources/datasource.yaml index 778aa61..99f62c5 100644 --- a/cbmonitor/provisioning/datasources/datasource.yaml +++ b/cbmonitor/provisioning/datasources/datasource.yaml @@ -1,19 +1,8 @@ apiVersion: 1 -# The `prometheus` and `cbdatasource` datasources are managed by the -# cbmonitor app plugin itself (reconciled at runtime from app jsonData via -# the Grafana HTTP API). Don't add them back here or you'll get duplicates. -datasources: -- name: ProxyPrometheus - uid: proxyprometheus - type: prometheus - access: proxy - url: ${PROXY_PROMETHEUS_URL} - basicAuth: false - isDefault: false - jsonData: - cacheLevel: High - httpMethod: GET - prometheusType: Mimir - prometheusVersion: 2.9.1 - secureJsonData: {} +# The single `prometheus` datasource is managed by the cbmonitor app plugin +# itself (reconciled at runtime from app jsonData via the Grafana HTTP API). +# It points at the datasource-gateway sidecar when the gateway is enabled, +# otherwise at Prometheus/Mimir directly. Don't add it back here or you'll get +# duplicates. +datasources: [] diff --git a/cbmonitor/provisioning/plugins/plugins.yaml b/cbmonitor/provisioning/plugins/plugins.yaml index 4db95e5..357091b 100644 --- a/cbmonitor/provisioning/plugins/plugins.yaml +++ b/cbmonitor/provisioning/plugins/plugins.yaml @@ -1,6 +1,5 @@ apiVersion: 1 # No plugins to enable here. The cbmonitor app plugin itself is provisioned -# via plugins/apps.yaml; datasource plugins (prometheus, couchbase-datasource) -# come from Grafana's built-ins / bundled includes. +# via plugins/apps.yaml; the Prometheus datasource comes from Grafana's built-ins. plugins: [] diff --git a/cbmonitor/src/components/App/App.tsx b/cbmonitor/src/components/App/App.tsx index 7e8a9e1..c3c0c93 100644 --- a/cbmonitor/src/components/App/App.tsx +++ b/cbmonitor/src/components/App/App.tsx @@ -1,7 +1,7 @@ -import React, { useEffect } from 'react'; +import React from 'react'; import { AppRootProps } from '@grafana/data'; import { config } from '@grafana/runtime'; -import { CB_DATASOURCE_REF, PROM_DATASOURCE_REF } from '../../constants'; +import { PROM_DATASOURCE_REF } from '../../constants'; import { SceneApp, useSceneApp } from '@grafana/scenes'; import { Alert } from '@grafana/ui'; import { PluginPropsContext } from 'utils/utils.plugin'; @@ -9,7 +9,6 @@ import { AppNavHeader } from '../AppNavHeader/AppNavHeader'; import { snapshotSearchPage, snapshotViewPage } from '../../pages/snapshotViewPage'; import { preferencesPage } from '../../pages/preferencesPage'; import { comparisonPage } from '../../components/SnapshotDisplay/comparisonInstance'; -import { dataSourceService } from '../../services/datasourceService'; // Defines the app and its pages function getCBMonitorApp(){ @@ -31,22 +30,14 @@ function getCBMonitorApp(){ function CBMonitorHome() { const scene = useSceneApp(getCBMonitorApp); - // Initialize datasource configuration from backend when app mounts - useEffect(() => { - dataSourceService.initializeConfig().catch((error) => { - console.error('[App] Failed to initialize datasource config:', error); - }); - }, []); - const datasources = Object.values(config.datasources); - const haveCb = datasources.some((d) => d.uid === CB_DATASOURCE_REF.uid); const haveProm = datasources.some((d) => d.uid === PROM_DATASOURCE_REF.uid); return ( <> - {!haveCb && !haveProm && ( + {!haveProm && ( - {JSON.stringify(CB_DATASOURCE_REF)} or {JSON.stringify(PROM_DATASOURCE_REF)} datasource is required to use this app. + The {JSON.stringify(PROM_DATASOURCE_REF)} datasource is required to use this app. Available datasources: