diff --git a/README.md b/README.md index 115e888..7d4deb0 100644 --- a/README.md +++ b/README.md @@ -75,7 +75,9 @@ reaches a different upstream with no change to the Worker. (the gRPC method, and the Namespace the proxy resolved from the request rather than from anything the caller claims), so it can decide per Namespace and per method rather than only whether the caller is who it says it is. - **Prometheus metrics.** Expose request latency and counts, routing decisions, and encryption activity on `/metrics`. - The listen address and the namespace prefixed onto every metric are set under `metrics:` in the config. + The listen address and the namespace prefixed onto every metric are set under `metrics:` in the config, which can + also name request metadata to carry onto the request-scoped metrics as extra labels, so they can be sliced by a + dimension only your callers know. - **Codec-transparent.** The gateway never parses payloads. It peeks the Namespace, picks an upstream, and relays raw frames in both directions. - **Multiple deployment options.** Ship as a Go binary, a container image, or a Helm chart. diff --git a/e2e/metric_tags_test.go b/e2e/metric_tags_test.go new file mode 100644 index 0000000..19a7c9a --- /dev/null +++ b/e2e/metric_tags_test.go @@ -0,0 +1,129 @@ +package e2e + +import ( + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" + "go.temporal.io/api/common/v1" + "go.temporal.io/api/query/v1" + "go.temporal.io/api/workflowservice/v1" + "google.golang.org/grpc" + "google.golang.org/grpc/metadata" + + "github.com/temporalio/temporal-proxy/internal/config" + "github.com/temporalio/temporal-proxy/internal/dataplane/dataplanetest" +) + +// TestEndToEndMetricTags drives a payload-carrying call through the full stack +// with a metrics tag configured, and proves the caller's metadata reaches the +// labels of every request-scoped collector. +// +// The vault_ops assertion is the one that matters: that collector lives on the +// per-upstream hop, on the far side of a unix socket that context values do not +// cross. It passes only because the value is read from the metadata the gateway +// forwards, and it is what fails if this is ever reworked into a single +// interceptor stashing values in a context. +func TestEndToEndMetricTags(t *testing.T) { + t.Parallel() + + up := dataplanetest.NewUpstream(t) + + cfg := dataplanetest.Config(up) + // Mixed case on purpose: gRPC canonicalizes metadata keys, so a config + // written this way still has to match what arrives on the wire. + cfg.Metrics.Tags = []config.MetricTag{{Header: "X-Tenant", Label: "tenant"}} + // Encryption is what makes the per-upstream hop emit vault_ops at all. + cfg.Encryption = config.Encryption{ + Enabled: true, + Default: &config.KeyPolicy{URI: testingKeyURI(t), Duration: time.Hour}, + } + + f := dataplanetest.StartApp(t, cfg) + + ctx := metadata.AppendToOutgoingContext(f.Context(), "x-tenant", "acme") + _, err := f.Client().QueryWorkflow(ctx, queryWithPayload(), grpc.WaitForReady(true)) + require.NoError(t, err) + + // Hop 1: the gateway's own interceptor and the routing decision. + requireLabel(t, f, "test_server_requests_total", "tenant", "acme") + requireLabel(t, f, "test_router_decisions_total", "tenant", "acme") + + // Hop 2: across the socket. + requireLabel(t, f, "test_encryption_vault_ops_total", "tenant", "acme") +} + +// TestEndToEndMetricTagsWithoutTheHeader proves a configured label is always +// present, empty when the caller sent nothing, rather than the series being +// reported without it. +func TestEndToEndMetricTagsWithoutTheHeader(t *testing.T) { + t.Parallel() + + up := dataplanetest.NewUpstream(t) + + cfg := dataplanetest.Config(up) + cfg.Metrics.Tags = []config.MetricTag{{Header: "x-tenant", Label: "tenant"}} + cfg.Encryption = config.Encryption{ + Enabled: true, + Default: &config.KeyPolicy{URI: testingKeyURI(t), Duration: time.Hour}, + } + + f := dataplanetest.StartApp(t, cfg) + + _, err := f.Client().QueryWorkflow(f.Context(), queryWithPayload(), grpc.WaitForReady(true)) + require.NoError(t, err) + + requireLabel(t, f, "test_server_requests_total", "tenant", "") + requireLabel(t, f, "test_router_decisions_total", "tenant", "") + requireLabel(t, f, "test_encryption_vault_ops_total", "tenant", "") +} + +// queryWithPayload is a QueryWorkflow request carrying one payload, so the +// encryption interceptor on the per-upstream hop has something to seal. +func queryWithPayload() *workflowservice.QueryWorkflowRequest { + return &workflowservice.QueryWorkflowRequest{ + Namespace: "ns1", + Execution: &common.WorkflowExecution{WorkflowId: "wf-1"}, + Query: &query.WorkflowQuery{ + QueryType: "state", + QueryArgs: &common.Payloads{Payloads: []*common.Payload{{ + Metadata: map[string][]byte{wireEncoding: []byte("json/plain")}, + Data: []byte(`"hi"`), + }}}, + }, + } +} + +// requireLabel asserts that some series in the named metric family carries +// label=want. It compares against the gathered registry rather than a scrape, +// which is equivalent: the metrics module hands that same Gatherer to its +// /metrics handler. +func requireLabel(t *testing.T, f *dataplanetest.Fixture, family, label, want string) { + t.Helper() + + found, err := testutil.GatherAndLint(f.Gatherer(), family) + require.NoError(t, err) + require.Empty(t, found, "collector %s failed prometheus linting", family) + + mfs, err := f.Gatherer().Gather() + require.NoError(t, err) + + for _, mf := range mfs { + if mf.GetName() != family { + continue + } + + for _, m := range mf.GetMetric() { + for _, lp := range m.GetLabel() { + if lp.GetName() == label && lp.GetValue() == want { + return + } + } + } + + t.Fatalf("metric family %s has no series with %s=%q, got %v", family, label, want, mf.GetMetric()) + } + + t.Fatalf("metric family %s was never registered", family) +} diff --git a/internal/config/metrics.go b/internal/config/metrics.go index 99b1e58..d95d2f3 100644 --- a/internal/config/metrics.go +++ b/internal/config/metrics.go @@ -1,23 +1,189 @@ package config -import "github.com/temporalio/temporal-proxy/pkg/validation" +import ( + "errors" + "fmt" + "regexp" + "slices" + "strings" -// Metrics configures the Prometheus endpoint. HostPort is the address the -// /metrics handler listens on, and Namespace is the prefix stamped onto every -// collector: a Prometheus namespace, unrelated to a Temporal namespace. Load -// defaults both, so neither is empty in a loaded config. -type Metrics struct { - HostPort string `yaml:"hostPort"` - Namespace string `yaml:"namespace"` + "github.com/temporalio/temporal-proxy/pkg/validation" +) + +const ( + binaryHeaderSuffix = "-bin" // gRPC binary metadata marker + reservedHeaderPrefix = "grpc-" // gRPC reserved metadata prefix + reservedLabelPrefix = "__" // Prometheus reserved label prefix +) + +var ( + // promReservedLabels are the label names Prometheus keeps for a histogram's + // bucket bound and a summary's quantile. + promReservedLabels = []string{"le", "quantile"} + + // mdKeyRegex matches the characters gRPC allows in a metadata key. + mdKeyRegex = regexp.MustCompile(`^[a-zA-Z0-9._-]+$`) + + // promLabelRegex matches the Prometheus label name grammar. + promLabelRegex = regexp.MustCompile(`^[a-zA-Z_][a-zA-Z0-9_]*$`) +) + +type ( + // Metrics configures the Prometheus endpoint. HostPort is the address the + // /metrics handler listens on, and Namespace is the prefix stamped onto every + // collector: a Prometheus namespace, unrelated to a Temporal namespace. Tags + // name the request metadata to carry through as labels. Load defaults + // HostPort and Namespace, so neither is empty in a loaded config. + Metrics struct { + HostPort string `yaml:"hostPort"` + Namespace string `yaml:"namespace"` + Tags []MetricTag `yaml:"tags"` + } + + // MetricTag pairs an inbound request metadata Header with the Prometheus + // Label it is reported under. It is written in YAML as "
: