diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index 4d39723..f7d3c1f 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -8,7 +8,7 @@ on: jobs: test: - name: Test (${{ matrix.module }}) + name: Test (${{ matrix.module }}${{ matrix.suite && format(' {0}', matrix.suite) || '' }}) runs-on: ubuntu-latest strategy: matrix: @@ -27,6 +27,9 @@ jobs: test-cmd: go test ./... - module: queryapi test-cmd: task test + - module: queryapi + suite: integration-prometheus + test-cmd: task test-integration-prometheus - module: cli test-cmd: go test -race ./... steps: diff --git a/config/operator/iam/protected-resources/metrics.yaml b/config/operator/iam/protected-resources/metrics.yaml index 2ff766e..dfbca41 100644 --- a/config/operator/iam/protected-resources/metrics.yaml +++ b/config/operator/iam/protected-resources/metrics.yaml @@ -1,6 +1,5 @@ -# The metric query surface. queryapi's Prometheus-shaped routes return 501 -# today; declaring the permissions now means the endpoints cannot ship -# ungated. See queryapi/README.md. +# The metric query surface. queryapi's Prometheus-shaped routes are gated by +# these permissions like the log routes. See queryapi/README.md. apiVersion: iam.miloapis.com/v1alpha1 kind: ProtectedResource metadata: diff --git a/config/queryapi/deployment.yaml b/config/queryapi/deployment.yaml index ca6fdb7..25e768a 100644 --- a/config/queryapi/deployment.yaml +++ b/config/queryapi/deployment.yaml @@ -35,7 +35,10 @@ spec: # the rest of this list alone. Same reason activity's deployment is # written this way; see queryapi/README.md for a worked patch. args: - - --storage=$(STORAGE) + - --logs-storage=$(STORAGE) + - --metrics-storage=$(METRICS_STORAGE) + - --prometheus-url=$(PROMETHEUS_URL) + - --prometheus-tenancy=$(PROMETHEUS_TENANCY) - --secure-port=$(SECURE_PORT) - --tls-cert-file=$(TLS_CERT_FILE) - --tls-private-key-file=$(TLS_PRIVATE_KEY_FILE) @@ -51,6 +54,16 @@ spec: # once it can. - name: STORAGE value: "fake" + # "prometheus" also needs PROMETHEUS_URL and PROMETHEUS_TENANCY. + # extra-label is the only tenancy mode, and it requires + # VictoriaMetrics: a backend that ignores extra_label applies no + # isolation. + - name: METRICS_STORAGE + value: "fake" + - name: PROMETHEUS_URL + value: "" + - name: PROMETHEUS_TENANCY + value: "" # TLS is the whole point of the identity chain, not a hardening # extra: only a TLS connection carries the front proxy's client # certificate, and only that certificate makes the forwarded diff --git a/queryapi/README.md b/queryapi/README.md index a758351..5638c8f 100644 --- a/queryapi/README.md +++ b/queryapi/README.md @@ -39,12 +39,15 @@ permissions, not collections. ## Storage backends -| `--storage` | Behaviour | +| `--logs-storage` | Behaviour | | --- | --- | | `fake` (default) | Deterministic synthetic logs, a pure function of the query's time range. Same range always returns the same rows. | | `clickhouse` | Serves log queries and label/stream discovery against the `o11y.logs` schema via LogQL-to-SQL translation. `/readyz` pings the backend. | -Metric (Prometheus-shaped) endpoints return 501 on both backends. +| `--metrics-storage` | Behaviour | +| --- | --- | +| `fake` (default) | One synthetic series, `fake_up`, for the caller's project. | +| `prometheus` | Any Prometheus query API backend, confined to the caller's project by `--prometheus-tenancy`. | ## Flags @@ -71,7 +74,14 @@ them without restating the rest. See [Deployment](#deployment). | `--authorization-always-allow-paths` | `/healthz,/readyz,/livez,/metrics` | | `--authorization-webhook-cache-authorized-ttl` | `10s` | | `--authorization-webhook-cache-unauthorized-ttl` | `10s` | -| `--storage` | `fake` | +| `--logs-storage` | `fake` | +| `--storage` | deprecated alias for `--logs-storage`; removed in the next release | +| `--metrics-storage` | `fake` | +| `--prometheus-url` | none; required with `--metrics-storage=prometheus` | +| `--prometheus-tenancy` | none; required with `--metrics-storage=prometheus`. The only mode is `extra-label`, which requires VictoriaMetrics | +| `--prometheus-project-label` | `resourcemanager_datumapis_com_project_name` | +| `--max-response-bytes` | `33554432` (32 MiB) | +| `--max-range` | `744h` | | `--fake-rate` | `2` (lines/sec) | | `--query-timeout` | `30s` | | `--default-limit` | `100` | @@ -99,7 +109,7 @@ serving stack owns the `http.Server` and sets them (32s and 90s). ## ClickHouse environment -Used only with `--storage=clickhouse`. Authentication is by client +Used only with `--logs-storage=clickhouse`. Authentication is by client certificate, not password, matching `clickhouse-migrate`. | Env var | Default | @@ -194,8 +204,8 @@ hostnames and customer identifiers. The `telemetry.miloapis.com-viewer` role ([`config/operator/iam/`](../config/operator/iam/)) grants both the queryapi route permissions and the proxy permissions (`logs/api.get`, `metrics/v1.get` and `metrics/v1.create`). -The metrics routes return 501 today and are gated anyway, so they cannot ship -unguarded. +The metrics routes are gated the same way as the log routes, so they cannot +ship unguarded. The split is only enforceable here, not at the aggregator, so a metadata-only role depends on queryapi's check holding rather than on Milo refusing the diff --git a/queryapi/Taskfile.yaml b/queryapi/Taskfile.yaml index 84765fd..adfdccf 100644 --- a/queryapi/Taskfile.yaml +++ b/queryapi/Taskfile.yaml @@ -39,6 +39,33 @@ tasks: done - CLICKHOUSE_TEST_ADDR=127.0.0.1:{{.PORT}} CLICKHOUSE_TEST_PASSWORD={{.PASSWORD}} go test -count=1 -run TestStoreAgainstClickHouse ./internal/storage/clickhouse/ + test-integration-prometheus: + desc: Run the Prometheus store's integration test against a throwaway VictoriaMetrics (needs docker) + vars: + IMAGE: victoriametrics/victoria-metrics:v1.114.0 + CONTAINER: queryapi-vm-test + PORT: 18428 + cmds: + - docker rm -f {{.CONTAINER}} >/dev/null 2>&1 || true + # latencyOffset=0: just-imported samples are queryable at once. The result + # cache stays on, as in production. + - >- + docker run -d --name {{.CONTAINER}} -p {{.PORT}}:8428 {{.IMAGE}} + -search.latencyOffset=0s -retentionPeriod=1d + - defer: docker rm -f {{.CONTAINER}} >/dev/null + - | + for i in $(seq 1 60); do + curl -fsS http://127.0.0.1:{{.PORT}}/-/healthy >/dev/null 2>&1 && break + sleep 1 + done + # A renamed or skipped test must fail the task, not pass it: -run that + # matches nothing exits 0. grep without -q reads to EOF: -q would SIGPIPE tee. + - | + set -o pipefail + PROMETHEUS_TEST_URL=http://127.0.0.1:{{.PORT}} \ + go test -count=1 -v -run '^TestStoreAgainstVictoriaMetrics$' ./internal/storage/prometheus/ \ + | tee /dev/stderr | grep -- '--- PASS: TestStoreAgainstVictoriaMetrics' >/dev/null + lint: desc: Run golangci-lint (assumes it's installed on PATH -- see CI's pinned version) cmds: diff --git a/queryapi/apiserver.go b/queryapi/apiserver.go index 9df3572..6b79634 100644 --- a/queryapi/apiserver.go +++ b/queryapi/apiserver.go @@ -26,9 +26,10 @@ const kubeVersion = "1.34" // ExtraConfig is everything the generic apiserver knows nothing about. type ExtraConfig struct { - Logger *slog.Logger - Store storage.LogStore - Query Config + Logger *slog.Logger + LogStore storage.LogStore + MetricStore storage.MetricStore + Query Config } // ServerConfig combines the generic apiserver configuration with queryapi's. @@ -75,14 +76,17 @@ func NewGenericConfig() *genericapiserver.RecommendedConfig { // to stop the process, and there is no later point at which that is still a // startup failure rather than a caller's 403. func (c *ServerConfig) Complete() (CompletedServerConfig, error) { - if c.Extra.Store == nil { + if c.Extra.LogStore == nil { return CompletedServerConfig{}, fmt.Errorf("a storage backend is required") } + if c.Extra.MetricStore == nil { + return CompletedServerConfig{}, fmt.Errorf("no metric store configured") + } if c.Extra.Logger == nil { return CompletedServerConfig{}, fmt.Errorf("a logger is required") } - router, err := newRouter(c.Extra.Logger, c.Extra.Store, c.Extra.Query) + router, err := newRouter(c.Extra.Logger, c.Extra.LogStore, c.Extra.MetricStore, c.Extra.Query) if err != nil { return CompletedServerConfig{}, err } @@ -122,8 +126,8 @@ func (c *ServerConfig) Complete() (CompletedServerConfig, error) { type Server struct { GenericAPIServer *genericapiserver.GenericAPIServer - logger *slog.Logger - store storage.LogStore + logger *slog.Logger + logStore storage.LogStore } // New builds the server and installs everything queryapi adds to the generic @@ -135,15 +139,18 @@ func (c CompletedServerConfig) New() (*Server, error) { return nil, err } - s := &Server{GenericAPIServer: generic, logger: c.extra.Logger, store: c.extra.Store} + s := &Server{GenericAPIServer: generic, logger: c.extra.Logger, logStore: c.extra.LogStore} s.installDiscovery() s.installRoutes(c.extra.Logger, c.router) // The framework serves /readyz; this is the check that used to be a // hand-registered route. /healthz and /livez stay independent of storage, // or an outage that only makes queries fail would also restart the pod. - if err := generic.AddReadyzChecks(storageReady(c.extra.Store, c.extra.Query.QueryTimeout)); err != nil { - return nil, fmt.Errorf("register the storage readiness check: %w", err) + if err := generic.AddReadyzChecks( + storageReady("logs-storage", c.extra.LogStore, c.extra.Query.QueryTimeout), + storageReady("metrics-storage", c.extra.MetricStore, c.extra.Query.QueryTimeout), + ); err != nil { + return nil, fmt.Errorf("register the storage readiness checks: %w", err) } return s, nil @@ -208,7 +215,7 @@ func (s *Server) installRoutes(logger *slog.Logger, r *router) { // closing the storage backend. func (s *Server) Run(ctx context.Context) error { defer func() { - closer, ok := s.store.(io.Closer) + closer, ok := s.logStore.(io.Closer) if !ok { return } diff --git a/queryapi/authorization_test.go b/queryapi/authorization_test.go index 80aeb23..5d345df 100644 --- a/queryapi/authorization_test.go +++ b/queryapi/authorization_test.go @@ -10,6 +10,7 @@ import ( "io" "net/http" "net/http/httptest" + "strings" "testing" "github.com/getkin/kin-openapi/openapi3filter" @@ -38,7 +39,6 @@ func forwardIdentity(req *http.Request) { // authorizedRoutes is every route the router registers, with the status it // answers once the caller is authorized and the permission it is reviewed for. -// The metrics handlers are stubs, hence the 501s -- they are gated all the same. var authorizedRoutes = []struct { name string method string @@ -67,24 +67,24 @@ var authorizedRoutes = []struct { http.StatusOK, authz.Permission{Resource: "logs", Verb: "getSeries"}, }, { - "prom query", http.MethodPost, "/metrics/api/v1/query", - http.StatusNotImplemented, authz.Permission{Resource: "metrics", Verb: "query"}, + "prom query", http.MethodPost, "/metrics/api/v1/query?query=up", + http.StatusOK, authz.Permission{Resource: "metrics", Verb: "query"}, }, { - "prom query_range", http.MethodPost, "/metrics/api/v1/query_range", - http.StatusNotImplemented, authz.Permission{Resource: "metrics", Verb: "query"}, + "prom query_range", http.MethodPost, "/metrics/api/v1/query_range?query=up&start=0&end=60&step=15", + http.StatusOK, authz.Permission{Resource: "metrics", Verb: "query"}, }, { "prom labels", http.MethodGet, "/metrics/api/v1/labels", - http.StatusNotImplemented, authz.Permission{Resource: "metrics", Verb: "getLabels"}, + http.StatusOK, authz.Permission{Resource: "metrics", Verb: "getLabels"}, }, { "prom label values", http.MethodGet, "/metrics/api/v1/label/service/values", - http.StatusNotImplemented, authz.Permission{Resource: "metrics", Verb: "getLabels"}, + http.StatusOK, authz.Permission{Resource: "metrics", Verb: "getLabels"}, }, { - "prom series", http.MethodGet, "/metrics/api/v1/series", - http.StatusNotImplemented, authz.Permission{Resource: "metrics", Verb: "getSeries"}, + "prom series", http.MethodGet, "/metrics/api/v1/series?match[]=up", + http.StatusOK, authz.Permission{Resource: "metrics", Verb: "getSeries"}, }, } @@ -132,7 +132,7 @@ func TestEveryRouteIsAuthorized(t *testing.T) { for _, tc := range authorizedRoutes { t.Run(tc.name, func(t *testing.T) { allow := &stubAuthorizer{decision: authorizer.DecisionAllow} - allowed := serve(t, fake.New(2), queryapi.DefaultConfig(), withAuthorizer(allow)) + allowed := serve(t, fake.NewLogStore(2), queryapi.DefaultConfig(), withAuthorizer(allow)) if got := do(t, allowed, tc.method, prefix+tc.path); got != tc.allowed { t.Errorf("allowed: status = %d, want %d", got, tc.allowed) @@ -148,7 +148,7 @@ func TestEveryRouteIsAuthorized(t *testing.T) { authz.APIGroup, tc.perm.Resource, tc.perm.Verb) } - denied := serve(t, fake.New(2), queryapi.DefaultConfig(), + denied := serve(t, fake.NewLogStore(2), queryapi.DefaultConfig(), withAuthorizer(&stubAuthorizer{decision: authorizer.DecisionDeny})) if got := do(t, denied, tc.method, prefix+tc.path); got != http.StatusForbidden { t.Errorf("denied: status = %d, want 403", got) @@ -182,14 +182,14 @@ func TestAuthorizerFailureIsNotServed(t *testing.T) { for name, tc := range cases { t.Run(name, func(t *testing.T) { - store := &recordingStore{} - srv := serve(t, store, queryapi.DefaultConfig(), withAuthorizer(tc.delegate)) + logStore := &recordingLogStore{} + srv := serve(t, logStore, queryapi.DefaultConfig(), withAuthorizer(tc.delegate)) got := do(t, srv, http.MethodGet, prefix+`/logs/loki/api/v1/query?query={service_name="waf"}`) if got != tc.want { t.Errorf("status = %d, want %d", got, tc.want) } - if seen := store.seen(); seen != "" { + if seen := logStore.seen(); seen != "" { t.Errorf("backend saw project %q, want none", seen) } }) @@ -208,13 +208,35 @@ func TestNilAuthorizerIsRefused(t *testing.T) { _, err := (&queryapi.ServerConfig{ Generic: generic, Extra: queryapi.ExtraConfig{ - Logger: discardLogger(), - Store: fake.New(2), - Query: queryapi.DefaultConfig(), + Logger: discardLogger(), + LogStore: fake.NewLogStore(2), + MetricStore: fake.NewMetricStore(), + Query: queryapi.DefaultConfig(), }, }).Complete() - if err == nil { - t.Fatal("Complete() = nil error, want a refusal to build a server with no authorizer") + if err == nil || !strings.Contains(err.Error(), "an authorizer is required") { + t.Fatalf("Complete() error = %v, want the authorizer refusal", err) + } +} + +// TestNilMetricStoreIsRefused pins that the server cannot be assembled +// without a metric store. +func TestNilMetricStoreIsRefused(t *testing.T) { + generic := queryapi.NewGenericConfig() + generic.ExternalAddress = "queryapi.test:443" + generic.Authentication.Authenticator = testAuthenticator() + generic.Authorization.Authorizer = allowAll().authz + + _, err := (&queryapi.ServerConfig{ + Generic: generic, + Extra: queryapi.ExtraConfig{ + Logger: discardLogger(), + LogStore: fake.NewLogStore(2), + Query: queryapi.DefaultConfig(), + }, + }).Complete() + if err == nil || !strings.Contains(err.Error(), "no metric store configured") { + t.Fatalf("Complete() error = %v, want the metric store refusal", err) } } @@ -225,7 +247,7 @@ func TestNilAuthorizerIsRefused(t *testing.T) { // route could not evaluate a policy bound to a group. func TestReviewCarriesTheWholeCaller(t *testing.T) { az := &stubAuthorizer{decision: authorizer.DecisionAllow} - srv := serve(t, fake.New(2), queryapi.DefaultConfig(), withAuthorizer(az)) + srv := serve(t, fake.NewLogStore(2), queryapi.DefaultConfig(), withAuthorizer(az)) if got := do(t, srv, http.MethodGet, prefix+"/logs/loki/api/v1/label/severity/values"); got != http.StatusOK { t.Fatalf("status = %d, want 200", got) @@ -284,8 +306,8 @@ func TestUnmappedPathIsDeniedWithoutAReview(t *testing.T) { for name, path := range paths { t.Run(name, func(t *testing.T) { az := &stubAuthorizer{decision: authorizer.DecisionAllow} - store := &recordingStore{} - srv := serve(t, store, queryapi.DefaultConfig(), withAuthorizer(az)) + logStore := &recordingLogStore{} + srv := serve(t, logStore, queryapi.DefaultConfig(), withAuthorizer(az)) if got := do(t, srv, http.MethodGet, path); got != http.StatusForbidden { t.Errorf("status = %d, want 403", got) @@ -293,7 +315,7 @@ func TestUnmappedPathIsDeniedWithoutAReview(t *testing.T) { if reviews := az.reviews(); len(reviews) != 0 { t.Errorf("authorizer saw %d telemetry reviews, want none", len(reviews)) } - if seen := store.seen(); seen != "" { + if seen := logStore.seen(); seen != "" { t.Errorf("backend saw project %q, want none", seen) } }) @@ -305,7 +327,7 @@ func TestUnmappedPathIsDeniedWithoutAReview(t *testing.T) { // and it never becomes the tenancy a review is evaluated against. func TestCraftedProjectPathIsNotReviewed(t *testing.T) { az := &stubAuthorizer{decision: authorizer.DecisionAllow} - srv := serve(t, fake.New(2), queryapi.DefaultConfig(), withAuthorizer(az)) + srv := serve(t, fake.NewLogStore(2), queryapi.DefaultConfig(), withAuthorizer(az)) if got := do(t, srv, http.MethodGet, prefix+"/logs/loki/api/v1/label/projects/values"); got != http.StatusOK { t.Fatalf("status = %d, want 200", got) @@ -327,7 +349,7 @@ func TestCraftedProjectPathIsNotReviewed(t *testing.T) { // says so, and this checks that it still does. func TestForbiddenResponseIsTheFrameworkStatus(t *testing.T) { router := loadSpec(t) - srv := serve(t, fake.New(2), queryapi.DefaultConfig(), + srv := serve(t, fake.NewLogStore(2), queryapi.DefaultConfig(), withAuthorizer(&stubAuthorizer{decision: authorizer.DecisionDeny})) const path = "/logs/loki/api/v1/labels" @@ -379,8 +401,8 @@ func TestAuthenticationFailureIsUnauthorized(t *testing.T) { d := withAuthorizer(az) d.authn = failingAuthenticator{} - store := &recordingStore{} - srv := serve(t, store, queryapi.DefaultConfig(), d) + logStore := &recordingLogStore{} + srv := serve(t, logStore, queryapi.DefaultConfig(), d) got := do(t, srv, http.MethodGet, prefix+`/logs/loki/api/v1/query?query={service_name="waf"}`) if got != http.StatusUnauthorized { @@ -389,7 +411,7 @@ func TestAuthenticationFailureIsUnauthorized(t *testing.T) { if reviews := az.reviews(); len(reviews) != 0 { t.Errorf("authorizer saw %d reviews, want none", len(reviews)) } - if seen := store.seen(); seen != "" { + if seen := logStore.seen(); seen != "" { t.Errorf("backend saw project %q, want none", seen) } } @@ -416,8 +438,8 @@ func TestNoHeaderNamesTheProject(t *testing.T) { // An allow-everything reviewer, so that anything reaching the // backend would be visible rather than hidden behind a 403. az := &stubAuthorizer{decision: authorizer.DecisionAllow} - store := &recordingStore{} - srv := serve(t, store, queryapi.DefaultConfig(), withAuthorizer(az)) + logStore := &recordingLogStore{} + srv := serve(t, logStore, queryapi.DefaultConfig(), withAuthorizer(az)) code, _ := doBody(t, srv, http.MethodGet, prefix+`/logs/loki/api/v1/query?query={service_name="waf"}`, @@ -430,7 +452,7 @@ func TestNoHeaderNamesTheProject(t *testing.T) { t.Errorf("review carried parent-name %v for a caller that named its own project", got) } } - if seen := store.seen(); seen != "" { + if seen := logStore.seen(); seen != "" { t.Errorf("backend saw project %q, want none", seen) } }) @@ -444,8 +466,8 @@ func TestNoHeaderNamesTheProject(t *testing.T) { // miloauth filter behind the routes is. func TestUnscopedCallerIsNotServed(t *testing.T) { az := &stubAuthorizer{decision: authorizer.DecisionAllow} - store := &recordingStore{} - srv := serve(t, store, queryapi.DefaultConfig(), withAuthorizer(az)) + logStore := &recordingLogStore{} + srv := serve(t, logStore, queryapi.DefaultConfig(), withAuthorizer(az)) code, _ := doBody(t, srv, http.MethodGet, prefix+`/logs/loki/api/v1/query?query={service_name="waf"}`, @@ -458,7 +480,7 @@ func TestUnscopedCallerIsNotServed(t *testing.T) { if code != http.StatusForbidden { t.Errorf("status = %d, want 403", code) } - if seen := store.seen(); seen != "" { + if seen := logStore.seen(); seen != "" { t.Errorf("backend saw project %q, want none", seen) } } @@ -467,7 +489,7 @@ func TestUnscopedCallerIsNotServed(t *testing.T) { // write, the way TestHandlersConformToSpec checks the success ones. func TestErrorResponsesConformToSpec(t *testing.T) { router := loadSpec(t) - srv := serve(t, fake.New(2), queryapi.DefaultConfig(), allowAll()) + srv := serve(t, fake.NewLogStore(2), queryapi.DefaultConfig(), allowAll()) const path = "/logs/loki/api/v1/series" // match[] is required, so this is a handler-written 400. diff --git a/queryapi/cmd/main.go b/queryapi/cmd/main.go index 6d3e8c0..516db0f 100644 --- a/queryapi/cmd/main.go +++ b/queryapi/cmd/main.go @@ -20,6 +20,7 @@ import ( "go.datum.net/o11y/queryapi/internal/storage" "go.datum.net/o11y/queryapi/internal/storage/clickhouse" "go.datum.net/o11y/queryapi/internal/storage/fake" + "go.datum.net/o11y/queryapi/internal/storage/prometheus" ) func main() { @@ -39,16 +40,24 @@ func run(logger *slog.Logger, args []string) error { if err := fs.Parse(args); err != nil { return fmt.Errorf("parse flags: %w", err) } + if err := opts.Complete(); err != nil { + return fmt.Errorf("invalid configuration: %w", err) + } if err := opts.Validate(); err != nil { return fmt.Errorf("invalid configuration: %w", err) } - store, err := newStore(opts.Query) + logStore, err := newLogStore(opts.Query) if err != nil { return fmt.Errorf("configure storage: %w", err) } - cfg, err := opts.Config(logger, store) + metricStore, err := newMetricStore(opts.Query) + if err != nil { + return fmt.Errorf("configure metrics storage: %w", err) + } + + cfg, err := opts.Config(logger, logStore, metricStore) if err != nil { return err } @@ -68,11 +77,11 @@ func run(logger *slog.Logger, args []string) error { return server.Run(ctx) } -// newStore builds the configured storage backend. -func newStore(cfg queryapi.Config) (storage.LogStore, error) { - switch cfg.Storage { +// newLogStore builds the configured log storage backend. +func newLogStore(cfg queryapi.Config) (storage.LogStore, error) { + switch cfg.LogsStorage { case "fake": - return fake.New(cfg.FakeRate), nil + return fake.NewLogStore(cfg.FakeRate), nil case "clickhouse": chCfg, err := clickhouse.ConfigFromEnv() if err != nil { @@ -80,6 +89,27 @@ func newStore(cfg queryapi.Config) (storage.LogStore, error) { } return clickhouse.New(chCfg) default: - return nil, fmt.Errorf("unknown storage backend %q", cfg.Storage) + return nil, fmt.Errorf("unknown storage backend %q", cfg.LogsStorage) + } +} + +// newMetricStore builds the configured metric backend. +func newMetricStore(cfg queryapi.Config) (storage.MetricStore, error) { + switch cfg.MetricsStorage { + case "fake": + return fake.NewMetricStore(), nil + case "prometheus": + tenancy, err := prometheus.ParseTenancy(cfg.PrometheusTenancy) + if err != nil { + return nil, err + } + return prometheus.New(prometheus.Config{ + URL: cfg.PrometheusURL, + Tenancy: tenancy, + ProjectLabel: cfg.PrometheusProjectLabel, + MaxResponseBytes: cfg.MaxResponseBytes, + }) + default: + return nil, fmt.Errorf("unknown metrics storage backend %q", cfg.MetricsStorage) } } diff --git a/queryapi/config.go b/queryapi/config.go index 00f6508..6d56306 100644 --- a/queryapi/config.go +++ b/queryapi/config.go @@ -6,14 +6,28 @@ import ( "errors" "fmt" "time" + + "go.datum.net/o11y/queryapi/internal/storage/prometheus" ) // Config is what the query handlers themselves need. Everything about serving, // authentication and authorization lives in Options below, because it belongs // to the apiserver runtime rather than to a handler. type Config struct { - // Storage selects the backend: "fake" or "clickhouse". - Storage string + // LogsStorage selects the log backend: "fake" or "clickhouse". + LogsStorage string + + // MetricsStorage selects the metric backend: "fake" or "prometheus". + MetricsStorage string + + // PrometheusURL, PrometheusTenancy and PrometheusProjectLabel configure + // the prometheus backend. See internal/storage/prometheus. + PrometheusURL string + PrometheusTenancy string + PrometheusProjectLabel string + + // MaxResponseBytes bounds a metric backend response body. + MaxResponseBytes int64 // FakeRate is the fake backend's lines per second. FakeRate float64 @@ -21,6 +35,9 @@ type Config struct { // QueryTimeout bounds a single storage call, and the /readyz ping. QueryTimeout time.Duration + // MaxRange bounds a query_range window. + MaxRange time.Duration + // DefaultLimit and MaxLimit bound returned rows. Requests above MaxLimit // are clamped, not rejected, matching Loki. DefaultLimit int @@ -30,11 +47,15 @@ type Config struct { // DefaultConfig returns a Config with production-reasonable defaults. func DefaultConfig() Config { return Config{ - Storage: "fake", - FakeRate: 2, - QueryTimeout: 30 * time.Second, - DefaultLimit: 100, - MaxLimit: 5000, + LogsStorage: "fake", + MetricsStorage: "fake", + PrometheusProjectLabel: "resourcemanager_datumapis_com_project_name", + MaxResponseBytes: 32 << 20, + FakeRate: 2, + QueryTimeout: 30 * time.Second, + MaxRange: 31 * 24 * time.Hour, + DefaultLimit: 100, + MaxLimit: 5000, } } @@ -53,5 +74,23 @@ func (c Config) Validate() error { if c.QueryTimeout <= 0 { errs = append(errs, fmt.Errorf("query-timeout must be greater than zero, got %s", c.QueryTimeout)) } + if c.MaxRange <= 0 { + errs = append(errs, fmt.Errorf("max-range must be greater than zero, got %s", c.MaxRange)) + } + if c.MaxResponseBytes <= 0 { + errs = append(errs, fmt.Errorf("max-response-bytes must be greater than zero, got %d", c.MaxResponseBytes)) + } + switch c.MetricsStorage { + case "fake": + case "prometheus": + if c.PrometheusURL == "" { + errs = append(errs, errors.New("prometheus-url is required with --metrics-storage=prometheus")) + } + if _, err := prometheus.ParseTenancy(c.PrometheusTenancy); err != nil { + errs = append(errs, err) + } + default: + errs = append(errs, fmt.Errorf("unknown metrics storage backend %q: want fake or prometheus", c.MetricsStorage)) + } return errors.Join(errs...) } diff --git a/queryapi/config_test.go b/queryapi/config_test.go index 6173a2b..bdb4f0e 100644 --- a/queryapi/config_test.go +++ b/queryapi/config_test.go @@ -3,6 +3,7 @@ package queryapi import ( + "io" "strings" "testing" "time" @@ -54,16 +55,17 @@ func TestDefaultOptionsServeNothingFromEtcd(t *testing.T) { "secure-port", "tls-cert-file", "tls-private-key-file", "authentication-kubeconfig", "requestheader-client-ca-file", "authorization-kubeconfig", "authorization-always-allow-paths", - "storage", "query-timeout", + "logs-storage", "storage", "query-timeout", } { if fs.Lookup(name) == nil { t.Errorf("--%s is not registered", name) } } // There has never been a flag by which a caller names its own project, and - // there must not be one now. + // there must not be one now. --prometheus-project-label names the backend + // label the tenancy mode constrains, not a project. fs.VisitAll(func(f *pflag.Flag) { - if strings.Contains(f.Name, "project") { + if strings.Contains(f.Name, "project") && f.Name != "prometheus-project-label" { t.Errorf("--%s exists: nothing but a verified identity may name the project", f.Name) } }) @@ -95,3 +97,72 @@ func TestOptionsRejectAlwaysAllowingTheAPI(t *testing.T) { t.Fatal("Validate() = nil, want a refusal") } } + +func TestLogsStorageFlags(t *testing.T) { + cases := []struct { + name string + args []string + want string + wantErr bool + }{ + {"default", nil, "fake", false}, + {"logs-storage alone", []string{"--logs-storage=clickhouse"}, "clickhouse", false}, + {"deprecated storage alone", []string{"--storage=clickhouse"}, "clickhouse", false}, + {"both agree", []string{"--storage=clickhouse", "--logs-storage=clickhouse"}, "clickhouse", false}, + {"both conflict", []string{"--storage=fake", "--logs-storage=clickhouse"}, "", true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + o := NewOptions() + fs := pflag.NewFlagSet("test", pflag.ContinueOnError) + fs.SetOutput(io.Discard) // the deprecation warning + o.AddFlags(fs) + if err := fs.Parse(tc.args); err != nil { + t.Fatalf("parse: %v", err) + } + err := o.Complete() + if tc.wantErr { + if err == nil { + t.Fatalf("Complete() = nil, want a conflict error") + } + return + } + if err != nil { + t.Fatalf("Complete() = %v", err) + } + if o.Query.LogsStorage != tc.want { + t.Errorf("LogsStorage = %q, want %q", o.Query.LogsStorage, tc.want) + } + }) + } +} + +func TestMetricsStorageFlags(t *testing.T) { + cases := []struct { + name string + args []string + wantErr bool + }{ + {"default fake", nil, false}, + {"prometheus complete", []string{"--metrics-storage=prometheus", "--prometheus-url=http://vm:8429", "--prometheus-tenancy=extra-label"}, false}, + {"prometheus no url", []string{"--metrics-storage=prometheus", "--prometheus-tenancy=extra-label"}, true}, + {"prometheus no tenancy", []string{"--metrics-storage=prometheus", "--prometheus-url=http://vm:8429"}, true}, + {"prometheus unknown tenancy", []string{"--metrics-storage=prometheus", "--prometheus-url=http://vm:8429", "--prometheus-tenancy=rewrite"}, true}, + {"unknown backend", []string{"--metrics-storage=thanos"}, true}, + {"zero response limit", []string{"--max-response-bytes=0"}, true}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + o := NewOptions() + fs := pflag.NewFlagSet("test", pflag.ContinueOnError) + o.AddFlags(fs) + if err := fs.Parse(tc.args); err != nil { + t.Fatalf("parse: %v", err) + } + err := o.Query.Validate() + if (err != nil) != tc.wantErr { + t.Errorf("Validate() = %v, wantErr %v", err, tc.wantErr) + } + }) + } +} diff --git a/queryapi/deps_test.go b/queryapi/deps_test.go index f0239c3..d01d6c9 100644 --- a/queryapi/deps_test.go +++ b/queryapi/deps_test.go @@ -26,6 +26,7 @@ import ( "go.datum.net/o11y/queryapi" "go.datum.net/o11y/queryapi/internal/authz" "go.datum.net/o11y/queryapi/internal/storage" + "go.datum.net/o11y/queryapi/internal/storage/fake" ) // testAuthenticator reads the X-Remote-* headers Milo's aggregator sets, the @@ -149,7 +150,15 @@ func withAuthorizer(delegate authorizer.Authorizer) deps { // exactly one thing, the front proxy's client certificate check, which // testAuthenticator stands in for; every other filter, the route table, the // request info resolver and the authorization chain are the real ones. -func serve(t *testing.T, store storage.LogStore, cfg queryapi.Config, d deps) *httptest.Server { +func serve(t *testing.T, logStore storage.LogStore, cfg queryapi.Config, d deps) *httptest.Server { + t.Helper() + return serveWithMetricStore(t, logStore, fake.NewMetricStore(), cfg, d) +} + +// serveWithMetricStore is serve with a chosen metric store. +func serveWithMetricStore( + t *testing.T, logStore storage.LogStore, metricStore storage.MetricStore, cfg queryapi.Config, d deps, +) *httptest.Server { t.Helper() generic := queryapi.NewGenericConfig() @@ -164,9 +173,10 @@ func serve(t *testing.T, store storage.LogStore, cfg queryapi.Config, d deps) *h serverConfig := &queryapi.ServerConfig{ Generic: generic, Extra: queryapi.ExtraConfig{ - Logger: slog.New(slog.DiscardHandler), - Store: store, - Query: cfg, + Logger: slog.New(slog.DiscardHandler), + LogStore: logStore, + MetricStore: metricStore, + Query: cfg, }, } completed, err := serverConfig.Complete() diff --git a/queryapi/go.mod b/queryapi/go.mod index 83521b7..23bf5f5 100644 --- a/queryapi/go.mod +++ b/queryapi/go.mod @@ -6,6 +6,7 @@ require ( github.com/ClickHouse/clickhouse-go/v2 v2.42.0 github.com/getkin/kin-openapi v0.142.0 github.com/prometheus/client_golang v1.24.0 + github.com/prometheus/common v0.70.0 github.com/spf13/pflag v1.0.9 go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.69.0 k8s.io/api v0.35.0 @@ -66,7 +67,6 @@ require ( github.com/pierrec/lz4/v4 v4.1.22 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/client_model v0.6.2 // indirect - github.com/prometheus/common v0.70.0 // indirect github.com/prometheus/procfs v0.21.1 // indirect github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 // indirect github.com/segmentio/asm v1.2.1 // indirect diff --git a/queryapi/handler.go b/queryapi/handler.go index 8653807..7d7b065 100644 --- a/queryapi/handler.go +++ b/queryapi/handler.go @@ -21,9 +21,10 @@ import ( ) type handler struct { - logger *slog.Logger - store storage.LogStore - cfg Config + logger *slog.Logger + logStore storage.LogStore + metricStore storage.MetricStore + cfg Config } // parseLogQL parses raw, writing a 400 and returning false on failure. @@ -104,7 +105,7 @@ func (h *handler) serveLogs(w http.ResponseWriter, r *http.Request, q *logql.Que ctx, cancel := h.withTimeout(r) defer cancel() - iter, err := h.store.QueryLogs(ctx, storage.LogQuery{ + iter, err := h.logStore.QueryLogs(ctx, storage.LogQuery{ Matchers: q.Matchers, Filters: q.Filters, Range: tr, @@ -132,7 +133,7 @@ func (h *handler) lokiLabelNames(w http.ResponseWriter, r *http.Request) { ctx, cancel := h.withTimeout(r) defer cancel() - names, err := h.store.LabelNames(ctx, tr) + names, err := h.logStore.LabelNames(ctx, tr) if err != nil { h.writeStorageError(w, err) return @@ -154,7 +155,7 @@ func (h *handler) lokiLabelValues(w http.ResponseWriter, r *http.Request) { ctx, cancel := h.withTimeout(r) defer cancel() - values, err := h.store.LabelValues(ctx, r.PathValue("name"), tr) + values, err := h.logStore.LabelValues(ctx, r.PathValue("name"), tr) if err != nil { h.writeStorageError(w, err) return @@ -186,7 +187,7 @@ func (h *handler) lokiSeries(w http.ResponseWriter, r *http.Request) { if !ok { return } - series, err := h.store.Series(ctx, q.Matchers, tr) + series, err := h.logStore.Series(ctx, q.Matchers, tr) if err != nil { h.writeStorageError(w, err) return @@ -201,26 +202,6 @@ func (h *handler) lokiSeries(w http.ResponseWriter, r *http.Request) { writeData(w, out) } -func (h *handler) promQuery(w http.ResponseWriter, _ *http.Request) { - writeNotImplemented(w, "promQuery") -} - -func (h *handler) promQueryRange(w http.ResponseWriter, _ *http.Request) { - writeNotImplemented(w, "promQueryRange") -} - -func (h *handler) promSeries(w http.ResponseWriter, _ *http.Request) { - writeNotImplemented(w, "promSeries") -} - -func (h *handler) promLabelNames(w http.ResponseWriter, _ *http.Request) { - writeNotImplemented(w, "promLabelNames") -} - -func (h *handler) promLabelValues(w http.ResponseWriter, _ *http.Request) { - writeNotImplemented(w, "promLabelValues") -} - func orEmpty(values []string) []string { if values == nil { return []string{} @@ -249,10 +230,6 @@ func (h *handler) writeStorageError(w http.ResponseWriter, err error) { } } -func writeNotImplemented(w http.ResponseWriter, op string) { - writeError(w, http.StatusNotImplemented, op+" is not yet implemented") -} - // writeError writes the Loki/Prometheus-style error envelope openapi.yaml's // ErrorResponse describes. func writeError(w http.ResponseWriter, status int, message string) { diff --git a/queryapi/handler_test.go b/queryapi/handler_test.go index 371f8aa..a3a2d79 100644 --- a/queryapi/handler_test.go +++ b/queryapi/handler_test.go @@ -63,7 +63,7 @@ func TestHandlersConformToSpec(t *testing.T) { router := loadSpec(t) cfg := queryapi.DefaultConfig() - srv := serve(t, fake.New(2), cfg, allowAll()) + srv := serve(t, fake.NewLogStore(2), cfg, allowAll()) cases := []struct { name string @@ -166,8 +166,8 @@ func TestHandlersConformToSpec(t *testing.T) { // authenticated and then denied, which is a 403 rather than the 401 queryapi // used to answer before it ran on the apiserver runtime. func TestAnonymousRequestIsNotServed(t *testing.T) { - store := &recordingStore{} - srv := serve(t, store, queryapi.DefaultConfig(), allowAll()) + logStore := &recordingLogStore{} + srv := serve(t, logStore, queryapi.DefaultConfig(), allowAll()) resp, err := http.Get(srv.URL + prefix + "/logs/loki/api/v1/query?query=%7Bservice_name%3D%22waf%22%7D") if err != nil { @@ -182,7 +182,7 @@ func TestAnonymousRequestIsNotServed(t *testing.T) { if resp.StatusCode != http.StatusForbidden { t.Fatalf("status = %d, want 403", resp.StatusCode) } - if got := store.seen(); got != "" { + if got := logStore.seen(); got != "" { t.Errorf("backend saw project %q, want none", got) } } @@ -192,7 +192,7 @@ func TestAnonymousRequestIsNotServed(t *testing.T) { // duplicates, and a missing selector is a 400. func TestSeriesUnionsSelectors(t *testing.T) { cfg := queryapi.DefaultConfig() - srv := serve(t, fake.New(20), cfg, allowAll()) + srv := serve(t, fake.NewLogStore(20), cfg, allowAll()) get := func(t *testing.T, query string) (int, []map[string]string) { t.Helper() @@ -253,14 +253,14 @@ func TestSeriesUnionsSelectors(t *testing.T) { } } -// recordingStore records the project each query was scoped to, so a routing +// recordingLogStore records the project each query was scoped to, so a routing // test can assert the identity that reached the backend. -type recordingStore struct { +type recordingLogStore struct { mu sync.Mutex project string } -func (s *recordingStore) note(ctx context.Context) error { +func (s *recordingLogStore) note(ctx context.Context) error { id, ok := miloauth.ProjectID(ctx) if !ok { return storage.ErrNoProject @@ -271,13 +271,13 @@ func (s *recordingStore) note(ctx context.Context) error { return nil } -func (s *recordingStore) seen() string { +func (s *recordingLogStore) seen() string { s.mu.Lock() defer s.mu.Unlock() return s.project } -func (s *recordingStore) QueryLogs(ctx context.Context, q storage.LogQuery) (storage.LogIterator, error) { +func (s *recordingLogStore) QueryLogs(ctx context.Context, q storage.LogQuery) (storage.LogIterator, error) { if err := s.note(ctx); err != nil { return nil, err } @@ -288,19 +288,19 @@ func (s *recordingStore) QueryLogs(ctx context.Context, q storage.LogQuery) (sto }}, nil } -func (s *recordingStore) LabelNames(ctx context.Context, _ storage.TimeRange) ([]string, error) { +func (s *recordingLogStore) LabelNames(ctx context.Context, _ storage.TimeRange) ([]string, error) { return nil, s.note(ctx) } -func (s *recordingStore) LabelValues(ctx context.Context, _ string, _ storage.TimeRange) ([]string, error) { +func (s *recordingLogStore) LabelValues(ctx context.Context, _ string, _ storage.TimeRange) ([]string, error) { return nil, s.note(ctx) } -func (s *recordingStore) Series(ctx context.Context, _ []logql.LabelMatcher, _ storage.TimeRange) ([]storage.LabelSet, error) { +func (s *recordingLogStore) Series(ctx context.Context, _ []logql.LabelMatcher, _ storage.TimeRange) ([]storage.LabelSet, error) { return nil, s.note(ctx) } -func (s *recordingStore) Ping(context.Context) error { return nil } +func (s *recordingLogStore) Ping(context.Context) error { return nil } type oneRowIterator struct { row storage.Row @@ -352,9 +352,9 @@ func TestForwardedPathShapes(t *testing.T) { for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - store := &recordingStore{} + logStore := &recordingLogStore{} cfg := queryapi.DefaultConfig() - srv := serve(t, store, cfg, allowAll()) + srv := serve(t, logStore, cfg, allowAll()) req, err := http.NewRequest(http.MethodGet, srv.URL+tc.path+`?query={service_name="waf"}&limit=5`, nil) @@ -402,7 +402,7 @@ func TestForwardedPathShapes(t *testing.T) { if got := payload.Data.Result[0].Values[0][1]; got != "recorded" { t.Errorf("line = %q, want %q", got, "recorded") } - if got := store.seen(); got != "proj-abc" { + if got := logStore.seen(); got != "proj-abc" { t.Errorf("backend saw project %q, want %q", got, "proj-abc") } }) @@ -415,7 +415,7 @@ func TestForwardedPathShapes(t *testing.T) { // query works. Anything else metric-shaped must still be a 400. func TestGrafanaHealthProbe(t *testing.T) { cfg := queryapi.DefaultConfig() - srv := serve(t, fake.New(2), cfg, allowAll()) + srv := serve(t, fake.NewLogStore(2), cfg, allowAll()) get := func(t *testing.T, query string) (int, string) { t.Helper() @@ -505,8 +505,8 @@ func TestCraftedProjectPathIsNotAnIdentity(t *testing.T) { const crafted = prefix + "/logs/loki/api/v1/label/projects/evil-corp/control-plane/v1alpha1/logs/loki/api/v1/query" t.Run("no identity at all", func(t *testing.T) { - store := &recordingStore{} - srv := serve(t, store, queryapi.DefaultConfig(), allowAll()) + logStore := &recordingLogStore{} + srv := serve(t, logStore, queryapi.DefaultConfig(), allowAll()) resp, err := http.Get(srv.URL + crafted + `?query={service_name="waf"}&limit=1`) if err != nil { @@ -523,14 +523,14 @@ func TestCraftedProjectPathIsNotAnIdentity(t *testing.T) { if resp.StatusCode != http.StatusForbidden { t.Errorf("status = %d, want 403", resp.StatusCode) } - if got := store.seen(); got != "" { + if got := logStore.seen(); got != "" { t.Errorf("backend saw project %q, want none", got) } }) t.Run("identity present, crafted path ignored", func(t *testing.T) { - store := &recordingStore{} - srv := serve(t, store, queryapi.DefaultConfig(), allowAll()) + logStore := &recordingLogStore{} + srv := serve(t, logStore, queryapi.DefaultConfig(), allowAll()) req, err := http.NewRequest(http.MethodGet, srv.URL+crafted+`?query={service_name="waf"}&limit=1`, nil) @@ -552,11 +552,11 @@ func TestCraftedProjectPathIsNotAnIdentity(t *testing.T) { // The crafted path must never become the project. Identity comes only // from the verified user extras; a 200 is served only scoped to // proj-abc, and the backend must never see evil-corp. - if got := store.seen(); got == "evil-corp" { + if got := logStore.seen(); got == "evil-corp" { t.Fatalf("backend saw project %q from a crafted path", got) } - if resp.StatusCode == http.StatusOK && store.seen() != "proj-abc" { - t.Errorf("served 200 for project %q, want proj-abc", store.seen()) + if resp.StatusCode == http.StatusOK && logStore.seen() != "proj-abc" { + t.Errorf("served 200 for project %q, want proj-abc", logStore.seen()) } }) } @@ -573,7 +573,7 @@ func TestCraftedProjectPathIsNotAnIdentity(t *testing.T) { func TestDiscoveryIsServed(t *testing.T) { // A reviewer that denies everything, to prove the probe does not depend on // one: the availability of the APIService must not turn on Milo's answer. - srv := serve(t, fake.New(2), queryapi.DefaultConfig(), + srv := serve(t, fake.NewLogStore(2), queryapi.DefaultConfig(), withAuthorizer(&stubAuthorizer{decision: authorizer.DecisionDeny})) probe := func(r *http.Request) { @@ -636,7 +636,7 @@ func TestDiscoveryIsServed(t *testing.T) { // covers them. They are non-resource paths, and the path authorizer abstains // on every resource request, so nothing listed there can open a query route. func TestProbesAreServedWithoutAuthorization(t *testing.T) { - srv := serve(t, fake.New(2), queryapi.DefaultConfig(), + srv := serve(t, fake.NewLogStore(2), queryapi.DefaultConfig(), withAuthorizer(&stubAuthorizer{decision: authorizer.DecisionDeny})) for _, path := range []string{"/healthz", "/livez", "/readyz", "/metrics"} { @@ -651,13 +651,13 @@ func TestProbesAreServedWithoutAuthorization(t *testing.T) { // backend and liveness does not: an outage that only makes queries fail must // take the pod out of service, not restart it. func TestReadyzReportsStorageUnavailable(t *testing.T) { - srv := serve(t, &unreachableStore{}, queryapi.DefaultConfig(), allowAll()) + srv := serve(t, &unreachableLogStore{}, queryapi.DefaultConfig(), allowAll()) code, body := doBody(t, srv, http.MethodGet, "/readyz", nil) if code != http.StatusInternalServerError { t.Errorf("/readyz status = %d, want 500: %s", code, body) } - if !strings.Contains(string(body), "[-]storage failed") { + if !strings.Contains(string(body), "[-]logs-storage failed") { t.Errorf("/readyz body = %s, want the storage check to be the one that failed", body) } if code, _ := doBody(t, srv, http.MethodGet, "/livez", nil); code != http.StatusOK { @@ -668,9 +668,44 @@ func TestReadyzReportsStorageUnavailable(t *testing.T) { } } -// unreachableStore fails Ping so the readiness check actually fails. -type unreachableStore struct{ recordingStore } +// TestReadyzReportsMetricsStorageUnavailable is the metrics counterpart: the +// metric backend alone failing takes /readyz down and leaves liveness up. +func TestReadyzReportsMetricsStorageUnavailable(t *testing.T) { + srv := serveWithMetricStore(t, fake.NewLogStore(2), &unreachableMetricStore{MetricStore: fake.NewMetricStore()}, + queryapi.DefaultConfig(), allowAll()) -func (*unreachableStore) Ping(context.Context) error { return errors.New("dial: connection refused") } + code, body := doBody(t, srv, http.MethodGet, "/readyz", nil) + if code != http.StatusInternalServerError { + t.Errorf("/readyz status = %d, want 500: %s", code, body) + } + if !strings.Contains(string(body), "[-]metrics-storage failed") { + t.Errorf("/readyz body = %s, want the metrics-storage check to be the one that failed", body) + } + if strings.Contains(string(body), "[-]logs-storage failed") { + t.Errorf("/readyz body = %s, log storage should be healthy", body) + } + if code, _ := doBody(t, srv, http.MethodGet, "/livez", nil); code != http.StatusOK { + t.Errorf("/livez status = %d, want 200", code) + } + if code, _ := doBody(t, srv, http.MethodGet, "/healthz", nil); code != http.StatusOK { + t.Errorf("/healthz status = %d, want 200", code) + } +} + +// unreachableMetricStore fails Ping so the metrics readiness check actually fails. +type unreachableMetricStore struct{ *fake.MetricStore } + +func (*unreachableMetricStore) Ping(context.Context) error { + return errors.New("dial: connection refused") +} + +var _ storage.MetricStore = (*unreachableMetricStore)(nil) + +// unreachableLogStore fails Ping so the readiness check actually fails. +type unreachableLogStore struct{ recordingLogStore } + +func (*unreachableLogStore) Ping(context.Context) error { + return errors.New("dial: connection refused") +} -var _ storage.LogStore = (*unreachableStore)(nil) +var _ storage.LogStore = (*unreachableLogStore)(nil) diff --git a/queryapi/health.go b/queryapi/health.go index 331dc2b..e9707cc 100644 --- a/queryapi/health.go +++ b/queryapi/health.go @@ -8,18 +8,18 @@ import ( "time" "k8s.io/apiserver/pkg/server/healthz" - - "go.datum.net/o11y/queryapi/internal/storage" ) +type pinger interface{ Ping(context.Context) error } + // storageReady is queryapi's only addition to the framework's health surface. // // The apiserver runtime serves /healthz, /livez and /readyz itself, so there // is nothing to register a route for -- only a check to add. It goes on // /readyz alone: liveness must not depend on the storage backend, or an // outage that merely makes queries fail would restart the pod as well. -func storageReady(store storage.LogStore, timeout time.Duration) healthz.HealthChecker { - return healthz.NamedCheck("storage", func(r *http.Request) error { +func storageReady(name string, store pinger, timeout time.Duration) healthz.HealthChecker { + return healthz.NamedCheck(name, func(r *http.Request) error { ctx, cancel := context.WithTimeout(r.Context(), timeout) defer cancel() return store.Ping(ctx) diff --git a/queryapi/iam_test.go b/queryapi/iam_test.go index 002a097..cab136d 100644 --- a/queryapi/iam_test.go +++ b/queryapi/iam_test.go @@ -88,7 +88,7 @@ func TestAggregatorPermissionsAreDeclaredAndGranted(t *testing.T) { readYAML(t, filepath.Join(iamDir, "roles", "telemetry-viewer.yaml"), &viewer) granted := sets.New(viewer.Spec.IncludedPermissions...) - router, err := newRouter(slog.New(slog.DiscardHandler), fake.New(1), DefaultConfig()) + router, err := newRouter(slog.New(slog.DiscardHandler), fake.NewLogStore(1), fake.NewMetricStore(), DefaultConfig()) if err != nil { t.Fatalf("newRouter: %v", err) } diff --git a/queryapi/internal/authz/permissions.go b/queryapi/internal/authz/permissions.go index 5016328..c8f5740 100644 --- a/queryapi/internal/authz/permissions.go +++ b/queryapi/internal/authz/permissions.go @@ -42,8 +42,8 @@ type Permission struct { // identifiers, which reading log lines does not necessarily imply, and that // boundary cannot be retrofitted once a single action covers both. // -// The metrics handlers return 501 today. Gating a stub costs nothing and keeps -// the endpoint from shipping unguarded later. +// The metrics handlers are gated like the log handlers, so no endpoint ships +// unguarded. var Permissions = map[Permission]struct{}{ {Resource: "logs", Verb: "query"}: {}, {Resource: "logs", Verb: "getLabels"}: {}, diff --git a/queryapi/internal/storage/fake/fake.go b/queryapi/internal/storage/fake/fake.go index c55c19f..5ff9911 100644 --- a/queryapi/internal/storage/fake/fake.go +++ b/queryapi/internal/storage/fake/fake.go @@ -50,20 +50,20 @@ var ( } ) -// Store is a synthetic storage.LogStore. -type Store struct { +// LogStore is a synthetic storage.LogStore. +type LogStore struct { interval time.Duration } -// New returns a Store generating ratePerSecond lines per second. -func New(ratePerSecond float64) *Store { +// NewLogStore returns a LogStore generating ratePerSecond lines per second. +func NewLogStore(ratePerSecond float64) *LogStore { if ratePerSecond <= 0 { ratePerSecond = 1 } - return &Store{interval: time.Duration(float64(time.Second) / ratePerSecond)} + return &LogStore{interval: time.Duration(float64(time.Second) / ratePerSecond)} } -func (s *Store) QueryLogs(ctx context.Context, q storage.LogQuery) (storage.LogIterator, error) { +func (s *LogStore) QueryLogs(ctx context.Context, q storage.LogQuery) (storage.LogIterator, error) { if _, ok := miloauth.ProjectID(ctx); !ok { return nil, storage.ErrNoProject } @@ -73,7 +73,7 @@ func (s *Store) QueryLogs(ctx context.Context, q storage.LogQuery) (storage.LogI return newIterator(ctx, s.interval, q), nil } -func (s *Store) LabelNames(ctx context.Context, _ storage.TimeRange) ([]string, error) { +func (s *LogStore) LabelNames(ctx context.Context, _ storage.TimeRange) ([]string, error) { if _, ok := miloauth.ProjectID(ctx); !ok { return nil, storage.ErrNoProject } @@ -94,7 +94,7 @@ func (s *Store) LabelNames(ctx context.Context, _ storage.TimeRange) ([]string, return names, nil } -func (s *Store) LabelValues(ctx context.Context, label string, _ storage.TimeRange) ([]string, error) { +func (s *LogStore) LabelValues(ctx context.Context, label string, _ storage.TimeRange) ([]string, error) { if _, ok := miloauth.ProjectID(ctx); !ok { return nil, storage.ErrNoProject } @@ -113,7 +113,7 @@ func (s *Store) LabelValues(ctx context.Context, label string, _ storage.TimeRan return values, nil } -func (s *Store) Series(ctx context.Context, matchers []logql.LabelMatcher, _ storage.TimeRange) ([]storage.LabelSet, error) { +func (s *LogStore) Series(ctx context.Context, matchers []logql.LabelMatcher, _ storage.TimeRange) ([]storage.LabelSet, error) { if _, ok := miloauth.ProjectID(ctx); !ok { return nil, storage.ErrNoProject } @@ -127,7 +127,7 @@ func (s *Store) Series(ctx context.Context, matchers []logql.LabelMatcher, _ sto return out, nil } -func (s *Store) Ping(context.Context) error { return nil } +func (s *LogStore) Ping(context.Context) error { return nil } // catalogue is every label set the generator can produce. Discovery is served // from here rather than from generated rows, so it is instant and populates diff --git a/queryapi/internal/storage/fake/fake_test.go b/queryapi/internal/storage/fake/fake_test.go index 04484f7..0cb0123 100644 --- a/queryapi/internal/storage/fake/fake_test.go +++ b/queryapi/internal/storage/fake/fake_test.go @@ -25,7 +25,7 @@ func testRange() storage.TimeRange { } // drain runs q against s and collects every row. -func drain(t *testing.T, s *fake.Store, q storage.LogQuery) []storage.Row { +func drain(t *testing.T, s *fake.LogStore, q storage.LogQuery) []storage.Row { t.Helper() iter, err := s.QueryLogs(testCtx(), q) if err != nil { @@ -58,7 +58,7 @@ func mustParse(t *testing.T, raw string) *logql.Query { } func TestRequiresProject(t *testing.T) { - s := fake.New(2) + s := fake.NewLogStore(2) _, err := s.QueryLogs(context.Background(), query(nil, nil, 10, storage.DirectionBackward)) if !errors.Is(err, storage.ErrNoProject) { t.Fatalf("QueryLogs without a project = %v, want ErrNoProject", err) @@ -69,11 +69,11 @@ func TestRequiresProject(t *testing.T) { } func TestDeterministic(t *testing.T) { - s := fake.New(2) + s := fake.NewLogStore(2) q := query(nil, nil, 50, storage.DirectionBackward) first := drain(t, s, q) - second := drain(t, fake.New(2), q) + second := drain(t, fake.NewLogStore(2), q) if len(first) != len(second) { t.Fatalf("row counts differ: %d vs %d", len(first), len(second)) @@ -86,7 +86,7 @@ func TestDeterministic(t *testing.T) { } func TestBackwardReturnsNewestFirst(t *testing.T) { - s := fake.New(2) + s := fake.NewLogStore(2) rows := drain(t, s, query(nil, nil, 10, storage.DirectionBackward)) if len(rows) != 10 { @@ -107,7 +107,7 @@ func TestBackwardReturnsNewestFirst(t *testing.T) { } func TestForwardReturnsOldestFirst(t *testing.T) { - s := fake.New(2) + s := fake.NewLogStore(2) rows := drain(t, s, query(nil, nil, 10, storage.DirectionForward)) for i := 1; i < len(rows); i++ { @@ -125,7 +125,7 @@ func TestForwardReturnsOldestFirst(t *testing.T) { } func TestLabelMatcherFilters(t *testing.T) { - s := fake.New(20) + s := fake.NewLogStore(20) q := mustParse(t, `{service_name="waf"}`) rows := drain(t, s, query(q.Matchers, nil, 100, storage.DirectionBackward)) @@ -140,7 +140,7 @@ func TestLabelMatcherFilters(t *testing.T) { } func TestAttributeLabelMatcherFilters(t *testing.T) { - s := fake.New(20) + s := fake.NewLogStore(20) q := mustParse(t, `{resource_name="gateway-us-east"}`) rows := drain(t, s, query(q.Matchers, nil, 100, storage.DirectionBackward)) @@ -155,7 +155,7 @@ func TestAttributeLabelMatcherFilters(t *testing.T) { } func TestLineFilterAppliesBeforeLimit(t *testing.T) { - s := fake.New(20) + s := fake.NewLogStore(20) q := mustParse(t, `{service_name="envoy-gateway"} |= "GET"`) rows := drain(t, s, query(q.Matchers, q.Filters, 20, storage.DirectionBackward)) @@ -170,7 +170,7 @@ func TestLineFilterAppliesBeforeLimit(t *testing.T) { } func TestNoMatchesIsEmptyNotError(t *testing.T) { - s := fake.New(2) + s := fake.NewLogStore(2) q := mustParse(t, `{service_name="does-not-exist"}`) rows := drain(t, s, query(q.Matchers, nil, 100, storage.DirectionBackward)) if len(rows) != 0 { @@ -179,7 +179,7 @@ func TestNoMatchesIsEmptyNotError(t *testing.T) { } func TestDiscovery(t *testing.T) { - s := fake.New(2) + s := fake.NewLogStore(2) names, err := s.LabelNames(testCtx(), testRange()) if err != nil { @@ -222,7 +222,7 @@ func TestDiscovery(t *testing.T) { } func TestRejectsNonPositiveLimit(t *testing.T) { - s := fake.New(2) + s := fake.NewLogStore(2) for _, limit := range []int{0, -1} { if _, err := s.QueryLogs(testCtx(), query(nil, nil, limit, storage.DirectionBackward)); !errors.Is(err, storage.ErrInvalidLimit) { t.Errorf("QueryLogs with limit %d = %v, want ErrInvalidLimit", limit, err) @@ -231,7 +231,7 @@ func TestRejectsNonPositiveLimit(t *testing.T) { } func TestPing(t *testing.T) { - if err := fake.New(2).Ping(context.Background()); err != nil { + if err := fake.NewLogStore(2).Ping(context.Background()); err != nil { t.Fatalf("Ping: %v", err) } } @@ -243,11 +243,11 @@ func TestPing(t *testing.T) { // then read every one of them from a single map, and it advertised dotted OTel // keys the LogQL parser rejects. func TestLabelNamesAreAllResolvable(t *testing.T) { - store := fake.New(2) + logStore := fake.NewLogStore(2) ctx := miloauth.WithProject(context.Background(), "proj-a") ran := storage.TimeRange{Start: time.Now().Add(-time.Hour), End: time.Now()} - names, err := store.LabelNames(ctx, ran) + names, err := logStore.LabelNames(ctx, ran) if err != nil { t.Fatalf("LabelNames: %v", err) } @@ -262,7 +262,7 @@ func TestLabelNamesAreAllResolvable(t *testing.T) { if _, err := logql.Parse("{" + name + `="x"}`); err != nil { t.Errorf("advertised name %q is not a valid LogQL label: %v", name, err) } - values, err := store.LabelValues(ctx, name, ran) + values, err := logStore.LabelValues(ctx, name, ran) if err != nil { t.Fatalf("LabelValues(%q): %v", name, err) } diff --git a/queryapi/internal/storage/fake/metrics.go b/queryapi/internal/storage/fake/metrics.go new file mode 100644 index 0000000..707b79a --- /dev/null +++ b/queryapi/internal/storage/fake/metrics.go @@ -0,0 +1,92 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package fake + +import ( + "context" + "encoding/json" + "time" + + "go.datum.net/o11y/queryapi/internal/miloauth" + "go.datum.net/o11y/queryapi/internal/storage" +) + +// FakeMetric is the one series the fake metric store reports. +const FakeMetric = "fake_up" + +const projectLabel = "resourcemanager_datumapis_com_project_name" + +// MetricStore is a synthetic storage.MetricStore. It does not evaluate PromQL: +// every query answers with FakeMetric for the caller's project, so handler +// and e2e tests get a well-formed Prometheus response. +type MetricStore struct{} + +// NewMetricStore returns a fake metric store. +func NewMetricStore() *MetricStore { return &MetricStore{} } + +var _ storage.MetricStore = (*MetricStore)(nil) + +func (*MetricStore) labels(ctx context.Context) (map[string]string, error) { + project, ok := miloauth.ProjectID(ctx) + if !ok { + return nil, storage.ErrNoProject + } + return map[string]string{"__name__": FakeMetric, projectLabel: project}, nil +} + +func sample(t time.Time) []any { return []any{float64(t.UnixMilli()) / 1000, "1"} } + +func (m *MetricStore) Query(ctx context.Context, q storage.InstantQuery) (json.RawMessage, error) { + ls, err := m.labels(ctx) + if err != nil { + return nil, err + } + return json.Marshal(map[string]any{ + "resultType": "vector", + "result": []any{map[string]any{"metric": ls, "value": sample(q.Time)}}, + }) +} + +func (m *MetricStore) QueryRange(ctx context.Context, q storage.RangeQuery) (json.RawMessage, error) { + ls, err := m.labels(ctx) + if err != nil { + return nil, err + } + values := [][]any{} + for t := q.Range.Start; !t.After(q.Range.End); t = t.Add(q.Step) { + values = append(values, sample(t)) + } + return json.Marshal(map[string]any{ + "resultType": "matrix", + "result": []any{map[string]any{"metric": ls, "values": values}}, + }) +} + +func (m *MetricStore) Series(ctx context.Context, _ []string, _ storage.TimeRange) (json.RawMessage, error) { + ls, err := m.labels(ctx) + if err != nil { + return nil, err + } + return json.Marshal([]map[string]string{ls}) +} + +func (m *MetricStore) LabelNames(ctx context.Context, _ []string, _ storage.TimeRange) (json.RawMessage, error) { + if _, err := m.labels(ctx); err != nil { + return nil, err + } + return json.Marshal([]string{"__name__", projectLabel}) +} + +func (m *MetricStore) LabelValues(ctx context.Context, name string, _ []string, _ storage.TimeRange) (json.RawMessage, error) { + ls, err := m.labels(ctx) + if err != nil { + return nil, err + } + values := []string{} + if v, ok := ls[name]; ok { + values = append(values, v) + } + return json.Marshal(values) +} + +func (*MetricStore) Ping(context.Context) error { return nil } diff --git a/queryapi/internal/storage/fake/metrics_test.go b/queryapi/internal/storage/fake/metrics_test.go new file mode 100644 index 0000000..1478233 --- /dev/null +++ b/queryapi/internal/storage/fake/metrics_test.go @@ -0,0 +1,100 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package fake_test + +import ( + "context" + "encoding/json" + "errors" + "slices" + "testing" + "time" + + "go.datum.net/o11y/queryapi/internal/miloauth" + "go.datum.net/o11y/queryapi/internal/storage" + "go.datum.net/o11y/queryapi/internal/storage/fake" +) + +func TestMetricsRequireAProject(t *testing.T) { + m := fake.NewMetricStore() + ctx := context.Background() + tr := storage.TimeRange{Start: time.Unix(0, 0), End: time.Unix(60, 0)} + calls := map[string]func() error{ + "query": func() error { _, err := m.Query(ctx, storage.InstantQuery{Query: "up"}); return err }, + "query_range": func() error { + _, err := m.QueryRange(ctx, storage.RangeQuery{Query: "up", Range: tr, Step: time.Second}) + return err + }, + "series": func() error { _, err := m.Series(ctx, []string{"up"}, tr); return err }, + "labels": func() error { _, err := m.LabelNames(ctx, nil, tr); return err }, + "label values": func() error { _, err := m.LabelValues(ctx, "__name__", nil, tr); return err }, + } + for name, call := range calls { + if err := call(); !errors.Is(err, storage.ErrNoProject) { + t.Errorf("%s: err = %v, want ErrNoProject", name, err) + } + } +} + +func TestMetricsAreScopedToTheCallersProject(t *testing.T) { + m := fake.NewMetricStore() + ctx := miloauth.WithProject(context.Background(), "proj-a") + + raw, err := m.Query(ctx, storage.InstantQuery{Query: "up", Time: time.Unix(1760400000, 0)}) + if err != nil { + t.Fatalf("Query: %v", err) + } + var vec struct { + ResultType string `json:"resultType"` + Result []struct { + Metric map[string]string `json:"metric"` + Value []any `json:"value"` + } `json:"result"` + } + if err := json.Unmarshal(raw, &vec); err != nil { + t.Fatalf("decode: %v", err) + } + if vec.ResultType != "vector" || len(vec.Result) != 1 { + t.Fatalf("got %s with %d series, want one vector series", vec.ResultType, len(vec.Result)) + } + if got := vec.Result[0].Metric["resourcemanager_datumapis_com_project_name"]; got != "proj-a" { + t.Errorf("project label = %q, want proj-a", got) + } + + values, err := m.LabelValues(ctx, "resourcemanager_datumapis_com_project_name", nil, storage.TimeRange{}) + if err != nil { + t.Fatalf("LabelValues: %v", err) + } + var got []string + if err := json.Unmarshal(values, &got); err != nil { + t.Fatalf("decode: %v", err) + } + if !slices.Equal(got, []string{"proj-a"}) { + t.Errorf("project label values = %v, want [proj-a]", got) + } +} + +func TestMetricsRangeHasOnePointPerStep(t *testing.T) { + m := fake.NewMetricStore() + ctx := miloauth.WithProject(context.Background(), "proj-a") + raw, err := m.QueryRange(ctx, storage.RangeQuery{ + Query: "up", + Range: storage.TimeRange{Start: time.Unix(0, 0), End: time.Unix(600, 0)}, + Step: time.Minute, + }) + if err != nil { + t.Fatalf("QueryRange: %v", err) + } + var mat struct { + ResultType string `json:"resultType"` + Result []struct { + Values [][]any `json:"values"` + } `json:"result"` + } + if err := json.Unmarshal(raw, &mat); err != nil { + t.Fatalf("decode: %v", err) + } + if mat.ResultType != "matrix" || len(mat.Result) != 1 || len(mat.Result[0].Values) != 11 { + t.Fatalf("got %+v, want one matrix series of 11 points", mat) + } +} diff --git a/queryapi/internal/storage/metrics.go b/queryapi/internal/storage/metrics.go new file mode 100644 index 0000000..c056d2c --- /dev/null +++ b/queryapi/internal/storage/metrics.go @@ -0,0 +1,62 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package storage + +import ( + "context" + "encoding/json" + "errors" + "regexp" + "time" +) + +var ( + // ErrBackendUnavailable is a metrics backend that could not be reached or + // answered with something other than a Prometheus API response. + ErrBackendUnavailable = errors.New("storage: metrics backend unavailable") + + // ErrResponseTooLarge is a backend response over the configured size limit. + ErrResponseTooLarge = errors.New("storage: backend response exceeds the size limit") +) + +// QueryError is a query the backend rejected. Message is the backend's own +// text and is safe to return to the caller. +type QueryError struct{ Message string } + +func (e *QueryError) Error() string { return "storage: query rejected: " + e.Message } + +// InstantQuery is a PromQL query evaluated at one instant. +type InstantQuery struct { + Query string + Time time.Time + Timeout time.Duration // 0: the backend's default +} + +// RangeQuery is a PromQL query evaluated at each Step across Range. +type RangeQuery struct { + Query string + Range TimeRange + Step time.Duration + Timeout time.Duration +} + +// MetricStore is a backend for the Prometheus query API. +// +// Each method returns the "data" member of the backend's success envelope +// unchanged; handlers add the envelope. Implementations confine every call to +// the project on ctx and return ErrNoProject without contacting the backend +// when there is none. A zero Start or End in a TimeRange is left to the +// backend's default. +type MetricStore interface { + Query(ctx context.Context, q InstantQuery) (json.RawMessage, error) + QueryRange(ctx context.Context, q RangeQuery) (json.RawMessage, error) + Series(ctx context.Context, matchers []string, tr TimeRange) (json.RawMessage, error) + LabelNames(ctx context.Context, matchers []string, tr TimeRange) (json.RawMessage, error) + LabelValues(ctx context.Context, name string, matchers []string, tr TimeRange) (json.RawMessage, error) + Ping(ctx context.Context) error +} + +var labelNameRE = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`) + +// ValidLabelName reports whether name is a Prometheus label name. +func ValidLabelName(name string) bool { return labelNameRE.MatchString(name) } diff --git a/queryapi/internal/storage/prometheus/integration_test.go b/queryapi/internal/storage/prometheus/integration_test.go new file mode 100644 index 0000000..3c7fc66 --- /dev/null +++ b/queryapi/internal/storage/prometheus/integration_test.go @@ -0,0 +1,249 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package prometheus_test + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "os" + "slices" + "strings" + "testing" + "time" + + "go.datum.net/o11y/queryapi/internal/miloauth" + "go.datum.net/o11y/queryapi/internal/storage" + "go.datum.net/o11y/queryapi/internal/storage/prometheus" +) + +const itMetric = "queryapi_it_requests_total" + +// seed imports a counter for proj-a (+1 per 15s, label only_a) and proj-b +// (+100 per 15s, label only_b) over the last ten minutes, then waits until VM +// serves both. +func seed(t *testing.T, base string, now time.Time) { + t.Helper() + var b strings.Builder + for i := 0; i <= 40; i++ { + ts := now.Add(-10 * time.Minute).Add(time.Duration(i) * 15 * time.Second).UnixMilli() + fmt.Fprintf(&b, "%s{%s=\"proj-a\",only_a=\"x\"} %d %d\n", itMetric, label, i, ts) + fmt.Fprintf(&b, "%s{%s=\"proj-b\",only_b=\"y\"} %d %d\n", itMetric, label, i*100, ts) + } + resp, err := http.Post(base+"/api/v1/import/prometheus", "text/plain", strings.NewReader(b.String())) + if err != nil { + t.Fatalf("import: %v", err) + } + if err := resp.Body.Close(); err != nil || resp.StatusCode/100 != 2 { + t.Fatalf("import: status %d, close %v", resp.StatusCode, err) + } + for range 60 { + if resp, err := http.Get(base + "/internal/force_flush"); err == nil { + _ = resp.Body.Close() + } + resp, err := http.Get(base + "/api/v1/series?match[]=" + itMetric) + if err == nil { + var r struct{ Data []map[string]string } + decodeErr := json.NewDecoder(resp.Body).Decode(&r) + _ = resp.Body.Close() + if decodeErr == nil && len(r.Data) == 2 { + return + } + } + time.Sleep(500 * time.Millisecond) + } + t.Fatalf("seeded series never became visible") +} + +type result struct { + ResultType string `json:"resultType"` + Result json.RawMessage `json:"result"` +} + +func series(t *testing.T, raw json.RawMessage) []map[string]string { + t.Helper() + var r result + if err := json.Unmarshal(raw, &r); err != nil { + t.Fatalf("decode %s: %v", raw, err) + } + var out []struct { + Metric map[string]string `json:"metric"` + Value []any `json:"value"` + } + if err := json.Unmarshal(r.Result, &out); err != nil { + t.Fatalf("decode result %s: %v", r.Result, err) + } + var ls []map[string]string + for _, s := range out { + ls = append(ls, s.Metric) + } + return ls +} + +func projects(ls []map[string]string) []string { + var ps []string + for _, l := range ls { + ps = append(ps, l[label]) + } + slices.Sort(ps) + return slices.Compact(ps) +} + +func TestStoreAgainstVictoriaMetrics(t *testing.T) { + base := os.Getenv("PROMETHEUS_TEST_URL") + if base == "" { + t.Skip("PROMETHEUS_TEST_URL not set; run `task test-integration-prometheus`") + } + now := time.Now().UTC().Truncate(time.Second) + seed(t, base, now) + + s, err := prometheus.New(prometheus.Config{ + URL: base, Tenancy: prometheus.TenancyExtraLabel, ProjectLabel: label, MaxResponseBytes: 32 << 20, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + if err := s.Ping(context.Background()); err != nil { + t.Fatalf("Ping: %v", err) + } + ctx := miloauth.WithProject(context.Background(), "proj-a") + tr := storage.TimeRange{Start: now.Add(-15 * time.Minute), End: now} + + query := func(q string) json.RawMessage { + t.Helper() + raw, err := s.Query(ctx, storage.InstantQuery{Query: q, Time: now}) + if err != nil { + t.Fatalf("Query(%s): %v", q, err) + } + return raw + } + + // Positive: proj-a sees its own series. + if got := projects(series(t, query(itMetric))); !slices.Equal(got, []string{"proj-a"}) { + t.Errorf("own series: projects = %v, want [proj-a]", got) + } + + // Negative: naming proj-b, or selecting everything, yields nothing of + // proj-b's. + if got := series(t, query(itMetric+`{`+label+`="proj-b"}`)); len(got) != 0 { + t.Errorf("explicit proj-b selector returned %v", got) + } + if got := projects(series(t, query(`{__name__=~"queryapi_it_.+"}`))); !slices.Equal(got, []string{"proj-a"}) { + t.Errorf("match-all selector: projects = %v, want [proj-a]", got) + } + + // An aggregate sees only proj-a's value (40), not the sum with proj-b's + // (4040). + var sum []struct { + Value []any `json:"value"` + } + var r result + if err := json.Unmarshal(query("sum("+itMetric+")"), &r); err != nil { + t.Fatal(err) + } + if err := json.Unmarshal(r.Result, &sum); err != nil || len(sum) != 1 || sum[0].Value[1] != "40" { + t.Errorf("sum = %+v (%v), want 40", sum, err) + } + + // The portal's reset-guarded increase parses and evaluates. + portal := fmt.Sprintf("sum(increase(%[1]s[5m]) * (resets(%[1]s[5m]) == bool 0))", itMetric) + if err := json.Unmarshal(query(portal), &r); err != nil || r.ResultType != "vector" { + t.Errorf("portal query: resultType %q, err %v", r.ResultType, err) + } + + // Grafana's health probe evaluates. VictoriaMetrics + // answers a scalar expression as a vector, not a scalar. + if err := json.Unmarshal(query("1+1"), &r); err != nil || r.ResultType != "vector" { + t.Errorf("1+1: resultType %q, err %v; want vector", r.ResultType, err) + } + + // Range query: positive and negative. + raw, err := s.QueryRange(ctx, storage.RangeQuery{Query: `{__name__=~"queryapi_it_.+"}`, Range: tr, Step: 30 * time.Second}) + if err != nil { + t.Fatalf("QueryRange: %v", err) + } + var mat struct { + Result []struct { + Metric map[string]string `json:"metric"` + } `json:"result"` + } + if err := json.Unmarshal(raw, &mat); err != nil || len(mat.Result) != 1 || mat.Result[0].Metric[label] != "proj-a" { + t.Errorf("range: %s (%v), want exactly proj-a's series", raw, err) + } + + // Cache carry-over: with VictoriaMetrics' rollup cache on, a range + // result computed for proj-b must not be served to proj-a. + rangeFor := func(project, q string) string { + t.Helper() + raw, err := s.QueryRange(miloauth.WithProject(context.Background(), project), + storage.RangeQuery{Query: q, Range: tr, Step: 30 * time.Second}) + if err != nil { + t.Fatalf("QueryRange(%s, %s): %v", project, q, err) + } + return string(raw) + } + for _, q := range []string{`{__name__=~"queryapi_it_.+"}`, "sum(" + itMetric + ")"} { + b1 := rangeFor("proj-b", q) + a1 := rangeFor("proj-a", q) + b2 := rangeFor("proj-b", q) + if !strings.Contains(b1, "proj-b") && !strings.Contains(b1, `"4000"`) { + t.Errorf("%s: proj-b result has no proj-b data (vacuous check): %s", q, b1) + } + if strings.Contains(a1, "proj-b") || strings.Contains(a1, "only_b") || strings.Contains(a1, `"4000"`) { + t.Errorf("%s: proj-a result carries proj-b data: %s", q, a1) + } + if strings.Contains(b2, "proj-a") || strings.Contains(b2, "only_a") { + t.Errorf("%s: proj-b result carries proj-a data: %s", q, b2) + } + } + + // Series. + raw, err = s.Series(ctx, []string{`{__name__=~".+"}`}, tr) + if err != nil { + t.Fatalf("Series: %v", err) + } + var sl []map[string]string + if err := json.Unmarshal(raw, &sl); err != nil { + t.Fatal(err) + } + if got := projects(sl); !slices.Equal(got, []string{"proj-a"}) { + t.Errorf("series: projects = %v, want [proj-a]", got) + } + + // Label names: only_a visible, only_b not. + raw, err = s.LabelNames(ctx, nil, tr) + if err != nil { + t.Fatalf("LabelNames: %v", err) + } + var names []string + if err := json.Unmarshal(raw, &names); err != nil { + t.Fatal(err) + } + if !slices.Contains(names, "only_a") || slices.Contains(names, "only_b") { + t.Errorf("label names = %v, want only_a and not only_b", names) + } + + // Label values. + values := func(name string) []string { + t.Helper() + raw, err := s.LabelValues(ctx, name, nil, tr) + if err != nil { + t.Fatalf("LabelValues(%s): %v", name, err) + } + var vs []string + if err := json.Unmarshal(raw, &vs); err != nil { + t.Fatal(err) + } + return vs + } + if got := values(label); !slices.Equal(got, []string{"proj-a"}) { + t.Errorf("project label values = %v, want [proj-a]", got) + } + if got := values("only_b"); len(got) != 0 { + t.Errorf("only_b values = %v, want none", got) + } + if got := values("only_a"); !slices.Equal(got, []string{"x"}) { + t.Errorf("only_a values = %v, want [x]", got) + } +} diff --git a/queryapi/internal/storage/prometheus/metrics.go b/queryapi/internal/storage/prometheus/metrics.go new file mode 100644 index 0000000..45f281f --- /dev/null +++ b/queryapi/internal/storage/prometheus/metrics.go @@ -0,0 +1,25 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package prometheus + +import ( + prom "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + "k8s.io/component-base/metrics/legacyregistry" +) + +// The apiserver serves /metrics from component-base's registry. +var registry = promauto.With(legacyregistry.Registerer()) + +var ( + backendRequests = registry.NewCounterVec(prom.CounterOpts{ + Name: "queryapi_metrics_backend_requests_total", + Help: "Requests sent to the metrics backend, by API endpoint and status code (\"error\" when no response arrived).", + }, []string{"endpoint", "code"}) + + backendDuration = registry.NewHistogramVec(prom.HistogramOpts{ + Name: "queryapi_metrics_backend_request_duration_seconds", + Help: "Metrics backend request latency in seconds, by API endpoint.", + Buckets: prom.DefBuckets, + }, []string{"endpoint"}) +) diff --git a/queryapi/internal/storage/prometheus/prometheus.go b/queryapi/internal/storage/prometheus/prometheus.go new file mode 100644 index 0000000..aadb611 --- /dev/null +++ b/queryapi/internal/storage/prometheus/prometheus.go @@ -0,0 +1,236 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +// Package prometheus is a storage.MetricStore for any server implementing the +// Prometheus query HTTP API. Tenancy is a separate, explicit choice: see +// Tenancy. +package prometheus + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + "go.datum.net/o11y/queryapi/internal/miloauth" + "go.datum.net/o11y/queryapi/internal/storage" +) + +// Tenancy is how the backend is made to confine a query to one project. +type Tenancy string + +// TenancyExtraLabel sends VictoriaMetrics's extra_label parameter. Backends +// that don't implement it, the Prometheus server among them, ignore it and +// apply no isolation, so this mode requires VictoriaMetrics or a server that +// honors extra_label. +const TenancyExtraLabel Tenancy = "extra-label" + +// ParseTenancy validates a --prometheus-tenancy value. +func ParseTenancy(s string) (Tenancy, error) { + switch Tenancy(s) { + case TenancyExtraLabel: + return TenancyExtraLabel, nil + case "": + return "", errors.New("prometheus tenancy is required; the only mode is extra-label") + default: + return "", fmt.Errorf("unknown prometheus tenancy %q; the only mode is extra-label", s) + } +} + +// Config configures a Store. +type Config struct { + URL string + Tenancy Tenancy + ProjectLabel string + MaxResponseBytes int64 + Client *http.Client +} + +// Store is a storage.MetricStore over the Prometheus query API. +type Store struct { + base *url.URL + cfg Config + client *http.Client +} + +var _ storage.MetricStore = (*Store)(nil) + +// New validates cfg and returns a Store. +func New(cfg Config) (*Store, error) { + if _, err := ParseTenancy(string(cfg.Tenancy)); err != nil { + return nil, err + } + base, err := url.Parse(cfg.URL) + if err != nil || (base.Scheme != "http" && base.Scheme != "https") || base.Host == "" { + return nil, fmt.Errorf("prometheus url %q must be an absolute http or https URL", cfg.URL) + } + if !storage.ValidLabelName(cfg.ProjectLabel) { + return nil, fmt.Errorf("prometheus project label %q is not a valid label name", cfg.ProjectLabel) + } + if cfg.MaxResponseBytes <= 0 { + return nil, fmt.Errorf("max response bytes must be greater than zero, got %d", cfg.MaxResponseBytes) + } + client := cfg.Client + if client == nil { + client = &http.Client{} + } + return &Store{base: base, cfg: cfg, client: client}, nil +} + +func (s *Store) Query(ctx context.Context, q storage.InstantQuery) (json.RawMessage, error) { + form := url.Values{"query": {q.Query}, "time": {formatTime(q.Time)}} + if q.Timeout > 0 { + form.Set("timeout", formatSeconds(q.Timeout)) + } + return s.do(ctx, "/api/v1/query", "/api/v1/query", form) +} + +func (s *Store) QueryRange(ctx context.Context, q storage.RangeQuery) (json.RawMessage, error) { + form := url.Values{ + "query": {q.Query}, + "start": {formatTime(q.Range.Start)}, + "end": {formatTime(q.Range.End)}, + "step": {formatSeconds(q.Step)}, + } + if q.Timeout > 0 { + form.Set("timeout", formatSeconds(q.Timeout)) + } + return s.do(ctx, "/api/v1/query_range", "/api/v1/query_range", form) +} + +func (s *Store) Series(ctx context.Context, matchers []string, tr storage.TimeRange) (json.RawMessage, error) { + return s.do(ctx, "/api/v1/series", "/api/v1/series", selectorForm(matchers, tr)) +} + +func (s *Store) LabelNames(ctx context.Context, matchers []string, tr storage.TimeRange) (json.RawMessage, error) { + return s.do(ctx, "/api/v1/labels", "/api/v1/labels", selectorForm(matchers, tr)) +} + +func (s *Store) LabelValues(ctx context.Context, name string, matchers []string, tr storage.TimeRange) (json.RawMessage, error) { + if !storage.ValidLabelName(name) { + return nil, &storage.QueryError{Message: "invalid label name " + strconv.Quote(name)} + } + // The metric label is the route, never the name, so it stays bounded. + return s.do(ctx, "/api/v1/label/"+name+"/values", "/api/v1/label/{name}/values", selectorForm(matchers, tr)) +} + +// Ping checks /-/healthy, which both Prometheus and VictoriaMetrics serve. +func (s *Store) Ping(ctx context.Context) (err error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, s.base.JoinPath("/-/healthy").String(), nil) + if err != nil { + return err + } + resp, err := s.client.Do(req) + if err != nil { + return fmt.Errorf("%w: %w", storage.ErrBackendUnavailable, err) + } + defer func() { err = errors.Join(err, resp.Body.Close()) }() + if resp.StatusCode != http.StatusOK { + return fmt.Errorf("%w: /-/healthy returned %d", storage.ErrBackendUnavailable, resp.StatusCode) + } + return nil +} + +func selectorForm(matchers []string, tr storage.TimeRange) url.Values { + form := url.Values{} + for _, m := range matchers { + form.Add("match[]", m) + } + if !tr.Start.IsZero() { + form.Set("start", formatTime(tr.Start)) + } + if !tr.End.IsZero() { + form.Set("end", formatTime(tr.End)) + } + return form +} + +// do sends form to path, confined to the project on ctx. route is the metric +// label for path. +func (s *Store) do(ctx context.Context, path, route string, form url.Values) (data json.RawMessage, err error) { + project, ok := miloauth.ProjectID(ctx) + if !ok { + return nil, storage.ErrNoProject + } + // Set, not Add: form is built from typed fields, so this is the only + // extra_label it can carry. + form.Set("extra_label", s.cfg.ProjectLabel+"="+project) + + req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.base.JoinPath(path).String(), + strings.NewReader(form.Encode())) + if err != nil { + return nil, err + } + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + + start := time.Now() + resp, err := s.client.Do(req) + backendDuration.WithLabelValues(route).Observe(time.Since(start).Seconds()) + if err != nil { + backendRequests.WithLabelValues(route, "error").Inc() + if ctxErr := ctx.Err(); ctxErr != nil { + return nil, ctxErr + } + return nil, fmt.Errorf("%w: %w", storage.ErrBackendUnavailable, err) + } + defer func() { err = errors.Join(err, resp.Body.Close()) }() + backendRequests.WithLabelValues(route, strconv.Itoa(resp.StatusCode)).Inc() + + body, err := io.ReadAll(io.LimitReader(resp.Body, s.cfg.MaxResponseBytes+1)) + if err != nil { + if ctxErr := ctx.Err(); ctxErr != nil { + return nil, ctxErr + } + return nil, fmt.Errorf("%w: read response: %w", storage.ErrBackendUnavailable, err) + } + if int64(len(body)) > s.cfg.MaxResponseBytes { + return nil, storage.ErrResponseTooLarge + } + data, err = decode(resp.StatusCode, body) + if err != nil { + if ctxErr := ctx.Err(); ctxErr != nil { + return nil, ctxErr + } + } + return data, err +} + +type envelope struct { + Status string `json:"status"` + Data json.RawMessage `json:"data"` + ErrorType string `json:"errorType"` + Error string `json:"error"` +} + +// decode trusts a body only if it is a Prometheus API envelope. Anything else, +// such as a proxy's HTML, means the backend is unavailable. +func decode(status int, body []byte) (json.RawMessage, error) { + var env envelope + isEnvelope := json.Unmarshal(body, &env) == nil && env.Status != "" + switch { + case isEnvelope && (status == http.StatusUnprocessableEntity || status == http.StatusServiceUnavailable) && isTimeout(env): + return nil, fmt.Errorf("%w: %s", context.DeadlineExceeded, env.Error) + case status == http.StatusOK && isEnvelope && env.Status == "success" && len(env.Data) > 0: + return env.Data, nil + case (status == http.StatusBadRequest || status == http.StatusUnprocessableEntity) && isEnvelope && env.Error != "": + return nil, &storage.QueryError{Message: env.Error} + default: + return nil, fmt.Errorf("%w: status %d: %.512s", storage.ErrBackendUnavailable, status, body) + } +} + +// isTimeout recognises a backend query timeout. Prometheus sends errorType +// "timeout"; VictoriaMetrics sends the HTTP status as errorType (observed +// 422) and says "timeout exceeded during query execution". +func isTimeout(env envelope) bool { + return env.ErrorType == "timeout" || strings.Contains(env.Error, "timeout exceeded") +} + +func formatTime(t time.Time) string { return t.UTC().Format(time.RFC3339Nano) } + +func formatSeconds(d time.Duration) string { return strconv.FormatFloat(d.Seconds(), 'f', -1, 64) } diff --git a/queryapi/internal/storage/prometheus/prometheus_test.go b/queryapi/internal/storage/prometheus/prometheus_test.go new file mode 100644 index 0000000..c95d928 --- /dev/null +++ b/queryapi/internal/storage/prometheus/prometheus_test.go @@ -0,0 +1,279 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package prometheus_test + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "net/url" + "slices" + "strings" + "sync" + "testing" + "time" + + "go.datum.net/o11y/queryapi/internal/miloauth" + "go.datum.net/o11y/queryapi/internal/storage" + "go.datum.net/o11y/queryapi/internal/storage/prometheus" +) + +const label = "resourcemanager_datumapis_com_project_name" + +type seen struct { + method, path string + form url.Values +} + +// backend answers every request with status and body, recording each one. +func backend(t *testing.T, status int, body string) (*httptest.Server, func() []seen) { + t.Helper() + var mu sync.Mutex + var reqs []seen + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if err := r.ParseForm(); err != nil { + t.Errorf("backend parse form: %v", err) + } + mu.Lock() + reqs = append(reqs, seen{r.Method, r.URL.Path, r.PostForm}) + mu.Unlock() + w.WriteHeader(status) + _, _ = io.WriteString(w, body) + })) + t.Cleanup(srv.Close) + return srv, func() []seen { mu.Lock(); defer mu.Unlock(); return slices.Clone(reqs) } +} + +func newStore(t *testing.T, url string) *prometheus.Store { + t.Helper() + s, err := prometheus.New(prometheus.Config{ + URL: url, Tenancy: prometheus.TenancyExtraLabel, ProjectLabel: label, MaxResponseBytes: 1 << 20, + }) + if err != nil { + t.Fatalf("New: %v", err) + } + return s +} + +var projCtx = miloauth.WithProject(context.Background(), "proj-a") + +const ok = `{"status":"success","data":{"resultType":"vector","result":[]}}` + +func keys(v url.Values) []string { + var ks []string + for k := range v { + ks = append(ks, k) + } + slices.Sort(ks) + return ks +} + +// Every call carries exactly one extra_label for the caller's project, and +// only the parameters the store built itself. +func TestEveryCallIsConfinedToTheProject(t *testing.T) { + srv, reqs := backend(t, http.StatusOK, ok) + s := newStore(t, srv.URL) + tr := storage.TimeRange{Start: time.Unix(0, 0), End: time.Unix(60, 0)} + + // The query text tries to smuggle its own extra_label; it must arrive as + // part of the query value, not as a parameter. + smuggle := `up&extra_label=` + label + `=proj-b&extra_filters[]={x="y"}` + calls := []struct { + name, path string + keys []string + call func() error + }{ + {"query", "/api/v1/query", []string{"extra_label", "query", "time", "timeout"}, func() error { + _, err := s.Query(projCtx, storage.InstantQuery{Query: smuggle, Time: time.Unix(1, 0), Timeout: time.Second}) + return err + }}, + {"query_range", "/api/v1/query_range", []string{"end", "extra_label", "query", "start", "step"}, func() error { + _, err := s.QueryRange(projCtx, storage.RangeQuery{Query: smuggle, Range: tr, Step: time.Second}) + return err + }}, + {"series", "/api/v1/series", []string{"end", "extra_label", "match[]", "start"}, func() error { + _, err := s.Series(projCtx, []string{"up"}, tr) + return err + }}, + {"labels", "/api/v1/labels", []string{"extra_label"}, func() error { + _, err := s.LabelNames(projCtx, nil, storage.TimeRange{}) + return err + }}, + {"label values", "/api/v1/label/__name__/values", []string{"extra_label", "match[]"}, func() error { + _, err := s.LabelValues(projCtx, "__name__", []string{"up"}, storage.TimeRange{}) + return err + }}, + } + for i, c := range calls { + if err := c.call(); err != nil { + t.Fatalf("%s: %v", c.name, err) + } + got := reqs()[i] + if got.method != http.MethodPost || got.path != c.path { + t.Errorf("%s: %s %s, want POST %s", c.name, got.method, got.path, c.path) + } + if el := got.form["extra_label"]; len(el) != 1 || el[0] != label+"=proj-a" { + t.Errorf("%s: extra_label = %q, want exactly [%s=proj-a]", c.name, el, label) + } + if k := keys(got.form); !slices.Equal(k, c.keys) { + t.Errorf("%s: parameters = %v, want %v", c.name, k, c.keys) + } + if strings.HasPrefix(c.name, "query") && got.form.Get("query") != smuggle { + t.Errorf("%s: query = %q, want the smuggled text verbatim", c.name, got.form.Get("query")) + } + } +} + +func TestNoProjectSendsNothing(t *testing.T) { + srv, reqs := backend(t, http.StatusOK, ok) + s := newStore(t, srv.URL) + if _, err := s.Query(context.Background(), storage.InstantQuery{Query: "up"}); !errors.Is(err, storage.ErrNoProject) { + t.Fatalf("err = %v, want ErrNoProject", err) + } + if n := len(reqs()); n != 0 { + t.Fatalf("backend received %d requests, want 0", n) + } +} + +func TestBackendResponses(t *testing.T) { + cases := []struct { + name string + status int + body string + check func(error) bool + }{ + {"success", 200, ok, func(err error) bool { return err == nil }}, + {"bad query (422)", 422, `{"status":"error","errorType":"422","error":"unknown func foo"}`, func(err error) bool { + var qe *storage.QueryError + return errors.As(err, &qe) && qe.Message == "unknown func foo" + }}, + {"bad query (400)", 400, `{"status":"error","errorType":"bad_data","error":"parse error"}`, func(err error) bool { + var qe *storage.QueryError + return errors.As(err, &qe) + }}, + // A proxy's HTML is not a query error. + {"html 400", 400, `bad request`, func(err error) bool { return errors.Is(err, storage.ErrBackendUnavailable) }}, + {"html 502", 502, `bad gateway`, func(err error) bool { return errors.Is(err, storage.ErrBackendUnavailable) }}, + // Observed from victoria-metrics v1.114.0 with timeout=1ms. + {"timeout (VictoriaMetrics, real)", 422, `{"status":"error","errorType":"422","error":"timeout exceeded during query execution: 0.001 seconds (elapsed 1.128 seconds); the timeout can be adjusted with ` + "`-search.maxQueryDuration`" + ` command-line flag"}`, func(err error) bool { + var qe *storage.QueryError + return errors.Is(err, context.DeadlineExceeded) && !errors.As(err, &qe) + }}, + {"timeout (503, errorType 503)", 503, `{"status":"error","errorType":"503","error":"timeout exceeded during query execution: 30 seconds"}`, func(err error) bool { + return errors.Is(err, context.DeadlineExceeded) + }}, + {"503 not a timeout", 503, `{"status":"error","errorType":"503","error":"couldn't start executing the request in 10s: too many concurrent requests"}`, func(err error) bool { + return errors.Is(err, storage.ErrBackendUnavailable) && !errors.Is(err, context.DeadlineExceeded) + }}, + {"timeout", 503, `{"status":"error","errorType":"timeout","error":"query timed out"}`, func(err error) bool { + return errors.Is(err, context.DeadlineExceeded) + }}, + {"200 error envelope", 200, `{"status":"error","error":"?"}`, func(err error) bool { return errors.Is(err, storage.ErrBackendUnavailable) }}, + {"too large", 200, `{"status":"success","data":"` + strings.Repeat("x", 1<<20) + `"}`, func(err error) bool { + return errors.Is(err, storage.ErrResponseTooLarge) + }}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + srv, _ := backend(t, tc.status, tc.body) + _, err := newStore(t, srv.URL).Query(projCtx, storage.InstantQuery{Query: "up"}) + if !tc.check(err) { + t.Errorf("err = %v", err) + } + }) + } +} + +func TestUnreachableAndDeadline(t *testing.T) { + srv, _ := backend(t, 200, ok) + addr := srv.URL + srv.Close() + if _, err := newStore(t, addr).Query(projCtx, storage.InstantQuery{Query: "up"}); !errors.Is(err, storage.ErrBackendUnavailable) { + t.Errorf("unreachable: err = %v, want ErrBackendUnavailable", err) + } + + slow := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + // The server only notices a closed connection once the body is read. + _, _ = io.Copy(io.Discard, r.Body) + <-r.Context().Done() + })) + defer slow.Close() + ctx, cancel := context.WithTimeout(projCtx, 50*time.Millisecond) + defer cancel() + if _, err := newStore(t, slow.URL).Query(ctx, storage.InstantQuery{Query: "up"}); !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("deadline: err = %v, want DeadlineExceeded", err) + } +} + +// Long queries travel in the body intact. +func TestLongQueryArrivesIntact(t *testing.T) { + srv, reqs := backend(t, 200, ok) + q := "sum(" + strings.Repeat(`up{job="a"} + `, 1200) + "up)" + if len(q) < 16<<10 { + t.Fatalf("test query is only %d bytes", len(q)) + } + if _, err := newStore(t, srv.URL).Query(projCtx, storage.InstantQuery{Query: q}); err != nil { + t.Fatalf("Query: %v", err) + } + if got := reqs()[0].form.Get("query"); got != q { + t.Errorf("query arrived as %d bytes, want %d", len(got), len(q)) + } +} + +// A label name that isn't one never reaches the backend. +func TestInvalidLabelNameSendsNothing(t *testing.T) { + srv, reqs := backend(t, 200, ok) + _, err := newStore(t, srv.URL).LabelValues(projCtx, "a/b", nil, storage.TimeRange{}) + var qe *storage.QueryError + if !errors.As(err, &qe) || len(reqs()) != 0 { + t.Errorf("err = %v, requests = %d; want a QueryError and none", err, len(reqs())) + } +} + +func TestBaseURLPathIsKept(t *testing.T) { + srv, reqs := backend(t, 200, ok) + if _, err := newStore(t, srv.URL+"/select/0/prometheus").Query(projCtx, storage.InstantQuery{Query: "up"}); err != nil { + t.Fatal(err) + } + if p := reqs()[0].path; p != "/select/0/prometheus/api/v1/query" { + t.Errorf("path = %s", p) + } +} + +func TestNewRejectsBadConfig(t *testing.T) { + good := prometheus.Config{URL: "http://vm:8429", Tenancy: prometheus.TenancyExtraLabel, ProjectLabel: label, MaxResponseBytes: 1} + bad := map[string]func(*prometheus.Config){ + "no tenancy": func(c *prometheus.Config) { c.Tenancy = "" }, + "unknown tenancy": func(c *prometheus.Config) { c.Tenancy = "rewrite" }, + "no url": func(c *prometheus.Config) { c.URL = "" }, + "relative url": func(c *prometheus.Config) { c.URL = "vm:8429" }, + "bad label": func(c *prometheus.Config) { c.ProjectLabel = "project-name" }, + "zero size limit": func(c *prometheus.Config) { c.MaxResponseBytes = 0 }, + } + if _, err := prometheus.New(good); err != nil { + t.Fatalf("good config: %v", err) + } + for name, mutate := range bad { + c := good + mutate(&c) + if _, err := prometheus.New(c); err == nil { + t.Errorf("%s: New succeeded, want an error", name) + } + } +} + +func TestPing(t *testing.T) { + healthy, reqs := backend(t, 200, "OK") + if err := newStore(t, healthy.URL).Ping(context.Background()); err != nil { + t.Errorf("healthy: %v", err) + } + if r := reqs(); r[0].method != http.MethodGet || r[0].path != "/-/healthy" { + t.Errorf("ping sent %s %s", r[0].method, r[0].path) + } + sick, _ := backend(t, 500, "") + if err := newStore(t, sick.URL).Ping(context.Background()); err == nil { + t.Error("unhealthy backend: Ping = nil") + } +} diff --git a/queryapi/openapi.yaml b/queryapi/openapi.yaml index ac1b178..b23e714 100644 --- a/queryapi/openapi.yaml +++ b/queryapi/openapi.yaml @@ -39,11 +39,10 @@ info: follow-up once the tail transport (chunked HTTP vs. WebSocket) is decided. - Metric endpoints are specced here so `datumctl`/portal and Prometheus-API - client work can start, but return 501 until implemented — logs are the - Phase 1 priority; see the roadmap in - enhancements/telemetry/README.md. - version: "0.3.0" + Metric endpoints accept PromQL (MetricsQL where the backend is + VictoriaMetrics), evaluated by the configured backend. queryapi confines + every metric query to the caller's project. + version: "0.4.0" servers: - url: "/projects/{projectId}/control-plane/apis/{o11yGroup}/v1alpha1" description: >- @@ -538,8 +537,9 @@ components: UnexpectedError: description: >- Any error status not explicitly listed above, including 501 Not - Implemented. The log endpoints are served from ClickHouse; the metric - endpoints are still stubs and answer 501 unconditionally. A log + Implemented. The log endpoints are served from ClickHouse; a metric + endpoint answers 422 when the backend response is too large, 502 when + the backend is unavailable, and 503 on timeout. A log endpoint returns 501 only for an operation the configured storage backend does not implement. content: @@ -678,9 +678,8 @@ components: query: type: string description: >- - PromQL-style label-matched series selector, e.g. - `cpu_usage{service="foo"}`. Aggregation functions (sum, avg, - rate) are not yet supported and are rejected with 400. + A PromQL expression. Aggregations and functions are evaluated by + the backend. example: 'cpu_usage{service="compute-workload"}' time: type: string @@ -694,9 +693,8 @@ components: query: type: string description: >- - PromQL-style label-matched series selector, e.g. - `cpu_usage{service="foo"}`. Aggregation functions (sum, avg, - rate) are not yet supported and are rejected with 400. + A PromQL expression. Aggregations and functions are evaluated by + the backend. example: 'cpu_usage{service="compute-workload"}' start: type: string @@ -707,25 +705,6 @@ components: timeout: type: string - PromVectorResult: - type: object - required: [metric, value] - properties: - metric: - type: object - additionalProperties: - type: string - description: Label set for this series, including __name__. - value: - type: array - description: >- - [unixTimestamp, stringValue] — a single sample, matching - Prometheus's "vector" result type for instant queries. - items: - type: string - minItems: 2 - maxItems: 2 - PromMatrixResult: type: object required: [metric, values] @@ -743,7 +722,9 @@ components: items: type: array items: - type: string + oneOf: + - type: number + - type: string minItems: 2 maxItems: 2 @@ -760,15 +741,9 @@ components: properties: resultType: type: string - enum: [vector] - description: >- - Always "vector" for the label-matched selector subset this - service accepts. Prometheus's "scalar"/"string" result types - don't apply to a plain series selector. + enum: [vector, matrix, scalar, string] result: - type: array - items: - $ref: "#/components/schemas/PromVectorResult" + description: "Shaped by resultType, as in the Prometheus HTTP API." PromRangeQueryResponse: type: object diff --git a/queryapi/options.go b/queryapi/options.go index a7eb303..3fcc14d 100644 --- a/queryapi/options.go +++ b/queryapi/options.go @@ -37,6 +37,11 @@ type Options struct { // Query is what the handlers need. Query Config + + // legacyStorage is the deprecated --storage, folded into + // Query.LogsStorage by Complete. + legacyStorage string + flags *pflag.FlagSet } // NewOptions returns Options with production-reasonable defaults. @@ -90,11 +95,25 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) { // names and meanings activity's deployment uses. o.Recommended.AddFlags(fs) - fs.StringVar(&o.Query.Storage, "storage", o.Query.Storage, "storage backend: fake or clickhouse") + fs.StringVar(&o.Query.LogsStorage, "logs-storage", o.Query.LogsStorage, "log storage backend: fake or clickhouse") + fs.StringVar(&o.Query.MetricsStorage, "metrics-storage", o.Query.MetricsStorage, "metric storage backend: fake or prometheus") + fs.StringVar(&o.Query.PrometheusURL, "prometheus-url", o.Query.PrometheusURL, + "base URL of the Prometheus-compatible query API") + fs.StringVar(&o.Query.PrometheusTenancy, "prometheus-tenancy", o.Query.PrometheusTenancy, + "how the backend confines a query to the caller's project; the only mode is extra-label, "+ + "which requires VictoriaMetrics or another server that honors extra_label") + fs.StringVar(&o.Query.PrometheusProjectLabel, "prometheus-project-label", o.Query.PrometheusProjectLabel, + "label the tenancy mode constrains to the caller's project") + fs.Int64Var(&o.Query.MaxResponseBytes, "max-response-bytes", o.Query.MaxResponseBytes, + "largest metric backend response accepted, in bytes") + fs.StringVar(&o.legacyStorage, "storage", "", "deprecated alias for --logs-storage") + _ = fs.MarkDeprecated("storage", "use --logs-storage; --storage is removed in the next release") + o.flags = fs fs.Float64Var(&o.Query.FakeRate, "fake-rate", o.Query.FakeRate, "synthetic log lines per second (fake backend)") fs.DurationVar(&o.Query.QueryTimeout, "query-timeout", o.Query.QueryTimeout, "maximum duration of a single storage query") + fs.DurationVar(&o.Query.MaxRange, "max-range", o.Query.MaxRange, "longest query_range window") fs.IntVar(&o.Query.DefaultLimit, "default-limit", o.Query.DefaultLimit, "default maximum log lines returned") fs.IntVar(&o.Query.MaxLimit, "max-limit", o.Query.MaxLimit, "hard ceiling on log lines returned") @@ -103,6 +122,20 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) { // extra, and from nowhere else. } +// Complete folds deprecated flags into their replacements. Call it after +// parsing and before Validate. +func (o *Options) Complete() error { + if o.legacyStorage == "" { + return nil + } + if o.flags != nil && o.flags.Changed("logs-storage") && o.Query.LogsStorage != o.legacyStorage { + return fmt.Errorf("--storage=%q conflicts with --logs-storage=%q; set only --logs-storage", + o.legacyStorage, o.Query.LogsStorage) + } + o.Query.LogsStorage = o.legacyStorage + return nil +} + // Validate reports whether the options are usable. func (o *Options) Validate() error { errs := o.Recommended.Validate() @@ -116,7 +149,7 @@ func (o *Options) Validate() error { } // Config assembles the server configuration. -func (o *Options) Config(logger *slog.Logger, store storage.LogStore) (*ServerConfig, error) { +func (o *Options) Config(logger *slog.Logger, logStore storage.LogStore, metricStore storage.MetricStore) (*ServerConfig, error) { // With no --tls-cert-file the apiserver convention is to self-sign into // --cert-dir. That is the local-development path only: in the cluster the // serving certificate comes from the cert-manager CSI driver @@ -143,6 +176,6 @@ func (o *Options) Config(logger *slog.Logger, store storage.LogStore) (*ServerCo return &ServerConfig{ Generic: generic, - Extra: ExtraConfig{Logger: logger, Store: store, Query: o.Query}, + Extra: ExtraConfig{Logger: logger, LogStore: logStore, MetricStore: metricStore, Query: o.Query}, }, nil } diff --git a/queryapi/prom.go b/queryapi/prom.go new file mode 100644 index 0000000..67e9e58 --- /dev/null +++ b/queryapi/prom.go @@ -0,0 +1,248 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package queryapi + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/url" + "strconv" + "strings" + "time" + + "go.datum.net/o11y/queryapi/internal/storage" +) + +// maxPoints is Prometheus's own ceiling on samples per series in a range +// query. +const maxPoints = 11000 + +// promForm parses the URL query and, for POST, the form body. +func promForm(w http.ResponseWriter, r *http.Request) (url.Values, bool) { + if err := r.ParseForm(); err != nil { + writePromError(w, http.StatusBadRequest, "bad_data", "invalid form: "+err.Error()) + return nil, false + } + return r.Form, true +} + +func required(w http.ResponseWriter, form url.Values, name string) (string, bool) { + v := strings.TrimSpace(form.Get(name)) + if v == "" { + writePromError(w, http.StatusBadRequest, "bad_data", name+" is required") + return "", false + } + return v, true +} + +func promTime(w http.ResponseWriter, form url.Values, name string, fallback time.Time) (time.Time, bool) { + t, err := parsePromTime(form.Get(name), fallback) + if err != nil { + writePromError(w, http.StatusBadRequest, "bad_data", name+": "+err.Error()) + return time.Time{}, false + } + return t, true +} + +// promTimeout caps a client timeout at --query-timeout. +func (h *handler) promTimeout(w http.ResponseWriter, form url.Values) (time.Duration, bool) { + raw := form.Get("timeout") + if raw == "" { + return h.cfg.QueryTimeout, true + } + d, err := parsePromDuration(raw) + if err != nil { + writePromError(w, http.StatusBadRequest, "bad_data", "timeout: "+err.Error()) + return 0, false + } + return min(d, h.cfg.QueryTimeout), true +} + +// promRange reads optional start/end. Absent bounds stay zero, which the +// store leaves to the backend's default. +func promRange(w http.ResponseWriter, form url.Values) (storage.TimeRange, bool) { + start, ok := promTime(w, form, "start", time.Time{}) + if !ok { + return storage.TimeRange{}, false + } + end, ok := promTime(w, form, "end", time.Time{}) + if !ok { + return storage.TimeRange{}, false + } + if !start.IsZero() && !end.IsZero() && end.Before(start) { + writePromError(w, http.StatusBadRequest, "bad_data", "end must not be before start") + return storage.TimeRange{}, false + } + return storage.TimeRange{Start: start, End: end}, true +} + +func (h *handler) promQuery(w http.ResponseWriter, r *http.Request) { + form, ok := promForm(w, r) + if !ok { + return + } + query, ok := required(w, form, "query") + if !ok { + return + } + at, ok := promTime(w, form, "time", time.Now().UTC()) + if !ok { + return + } + timeout, ok := h.promTimeout(w, form) + if !ok { + return + } + ctx, cancel := h.withTimeout(r) + defer cancel() + data, err := h.metricStore.Query(ctx, storage.InstantQuery{Query: query, Time: at, Timeout: timeout}) + h.writePromResult(w, data, err) +} + +func (h *handler) promQueryRange(w http.ResponseWriter, r *http.Request) { + form, ok := promForm(w, r) + if !ok { + return + } + query, ok := required(w, form, "query") + if !ok { + return + } + for _, name := range []string{"start", "end", "step"} { + if _, ok := required(w, form, name); !ok { + return + } + } + tr, ok := promRange(w, form) + if !ok { + return + } + step, err := parsePromDuration(form.Get("step")) + if err != nil { + writePromError(w, http.StatusBadRequest, "bad_data", "step: "+err.Error()) + return + } + window := tr.End.Sub(tr.Start) + if window > h.cfg.MaxRange { + writePromError(w, http.StatusBadRequest, "bad_data", + "range "+window.String()+" exceeds the maximum of "+h.cfg.MaxRange.String()) + return + } + if int64(window/step)+1 > maxPoints { + writePromError(w, http.StatusBadRequest, "bad_data", + "exceeded maximum resolution of 11,000 points per timeseries; try a larger step") + return + } + timeout, ok := h.promTimeout(w, form) + if !ok { + return + } + ctx, cancel := h.withTimeout(r) + defer cancel() + data, err := h.metricStore.QueryRange(ctx, storage.RangeQuery{Query: query, Range: tr, Step: step, Timeout: timeout}) + h.writePromResult(w, data, err) +} + +func (h *handler) promSeries(w http.ResponseWriter, r *http.Request) { + form, ok := promForm(w, r) + if !ok { + return + } + matchers := form["match[]"] + if len(matchers) == 0 { + writePromError(w, http.StatusBadRequest, "bad_data", "at least one match[] selector is required") + return + } + tr, ok := promRange(w, form) + if !ok { + return + } + ctx, cancel := h.withTimeout(r) + defer cancel() + data, err := h.metricStore.Series(ctx, matchers, tr) + h.writePromResult(w, data, err) +} + +func (h *handler) promLabelNames(w http.ResponseWriter, r *http.Request) { + form, ok := promForm(w, r) + if !ok { + return + } + tr, ok := promRange(w, form) + if !ok { + return + } + ctx, cancel := h.withTimeout(r) + defer cancel() + data, err := h.metricStore.LabelNames(ctx, form["match[]"], tr) + h.writePromResult(w, data, err) +} + +func (h *handler) promLabelValues(w http.ResponseWriter, r *http.Request) { + name := r.PathValue("name") + if !storage.ValidLabelName(name) { + writePromError(w, http.StatusBadRequest, "bad_data", "invalid label name "+strconv.Quote(name)) + return + } + form, ok := promForm(w, r) + if !ok { + return + } + tr, ok := promRange(w, form) + if !ok { + return + } + ctx, cancel := h.withTimeout(r) + defer cancel() + data, err := h.metricStore.LabelValues(ctx, name, form["match[]"], tr) + h.writePromResult(w, data, err) +} + +func (h *handler) writePromResult(w http.ResponseWriter, data json.RawMessage, err error) { + if err != nil { + h.writePromStorageError(w, err) + return + } + writeJSON(w, struct { + Status string `json:"status"` + Data json.RawMessage `json:"data"` + }{Status: "success", Data: data}) +} + +// writePromStorageError maps metric store failures. Only a QueryError's text +// reaches the client; anything about the backend is logged. +func (h *handler) writePromStorageError(w http.ResponseWriter, err error) { + var qe *storage.QueryError + switch { + case errors.As(err, &qe): + writePromError(w, http.StatusBadRequest, "bad_data", qe.Message) + case errors.Is(err, storage.ErrNoProject): + writePromError(w, http.StatusUnauthorized, "unauthorized", "request is not scoped to a project") + case errors.Is(err, storage.ErrResponseTooLarge): + writePromError(w, http.StatusUnprocessableEntity, "execution", + "response exceeds the size limit; narrow the query or the time range") + case errors.Is(err, context.Canceled): + writePromError(w, 499, "canceled", "client cancelled request") + case errors.Is(err, context.DeadlineExceeded): + writePromError(w, http.StatusServiceUnavailable, "timeout", "query timed out") + case errors.Is(err, storage.ErrBackendUnavailable): + h.logger.Error("metrics backend unavailable", "error", err) + writePromError(w, http.StatusBadGateway, "unavailable", "metrics backend unavailable") + default: + h.logger.Error("metric query failed", "error", err) + writePromError(w, http.StatusInternalServerError, "internal", "internal error") + } +} + +// writePromError writes Prometheus's error envelope. +func writePromError(w http.ResponseWriter, status int, errorType, message string) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(struct { + Status string `json:"status"` + ErrorType string `json:"errorType"` + Error string `json:"error"` + }{Status: "error", ErrorType: errorType, Error: message}) +} diff --git a/queryapi/prom_test.go b/queryapi/prom_test.go new file mode 100644 index 0000000..a3a6d0c --- /dev/null +++ b/queryapi/prom_test.go @@ -0,0 +1,275 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package queryapi_test + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/url" + "strings" + "sync" + "testing" + "time" + + "go.datum.net/o11y/queryapi" + "go.datum.net/o11y/queryapi/internal/storage" + "go.datum.net/o11y/queryapi/internal/storage/fake" +) + +// recordingMetricStore records what each handler asked the store for and answers +// with data or err. +type recordingMetricStore struct { + mu sync.Mutex + data json.RawMessage + err error + instant storage.InstantQuery + rng storage.RangeQuery + matchers []string + name string + calls int +} + +func (m *recordingMetricStore) record(f func()) (json.RawMessage, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.calls++ + f() + if m.err != nil { + return nil, m.err + } + if m.data == nil { + return json.RawMessage(`[]`), nil + } + return m.data, nil +} + +func (m *recordingMetricStore) Query(_ context.Context, q storage.InstantQuery) (json.RawMessage, error) { + return m.record(func() { m.instant = q }) +} + +func (m *recordingMetricStore) QueryRange(_ context.Context, q storage.RangeQuery) (json.RawMessage, error) { + return m.record(func() { m.rng = q }) +} + +func (m *recordingMetricStore) Series(_ context.Context, ms []string, _ storage.TimeRange) (json.RawMessage, error) { + return m.record(func() { m.matchers = ms }) +} + +func (m *recordingMetricStore) LabelNames(_ context.Context, ms []string, _ storage.TimeRange) (json.RawMessage, error) { + return m.record(func() { m.matchers = ms }) +} + +func (m *recordingMetricStore) LabelValues(_ context.Context, name string, ms []string, _ storage.TimeRange) (json.RawMessage, error) { + return m.record(func() { m.name, m.matchers = name, ms }) +} + +func (m *recordingMetricStore) Ping(context.Context) error { return nil } + +type promReply struct { + Status string `json:"status"` + ErrorType string `json:"errorType"` + Error string `json:"error"` + Data json.RawMessage `json:"data"` +} + +// promDo sends method to path with form as the body (POST) or the query +// string (GET), as an identified caller in proj-abc. +func promDo(t *testing.T, base, method, path string, form url.Values) (int, promReply) { + t.Helper() + target := base + prefix + path + var body *strings.Reader + if method == http.MethodPost { + body = strings.NewReader(form.Encode()) + } else { + if len(form) > 0 { + target += "?" + form.Encode() + } + body = strings.NewReader("") + } + req, err := http.NewRequest(method, target, body) + if err != nil { + t.Fatalf("build request: %v", err) + } + if method == http.MethodPost { + req.Header.Set("Content-Type", "application/x-www-form-urlencoded") + } + forwardIdentity(req) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("do: %v", err) + } + defer func() { + if err := resp.Body.Close(); err != nil { + t.Logf("close body: %v", err) + } + }() + var r promReply + if err := json.NewDecoder(resp.Body).Decode(&r); err != nil { + t.Fatalf("decode %s %s: %v", method, path, err) + } + return resp.StatusCode, r +} + +func newPromServer(t *testing.T, m storage.MetricStore) string { + t.Helper() + cfg := queryapi.DefaultConfig() + cfg.QueryTimeout = 30 * time.Second + return serveWithMetricStore(t, fake.NewLogStore(2), m, cfg, allowAll()).URL +} + +func TestPromQueryParameters(t *testing.T) { + m := &recordingMetricStore{} + base := newPromServer(t, m) + + code, r := promDo(t, base, http.MethodPost, "/metrics/api/v1/query", + url.Values{"query": {"up"}, "time": {"1760400000.123"}, "timeout": {"10s"}}) + if code != http.StatusOK || r.Status != "success" { + t.Fatalf("status %d %+v", code, r) + } + if m.instant.Query != "up" || !m.instant.Time.Equal(time.Unix(1760400000, 123_000_000)) || m.instant.Timeout != 10*time.Second { + t.Errorf("store got %+v", m.instant) + } + + // A client timeout above --query-timeout is capped, not rejected. + promDo(t, base, http.MethodPost, "/metrics/api/v1/query", url.Values{"query": {"up"}, "timeout": {"5m"}}) + if m.instant.Timeout != 30*time.Second { + t.Errorf("timeout = %v, want capped at 30s", m.instant.Timeout) + } +} + +func TestPromBadRequests(t *testing.T) { + m := &recordingMetricStore{} + base := newPromServer(t, m) + rangeForm := func(start, end, step string) url.Values { + return url.Values{"query": {"up"}, "start": {start}, "end": {end}, "step": {step}} + } + cases := []struct { + name, method, path string + form url.Values + }{ + {"query missing", http.MethodPost, "/metrics/api/v1/query", url.Values{}}, + {"bad time", http.MethodPost, "/metrics/api/v1/query", url.Values{"query": {"up"}, "time": {"soon"}}}, + {"zero timeout", http.MethodPost, "/metrics/api/v1/query", url.Values{"query": {"up"}, "timeout": {"0"}}}, + {"range no start", http.MethodPost, "/metrics/api/v1/query_range", rangeForm("", "60", "1")}, + {"range no end", http.MethodPost, "/metrics/api/v1/query_range", rangeForm("0", "", "1")}, + {"range no step", http.MethodPost, "/metrics/api/v1/query_range", rangeForm("0", "60", "")}, + {"range end before start", http.MethodPost, "/metrics/api/v1/query_range", rangeForm("60", "0", "1")}, + {"range over max-range", http.MethodPost, "/metrics/api/v1/query_range", rangeForm("0", "2678401", "3600")}, + {"range over 11000 points", http.MethodPost, "/metrics/api/v1/query_range", rangeForm("0", "11000", "1")}, + {"series no match", http.MethodGet, "/metrics/api/v1/series", url.Values{}}, + {"label name with digit", http.MethodGet, "/metrics/api/v1/label/1abc/values", nil}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + before := m.calls + code, r := promDo(t, base, tc.method, tc.path, tc.form) + if code != http.StatusBadRequest || r.ErrorType != "bad_data" { + t.Errorf("got %d %+v, want 400 bad_data", code, r) + } + if m.calls != before { + t.Errorf("store was called for a request that failed validation") + } + }) + } +} + +// An escaped slash decodes to "a/b" in the path value. The +// apiserver's filters may refuse it before the handler does, so any 4xx +// passes; reaching the store does not. +func TestPromLabelNameWithSlashNeverReachesTheStore(t *testing.T) { + m := &recordingMetricStore{} + base := newPromServer(t, m) + req, err := http.NewRequest(http.MethodGet, base+prefix+"/metrics/api/v1/label/a%2Fb/values", nil) + if err != nil { + t.Fatal(err) + } + forwardIdentity(req) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + if err := resp.Body.Close(); err != nil { + t.Logf("close body: %v", err) + } + if resp.StatusCode < 400 || resp.StatusCode >= 500 || m.calls != 0 { + t.Errorf("status %d, store calls %d; want a 4xx and none", resp.StatusCode, m.calls) + } +} + +func TestPromRangeAtTheLimitsIsAccepted(t *testing.T) { + m := &recordingMetricStore{} + base := newPromServer(t, m) + // 0..10999 at 1s is exactly 11,000 points. + code, r := promDo(t, base, http.MethodPost, "/metrics/api/v1/query_range", + url.Values{"query": {"up"}, "start": {"0"}, "end": {"10999"}, "step": {"1"}}) + if code != http.StatusOK { + t.Fatalf("got %d %+v", code, r) + } + if m.rng.Step != time.Second || !m.rng.Range.End.Equal(time.Unix(10999, 0)) { + t.Errorf("store got %+v", m.rng) + } +} + +func TestPromSelectorsAndLabelNames(t *testing.T) { + m := &recordingMetricStore{} + base := newPromServer(t, m) + + promDo(t, base, http.MethodGet, "/metrics/api/v1/series", url.Values{"match[]": {"up", `{job="a"}`}}) + if len(m.matchers) != 2 || m.matchers[0] != "up" || m.matchers[1] != `{job="a"}` { + t.Errorf("series matchers = %q", m.matchers) + } + + code, _ := promDo(t, base, http.MethodGet, "/metrics/api/v1/label/__name__/values", nil) + if code != http.StatusOK || m.name != "__name__" { + t.Errorf("label values: %d, name %q", code, m.name) + } +} + +// Scalar, string and matrix results pass through unchanged. +func TestPromDataPassesThroughUnchanged(t *testing.T) { + for _, data := range []string{ + `{"resultType":"scalar","result":[1760400000,"2"]}`, + `{"resultType":"string","result":[1760400000,"hello"]}`, + `{"resultType":"matrix","result":[]}`, + } { + m := &recordingMetricStore{data: json.RawMessage(data)} + base := newPromServer(t, m) + _, r := promDo(t, base, http.MethodPost, "/metrics/api/v1/query", url.Values{"query": {"1+1"}}) + if string(r.Data) != data { + t.Errorf("data = %s, want %s", r.Data, data) + } + } +} + +func TestPromStorageErrors(t *testing.T) { + cases := []struct { + err error + code int + errorType string + }{ + {&storage.QueryError{Message: "unknown function foo"}, http.StatusBadRequest, "bad_data"}, + {storage.ErrNoProject, http.StatusUnauthorized, "unauthorized"}, + {fmt.Errorf("wrapped: %w", context.DeadlineExceeded), http.StatusServiceUnavailable, "timeout"}, + {storage.ErrResponseTooLarge, http.StatusUnprocessableEntity, "execution"}, + {fmt.Errorf("%w: dial tcp 10.0.0.1:8429", storage.ErrBackendUnavailable), http.StatusBadGateway, "unavailable"}, + {errors.New("boom"), http.StatusInternalServerError, "internal"}, + } + for _, tc := range cases { + base := newPromServer(t, &recordingMetricStore{err: tc.err}) + code, r := promDo(t, base, http.MethodPost, "/metrics/api/v1/query", url.Values{"query": {"up"}}) + if code != tc.code || r.ErrorType != tc.errorType { + t.Errorf("%v: got %d %q, want %d %q", tc.err, code, r.ErrorType, tc.code, tc.errorType) + } + if strings.Contains(r.Error, "10.0.0.1") { + t.Errorf("%v: internal address leaked to the client: %q", tc.err, r.Error) + } + } + // The backend's own message for a bad query is returned. + base := newPromServer(t, &recordingMetricStore{err: &storage.QueryError{Message: "unknown function foo"}}) + if _, r := promDo(t, base, http.MethodPost, "/metrics/api/v1/query", url.Values{"query": {"foo()"}}); r.Error != "unknown function foo" { + t.Errorf("error = %q", r.Error) + } +} diff --git a/queryapi/promparams.go b/queryapi/promparams.go new file mode 100644 index 0000000..d6510c1 --- /dev/null +++ b/queryapi/promparams.go @@ -0,0 +1,78 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package queryapi + +import ( + "fmt" + "math" + "strconv" + "strings" + "time" + + "github.com/prometheus/common/model" +) + +// parsePromTime parses a Prometheus API timestamp: RFC 3339, or Unix seconds +// with an optional fraction. Empty returns fallback. +func parsePromTime(raw string, fallback time.Time) (time.Time, error) { + raw = strings.TrimSpace(raw) + if raw == "" { + return fallback, nil + } + if t, err := time.Parse(time.RFC3339Nano, raw); err == nil { + return t.UTC(), nil + } + if t, ok := parseUnixSeconds(raw); ok { + return t, nil + } + return time.Time{}, fmt.Errorf("invalid timestamp %q: want RFC 3339 or Unix seconds", raw) +} + +// parseUnixSeconds parses digits with an optional decimal fraction exactly, +// without a float, so nanoseconds survive. Digits past the ninth are +// truncated. +func parseUnixSeconds(raw string) (time.Time, bool) { + whole, frac, _ := strings.Cut(raw, ".") + if whole == "" || !isDigits(whole) || !isDigits(frac) { + return time.Time{}, false + } + sec, err := strconv.ParseInt(whole, 10, 64) + if err != nil { + return time.Time{}, false + } + if len(frac) > 9 { + frac = frac[:9] + } + var nsec int64 + if frac != "" { + nsec, _ = strconv.ParseInt(frac+strings.Repeat("0", 9-len(frac)), 10, 64) + } + return time.Unix(sec, nsec).UTC(), true +} + +func isDigits(s string) bool { + for _, r := range s { + if r < '0' || r > '9' { + return false + } + } + return true +} + +// parsePromDuration parses a Prometheus duration ("1m", "1h30m", "1d") or +// float seconds ("30", "0.5"). The result is always positive. +func parsePromDuration(raw string) (time.Duration, error) { + raw = strings.TrimSpace(raw) + if d, err := model.ParseDuration(raw); err == nil && d > 0 { + return time.Duration(d), nil + } + f, err := strconv.ParseFloat(raw, 64) + if err != nil || math.IsNaN(f) || math.IsInf(f, 0) || f <= 0 { + return 0, fmt.Errorf("invalid duration %q: want a Prometheus duration or positive seconds", raw) + } + ns := math.Round(f * float64(time.Second)) + if ns < 1 || ns >= math.MaxInt64 { + return 0, fmt.Errorf("duration %q out of range", raw) + } + return time.Duration(ns), nil +} diff --git a/queryapi/promparams_test.go b/queryapi/promparams_test.go new file mode 100644 index 0000000..3bdf4c0 --- /dev/null +++ b/queryapi/promparams_test.go @@ -0,0 +1,79 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package queryapi + +import ( + "testing" + "time" +) + +func TestParsePromTime(t *testing.T) { + fallback := time.Unix(42, 0).UTC() + cases := []struct { + raw string + want time.Time + wantErr bool + }{ + {"", fallback, false}, + {"1760400000", time.Unix(1760400000, 0).UTC(), false}, + {"1760400000.123", time.Unix(1760400000, 123_000_000).UTC(), false}, + {"1760400000.123456789", time.Unix(1760400000, 123_456_789).UTC(), false}, + {"1760400000.1234567891", time.Unix(1760400000, 123_456_789).UTC(), false}, // truncated past ns + {"1760400000.", time.Unix(1760400000, 0).UTC(), false}, + {"2026-10-14T00:00:00Z", time.Date(2026, 10, 14, 0, 0, 0, 0, time.UTC), false}, + {"2026-10-14T00:00:00.5Z", time.Date(2026, 10, 14, 0, 0, 0, 500_000_000, time.UTC), false}, + {"2026-10-14T02:00:00+02:00", time.Date(2026, 10, 14, 0, 0, 0, 0, time.UTC), false}, + {"abc", time.Time{}, true}, + {".5", time.Time{}, true}, + {"1e9", time.Time{}, true}, + {"-5", time.Time{}, true}, + } + for _, tc := range cases { + got, err := parsePromTime(tc.raw, fallback) + if tc.wantErr { + if err == nil { + t.Errorf("parsePromTime(%q) = %v, want an error", tc.raw, got) + } + continue + } + if err != nil || !got.Equal(tc.want) { + t.Errorf("parsePromTime(%q) = %v, %v; want %v", tc.raw, got, err, tc.want) + } + } +} + +func TestParsePromDuration(t *testing.T) { + cases := []struct { + raw string + want time.Duration + wantErr bool + }{ + {"30s", 30 * time.Second, false}, + {"1m", time.Minute, false}, + {"1h30m", 90 * time.Minute, false}, + {"1d", 24 * time.Hour, false}, + {"30", 30 * time.Second, false}, + {"0.5", 500 * time.Millisecond, false}, + {"", 0, true}, + {"-1", 0, true}, + {"NaN", 0, true}, + {"Inf", 0, true}, + {"soon", 0, true}, + {"1e-10", 0, true}, + {"0.0000000001", 0, true}, + {"1e30", 0, true}, + {"300000000000", 0, true}, + } + for _, tc := range cases { + got, err := parsePromDuration(tc.raw) + if tc.wantErr { + if err == nil { + t.Errorf("parsePromDuration(%q) = %v, want an error", tc.raw, got) + } + continue + } + if err != nil || got != tc.want { + t.Errorf("parsePromDuration(%q) = %v, %v; want %v", tc.raw, got, err, tc.want) + } + } +} diff --git a/queryapi/routes.go b/queryapi/routes.go index 48a9326..d98f552 100644 --- a/queryapi/routes.go +++ b/queryapi/routes.go @@ -45,8 +45,8 @@ type router struct { } // newRouter builds the tenant mux. -func newRouter(logger *slog.Logger, store storage.LogStore, cfg Config) (*router, error) { - h := &handler{logger: logger, store: store, cfg: cfg} +func newRouter(logger *slog.Logger, logStore storage.LogStore, metricStore storage.MetricStore, cfg Config) (*router, error) { + h := &handler{logger: logger, logStore: logStore, metricStore: metricStore, cfg: cfg} // Log endpoints mimic Loki's HTTP API, metric endpoints Prometheus's. // /logs and /metrics scope each signal to its own resource segment, which diff --git a/queryapi/routes_test.go b/queryapi/routes_test.go index 98166f9..12d5da3 100644 --- a/queryapi/routes_test.go +++ b/queryapi/routes_test.go @@ -119,7 +119,7 @@ func TestCanonicalPathPreservesQuery(t *testing.T) { // comes back as a non-resource request, which authz's guard denies and which // the tenant mux would answer 404 anyway. func TestRequestInfoResolverIsFailClosed(t *testing.T) { - router, err := newRouter(slog.New(slog.DiscardHandler), fake.New(2), DefaultConfig()) + router, err := newRouter(slog.New(slog.DiscardHandler), fake.NewLogStore(2), fake.NewMetricStore(), DefaultConfig()) if err != nil { t.Fatalf("newRouter: %v", err) } @@ -198,7 +198,7 @@ func TestRequestInfoResolverIsFailClosed(t *testing.T) { // against are resolved by the framework's own resolver, which is what that // flag matches on. func TestRequestInfoResolverDelegatesOutsideTheGroup(t *testing.T) { - router, err := newRouter(slog.New(slog.DiscardHandler), fake.New(2), DefaultConfig()) + router, err := newRouter(slog.New(slog.DiscardHandler), fake.NewLogStore(2), fake.NewMetricStore(), DefaultConfig()) if err != nil { t.Fatalf("newRouter: %v", err) }