From 9319f0ad955060d31bb8685100402507397e7a46 Mon Sep 17 00:00:00 2001 From: "David Muto (pseudomuto)" Date: Sun, 26 Jul 2026 13:26:50 -0400 Subject: [PATCH] [encryption]: Add payload encryption observability metrics The payload encryption subsystem emitted no telemetry, leaving operators blind to KMS pressure, DEK cache effectiveness, and encryption error rates. This adds seven Prometheus metrics under the "encryption" subsystem: - encryption_dek_ops_total (operation, result, namespace) and encryption_dek_ops_duration_secs (operation, namespace) for payload seal/open - encryption_kek_ops_total (provider, operation, result) and encryption_kek_ops_duration_secs (provider, operation) for KEK wrap/unwrap - encryption_dek_cache_hits_total / encryption_dek_cache_misses_total / encryption_dek_cache_size To keep pkg/crypto free of both Prometheus and internal packages, the Vault emits cache events through a small Observer interface, whose methods take a struct so fields can be added later without breaking implementers. internal/kms implements it with a Prometheus-backed Reporter and meters KEK wrap/unwrap through a crypto.KEK decorator that resolves the provider label from the URI scheme. DEK payload operations are recorded by the proxy encryption interceptor, the one place the namespace is known. Both reporters are built once and share the "encryption" subsystem with disjoint metric names, so there is no duplicate-registration risk. Metric collection is behavior-preserving: the KEK decorator returns the wrapped call's result and error verbatim, and the interceptor's timing is purely additive to the existing seal/open path. The namespace label on dek_ops is a deliberate choice to allow per-namespace attribution of encryption work; the remaining metrics avoid high-cardinality labels to keep series counts bounded. --- e2e/harness_test.go | 2 + internal/kms/fx.go | 29 ++++-- internal/kms/fx_test.go | 39 +++++--- internal/kms/metered.go | 71 +++++++++++++ internal/kms/metered_test.go | 87 ++++++++++++++++ internal/kms/reporter.go | 113 +++++++++++++++++++++ internal/kms/reporter_test.go | 86 ++++++++++++++++ internal/metrics/factory.go | 12 +++ internal/metrics/factory_test.go | 24 +++++ internal/proxy/encryption.go | 37 +++++-- internal/proxy/encryption_test.go | 160 +++++++++++++++++++++++++++--- internal/proxy/fx.go | 11 +- internal/proxy/fx_test.go | 5 + internal/proxy/reporter.go | 38 +++++++ pkg/crypto/observer.go | 25 +++++ pkg/crypto/observer_test.go | 81 +++++++++++++++ pkg/crypto/vault.go | 20 ++++ 17 files changed, 795 insertions(+), 45 deletions(-) create mode 100644 internal/kms/metered.go create mode 100644 internal/kms/metered_test.go create mode 100644 internal/kms/reporter.go create mode 100644 internal/kms/reporter_test.go create mode 100644 internal/proxy/reporter.go create mode 100644 pkg/crypto/observer.go create mode 100644 pkg/crypto/observer_test.go diff --git a/e2e/harness_test.go b/e2e/harness_test.go index 14b90ce..d6931c6 100644 --- a/e2e/harness_test.go +++ b/e2e/harness_test.go @@ -8,6 +8,7 @@ import ( "testing" "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" "github.com/stretchr/testify/require" "go.temporal.io/api/workflowservice/v1" "go.uber.org/fx" @@ -124,6 +125,7 @@ func newProxyApp(t *testing.T, cfg *config.Config) *fx.App { fx.Supply(fx.Annotate(t.Context(), fx.As(new(context.Context)))), fx.Supply(cfg), fx.Provide(func() *crypto.Vault { return nil }), + fx.Provide(func() *metrics.Factory { return metrics.New("test", promauto.With(prometheus.NewRegistry())) }), connect.Module, protoutil.Module, proxy.Module, diff --git a/internal/kms/fx.go b/internal/kms/fx.go index a46a479..4a01a7e 100644 --- a/internal/kms/fx.go +++ b/internal/kms/fx.go @@ -11,6 +11,7 @@ import ( "go.uber.org/fx" "github.com/temporalio/temporal-proxy/internal/config" + "github.com/temporalio/temporal-proxy/internal/metrics" "github.com/temporalio/temporal-proxy/pkg/crypto" "github.com/temporalio/temporal-proxy/pkg/logger" "github.com/temporalio/temporal-proxy/pkg/logger/tag" @@ -35,17 +36,20 @@ var Module = fx.Options( return nil, nil } + reporter := NewReporter(p.Factory.ForSubsystem("encryption")) + r, err := createKEKRegistry( p.Context, p.Lifecycle, p.Config, p.Logger, + reporter, ) if err != nil { return nil, err } - v, err := createVault(p.Config, r) + v, err := createVault(p.Config, r, reporter) if err != nil { _ = r.Close() return nil, err @@ -83,6 +87,7 @@ type ( Config *config.Config Lifecycle fx.Lifecycle Logger logger.Logger + Factory *metrics.Factory } // vaultRefresher is the subset of *crypto.Vault the rotation loop depends @@ -126,9 +131,12 @@ func runRotation(ctx context.Context, v vaultRefresher, interval time.Duration, // createVault builds a vault from the registry, applying the configured cache // size and, when a default key policy is set, its DEK duration and renewal lead // time. -func createVault(c *config.Config, r *crypto.KEKRegistry) (*crypto.Vault, error) { - opts := make([]crypto.VaultOption, 0, 2+len(c.Encryption.Overrides)) - opts = append(opts, crypto.WithCacheSize(c.Encryption.CacheSize)) +func createVault(c *config.Config, r *crypto.KEKRegistry, reporter *Reporter) (*crypto.Vault, error) { + opts := make([]crypto.VaultOption, 0, 3+len(c.Encryption.Overrides)) + opts = append(opts, + crypto.WithCacheSize(c.Encryption.CacheSize), + crypto.WithObserver(reporter), + ) if dp := c.Encryption.Default; dp != nil { opts = append(opts, crypto.WithDefaultKeyConfig(crypto.KeyConfig{ @@ -158,10 +166,10 @@ func createVault(c *config.Config, r *crypto.KEKRegistry) (*crypto.Vault, error) // createKEKRegistry opens the configured KEKs and assembles a registry, // registering an fx OnStop hook that closes the registry (and its KEKs) on // shutdown. -func createKEKRegistry(ctx context.Context, lc fx.Lifecycle, c *config.Config, logger logger.Logger) (*crypto.KEKRegistry, error) { +func createKEKRegistry(ctx context.Context, lc fx.Lifecycle, c *config.Config, logger logger.Logger, reporter *Reporter) (*crypto.KEKRegistry, error) { opts := []crypto.KEKRegistryOption{} if dp := c.Encryption.Default; dp != nil { - res, err := keyPolicyRegistryOpts(ctx, dp, logger, defaultNamespace, true) + res, err := keyPolicyRegistryOpts(ctx, dp, logger, defaultNamespace, true, reporter) if err != nil { return nil, err } @@ -173,7 +181,7 @@ func createKEKRegistry(ctx context.Context, lc fx.Lifecycle, c *config.Config, l // logs and, on a partial failure, a repeatable point of failure). for _, ns := range slices.Sorted(maps.Keys(c.Encryption.Overrides)) { policy := c.Encryption.Overrides[ns] - res, err := keyPolicyRegistryOpts(ctx, &policy, logger, ns, false) + res, err := keyPolicyRegistryOpts(ctx, &policy, logger, ns, false, reporter) if err != nil { return nil, err } @@ -207,9 +215,10 @@ func keyPolicyRegistryOpts( log logger.Logger, ns string, asDefault bool, + reporter *Reporter, ) ([]crypto.KEKRegistryOption, error) { opts := []crypto.KEKRegistryOption{} - keys, err := createKEKs(ctx, p, log, ns) + keys, err := createKEKs(ctx, p, log, ns, reporter) if err != nil { return nil, err } @@ -232,7 +241,7 @@ func keyPolicyRegistryOpts( // returning the KEKs in that order. The primary key is always element zero. If // any key fails to open, every key opened so far is closed before returning, so // a partial failure leaks no keepers. -func createKEKs(ctx context.Context, p *config.KeyPolicy, log logger.Logger, ns string) (_ []crypto.KEK, err error) { +func createKEKs(ctx context.Context, p *config.KeyPolicy, log logger.Logger, ns string, reporter *Reporter) (_ []crypto.KEK, err error) { log = log.With(tag.String("namespace", ns)) keys := make([]crypto.KEK, 0, len(p.DecryptURIs)+1) @@ -251,7 +260,7 @@ func createKEKs(ctx context.Context, p *config.KeyPolicy, log logger.Logger, ns return err } - keys = append(keys, k) + keys = append(keys, newMeteredKEK(k, providerForScheme(uri.Scheme), reporter)) return nil } diff --git a/internal/kms/fx_test.go b/internal/kms/fx_test.go index 778474d..57a4245 100644 --- a/internal/kms/fx_test.go +++ b/internal/kms/fx_test.go @@ -11,11 +11,14 @@ import ( "testing/synctest" "time" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" "github.com/stretchr/testify/require" "go.uber.org/fx" "go.uber.org/fx/fxtest" "github.com/temporalio/temporal-proxy/internal/config" + "github.com/temporalio/temporal-proxy/internal/metrics" "github.com/temporalio/temporal-proxy/pkg/crypto" "github.com/temporalio/temporal-proxy/pkg/logger" ) @@ -88,7 +91,7 @@ func TestCreateKEKs(t *testing.T) { t.Parallel() p := &config.KeyPolicy{URI: primary, DecryptURIs: []url.URL{distinctKeyURL(t, 2), distinctKeyURL(t, 3)}} - keys, err := createKEKs(t.Context(), p, log, defaultNamespace) + keys, err := createKEKs(t.Context(), p, log, defaultNamespace, newFxTestReporter(t)) require.NoError(t, err) t.Cleanup(func() { closeKEKs(t, keys) }) @@ -99,7 +102,7 @@ func TestCreateKEKs(t *testing.T) { t.Run("primary only", func(t *testing.T) { t.Parallel() - keys, err := createKEKs(t.Context(), &config.KeyPolicy{URI: primary}, log, defaultNamespace) + keys, err := createKEKs(t.Context(), &config.KeyPolicy{URI: primary}, log, defaultNamespace, newFxTestReporter(t)) require.NoError(t, err) t.Cleanup(func() { closeKEKs(t, keys) }) @@ -109,7 +112,7 @@ func TestCreateKEKs(t *testing.T) { t.Run("bad primary uri errors", func(t *testing.T) { t.Parallel() - _, err := createKEKs(t.Context(), &config.KeyPolicy{URI: url.URL{Scheme: "bogus", Host: "x"}}, log, defaultNamespace) + _, err := createKEKs(t.Context(), &config.KeyPolicy{URI: url.URL{Scheme: "bogus", Host: "x"}}, log, defaultNamespace, newFxTestReporter(t)) require.Error(t, err) }) @@ -117,7 +120,7 @@ func TestCreateKEKs(t *testing.T) { t.Parallel() p := &config.KeyPolicy{URI: primary, DecryptURIs: []url.URL{{Scheme: "bogus", Host: "x"}}} - _, err := createKEKs(t.Context(), p, log, defaultNamespace) + _, err := createKEKs(t.Context(), p, log, defaultNamespace, newFxTestReporter(t)) require.Error(t, err) }) } @@ -134,7 +137,7 @@ func TestKeyPolicyRegistryOpts(t *testing.T) { t.Run("asDefault registers a usable default key", func(t *testing.T) { t.Parallel() - opts, err := keyPolicyRegistryOpts(t.Context(), &config.KeyPolicy{URI: distinctKeyURL(t, 1)}, log, defaultNamespace, true) + opts, err := keyPolicyRegistryOpts(t.Context(), &config.KeyPolicy{URI: distinctKeyURL(t, 1)}, log, defaultNamespace, true, newFxTestReporter(t)) require.NoError(t, err) reg, err := crypto.NewKEKRegistry(opts...) @@ -145,7 +148,7 @@ func TestKeyPolicyRegistryOpts(t *testing.T) { t.Run("non-default registers only a namespace key", func(t *testing.T) { t.Parallel() - opts, err := keyPolicyRegistryOpts(t.Context(), &config.KeyPolicy{URI: distinctKeyURL(t, 2)}, log, "other", false) + opts, err := keyPolicyRegistryOpts(t.Context(), &config.KeyPolicy{URI: distinctKeyURL(t, 2)}, log, "other", false, newFxTestReporter(t)) require.NoError(t, err) _, err = crypto.NewKEKRegistry(opts...) @@ -158,7 +161,7 @@ func TestKeyPolicyRegistryOpts(t *testing.T) { t.Run("default namespace with asDefault false is a namespace key", func(t *testing.T) { t.Parallel() - opts, err := keyPolicyRegistryOpts(t.Context(), &config.KeyPolicy{URI: distinctKeyURL(t, 3)}, log, defaultNamespace, false) + opts, err := keyPolicyRegistryOpts(t.Context(), &config.KeyPolicy{URI: distinctKeyURL(t, 3)}, log, defaultNamespace, false, newFxTestReporter(t)) require.NoError(t, err) _, err = crypto.NewKEKRegistry(opts...) @@ -193,7 +196,7 @@ func TestCreateVault(t *testing.T) { t.Run(tt.name, func(t *testing.T) { t.Parallel() - v, err := createVault(&config.Config{Encryption: tt.enc}, reg) + v, err := createVault(&config.Config{Encryption: tt.enc}, reg, newFxTestReporter(t)) require.NoError(t, err) require.NotNil(t, v) }) @@ -221,7 +224,7 @@ func TestCreateVault_AppliesOverrideKeyConfig(t *testing.T) { }, }} - _, err = createVault(cfg, reg) + _, err = createVault(cfg, reg, newFxTestReporter(t)) require.Error(t, err) } @@ -231,7 +234,7 @@ func TestCreateKEKRegistry(t *testing.T) { lc := fxtest.NewLifecycle(t) cfg := &config.Config{Encryption: config.Encryption{Enabled: true, Default: &config.KeyPolicy{URI: distinctKeyURL(t, 1)}}} - reg, err := createKEKRegistry(t.Context(), lc, cfg, logger.NewNoopLogger()) + reg, err := createKEKRegistry(t.Context(), lc, cfg, logger.NewNoopLogger(), newFxTestReporter(t)) require.NoError(t, err) require.NotNil(t, reg) @@ -256,12 +259,12 @@ func TestCreateKEKRegistry_OverrideKeySelectedByNamespace(t *testing.T) { }, }} - reg, err := createKEKRegistry(t.Context(), lc, cfg, logger.NewNoopLogger()) + reg, err := createKEKRegistry(t.Context(), lc, cfg, logger.NewNoopLogger(), newFxTestReporter(t)) require.NoError(t, err) lc.RequireStart() t.Cleanup(func() { lc.RequireStop() }) - vault, err := createVault(cfg, reg) + vault, err := createVault(cfg, reg, newFxTestReporter(t)) require.NoError(t, err) // The override namespace seals under its own KEK. @@ -285,6 +288,7 @@ func TestModule_NoKeys_ProvidesNilVault(t *testing.T) { fx.Supply(fx.Annotate(t.Context(), fx.As(new(context.Context)))), fx.Supply(&config.Config{Encryption: config.Encryption{Enabled: false}}), fx.Provide(func() logger.Logger { return logger.NewNoopLogger() }), + fx.Provide(func() *metrics.Factory { return metrics.New("test", promauto.With(prometheus.NewRegistry())) }), Module, fx.Populate(&v), fx.NopLogger, @@ -312,6 +316,7 @@ func TestModule_DisabledWithKeys_ProvidesVault(t *testing.T) { fx.Supply(fx.Annotate(t.Context(), fx.As(new(context.Context)))), fx.Supply(cfg), fx.Provide(func() logger.Logger { return logger.NewNoopLogger() }), + fx.Provide(func() *metrics.Factory { return metrics.New("test", promauto.With(prometheus.NewRegistry())) }), Module, fx.Populate(&v), ) @@ -336,6 +341,7 @@ func TestModule_Enabled_ProvidesVaultAndRunsCleanly(t *testing.T) { fx.Supply(fx.Annotate(t.Context(), fx.As(new(context.Context)))), fx.Supply(cfg), fx.Provide(func() logger.Logger { return logger.NewNoopLogger() }), + fx.Provide(func() *metrics.Factory { return metrics.New("test", promauto.With(prometheus.NewRegistry())) }), Module, fx.Populate(&v), ) @@ -360,6 +366,7 @@ func TestModule_Enabled_InvalidURI_FailsConstruction(t *testing.T) { fx.Supply(fx.Annotate(t.Context(), fx.As(new(context.Context)))), fx.Supply(cfg), fx.Provide(func() logger.Logger { return logger.NewNoopLogger() }), + fx.Provide(func() *metrics.Factory { return metrics.New("test", promauto.With(prometheus.NewRegistry())) }), Module, fx.NopLogger, ) @@ -389,3 +396,11 @@ func closeKEKs(t *testing.T, keys []crypto.KEK) { require.NoError(t, k.Close()) } } + +// newFxTestReporter builds a Reporter backed by its own private registry, so +// fx.go call sites under test have a Reporter to record into without +// colliding with any other test's metrics. +func newFxTestReporter(t *testing.T) *Reporter { + t.Helper() + return NewReporter(metrics.New("test", promauto.With(prometheus.NewRegistry())).ForSubsystem("encryption")) +} diff --git a/internal/kms/metered.go b/internal/kms/metered.go new file mode 100644 index 0000000..2f03bcb --- /dev/null +++ b/internal/kms/metered.go @@ -0,0 +1,71 @@ +package kms + +import ( + "context" + "time" + + "github.com/temporalio/temporal-proxy/pkg/crypto" +) + +type ( + // meteredKEK decorates a crypto.KEK, recording each wrap (Encrypt) and unwrap + // (Decrypt) as a KEK operation. It embeds the wrapped KEK so ID and Close pass + // through unchanged, and returns the wrapped call's result and error verbatim so + // metering never alters behavior. + meteredKEK struct { + crypto.KEK + provider string + recorder kekRecorder + } + + // kekRecorder records KEK wrap/unwrap operations. *Reporter satisfies it; the + // interface lets meteredKEK be tested without a Prometheus-backed reporter. + kekRecorder interface { + KEKOp(provider, operation, result string, seconds float64) + } +) + +func (m *meteredKEK) Encrypt(ctx context.Context, b []byte) ([]byte, error) { + start := time.Now() + ct, err := m.KEK.Encrypt(ctx, b) + m.recorder.KEKOp(m.provider, "wrap", resultLabel(err), time.Since(start).Seconds()) + return ct, err +} + +func (m *meteredKEK) Decrypt(ctx context.Context, b []byte) ([]byte, error) { + start := time.Now() + pt, err := m.KEK.Decrypt(ctx, b) + m.recorder.KEKOp(m.provider, "unwrap", resultLabel(err), time.Since(start).Seconds()) + return pt, err +} + +// newMeteredKEK wraps k so its KMS calls are recorded under provider. +func newMeteredKEK(k crypto.KEK, provider string, r kekRecorder) crypto.KEK { + return &meteredKEK{KEK: k, provider: provider, recorder: r} +} + +// providerForScheme maps a KMS URI scheme to a stable, low-cardinality provider +// label. Unknown schemes are returned unchanged so a new backend still produces +// a usable (if unrecognized) label rather than an empty one. +func providerForScheme(scheme string) string { + switch scheme { + case "awskms": + return "aws" + case "gcpkms": + return "gcp" + case "azurekeyvault": + return "azure" + case "testing", "base64key": + return "testing" + default: + return scheme + } +} + +// resultLabel maps an error to the "result" metric label value. +func resultLabel(err error) string { + if err != nil { + return "error" + } + return "success" +} diff --git a/internal/kms/metered_test.go b/internal/kms/metered_test.go new file mode 100644 index 0000000..4fac84d --- /dev/null +++ b/internal/kms/metered_test.go @@ -0,0 +1,87 @@ +package kms + +import ( + "context" + "errors" + "testing" + + "github.com/stretchr/testify/require" +) + +type ( + stubKEK struct { + err error + } + + recordedKEKOp struct { + provider string + operation string + result string + } + + fakeRecorder struct { + ops []recordedKEKOp + } +) + +func TestMeteredKEKRecordsWrap(t *testing.T) { + t.Parallel() + + r := &fakeRecorder{} + k := newMeteredKEK(stubKEK{}, "aws", r) + + out, err := k.Encrypt(t.Context(), []byte("dek")) + require.NoError(t, err) + require.Equal(t, []byte("dek"), out) // result passed through unchanged + + require.Equal(t, []recordedKEKOp{{"aws", "wrap", "success"}}, r.ops) +} + +func TestMeteredKEKRecordsUnwrapError(t *testing.T) { + t.Parallel() + + r := &fakeRecorder{} + sentinel := errors.New("boom") + k := newMeteredKEK(stubKEK{err: sentinel}, "gcp", r) + + _, err := k.Decrypt(t.Context(), []byte("x")) + require.ErrorIs(t, err, sentinel) // error passed through unchanged + + require.Equal(t, []recordedKEKOp{{"gcp", "unwrap", "error"}}, r.ops) +} + +func TestProviderForScheme(t *testing.T) { + t.Parallel() + + cases := map[string]string{ + "awskms": "aws", + "gcpkms": "gcp", + "azurekeyvault": "azure", + "testing": "testing", + "base64key": "testing", + "weird": "weird", + } + for scheme, want := range cases { + require.Equal(t, want, providerForScheme(scheme), "scheme %q", scheme) + } +} + +func (s stubKEK) ID() string { return "stub" } +func (s stubKEK) Encrypt(_ context.Context, b []byte) ([]byte, error) { + if s.err != nil { + return nil, s.err + } + return b, nil +} + +func (s stubKEK) Decrypt(_ context.Context, b []byte) ([]byte, error) { + if s.err != nil { + return nil, s.err + } + return b, nil +} +func (s stubKEK) Close() error { return nil } + +func (f *fakeRecorder) KEKOp(provider, operation, result string, _ float64) { + f.ops = append(f.ops, recordedKEKOp{provider, operation, result}) +} diff --git a/internal/kms/reporter.go b/internal/kms/reporter.go new file mode 100644 index 0000000..b8d24de --- /dev/null +++ b/internal/kms/reporter.go @@ -0,0 +1,113 @@ +package kms + +import ( + "github.com/prometheus/client_golang/prometheus" + + "github.com/temporalio/temporal-proxy/internal/metrics" + "github.com/temporalio/temporal-proxy/pkg/crypto" +) + +type ( + // Reporter records encryption telemetry to Prometheus: KEK wrap/unwrap + // calls and DEK cache behavior. It implements crypto.Observer so a Vault can + // notify it of cache events, and exposes KEKOp for the KEK decorator. Handles + // for the low-cardinality KEK label combinations are pre-resolved so the emit + // path is a lock-free map read; an unexpected combination falls back to + // WithLabelValues. A Reporter is safe for concurrent use. + Reporter struct { + kekOps *prometheus.CounterVec + kekOpDur *prometheus.HistogramVec + cacheHits prometheus.Counter + cacheMisses prometheus.Counter + cacheSize prometheus.Gauge + + kekOpHandle map[kekOpKey]prometheus.Counter + kekDurHandle map[kekDurKey]prometheus.Observer + } + + kekOpKey struct { + provider string + operation string + result string + } + + kekDurKey struct { + provider string + operation string + } +) + +// NewReporter builds the Prometheus-backed encryption Reporter, pre-resolving +// the meaningful KEK label combinations so every series starts at zero. f must +// already be scoped to the "encryption" subsystem by the caller. +func NewReporter(f *metrics.Factory) *Reporter { + kekOps := f.NewCounter(prometheus.CounterOpts{ + Name: "kek_ops_total", + Help: "Total KEK operations (DEK wrap/unwrap), labeled by provider, operation, and result.", + }, []string{"provider", "operation", "result"}) + + kekOpDur := f.NewHistogram(prometheus.HistogramOpts{ + Name: "kek_ops_duration_secs", + Help: "Duration of KEK operations in seconds, labeled by provider and operation.", + }, []string{"provider", "operation"}) + + r := &Reporter{ + kekOps: kekOps, + kekOpDur: kekOpDur, + cacheHits: f.NewCounter(prometheus.CounterOpts{ + Name: "dek_cache_hits_total", + Help: "Total DEK cache hits.", + }, nil).WithLabelValues(), + cacheMisses: f.NewCounter(prometheus.CounterOpts{ + Name: "dek_cache_misses_total", + Help: "Total DEK cache misses.", + }, nil).WithLabelValues(), + cacheSize: f.NewGauge(prometheus.GaugeOpts{ + Name: "dek_cache_size", + Help: "Current number of entries in the DEK cache.", + }), + kekOpHandle: make(map[kekOpKey]prometheus.Counter), + kekDurHandle: make(map[kekDurKey]prometheus.Observer), + } + + providers := []string{"aws", "gcp", "azure", "testing"} + operations := []string{"wrap", "unwrap"} + results := []string{"success", "error"} + for _, p := range providers { + for _, op := range operations { + r.kekDurHandle[kekDurKey{p, op}] = kekOpDur.WithLabelValues(p, op) + for _, res := range results { + r.kekOpHandle[kekOpKey{p, op, res}] = kekOps.WithLabelValues(p, op, res) + } + } + } + + return r +} + +// KEKOp records a single KEK operation and its duration. +func (r *Reporter) KEKOp(provider, operation, result string, seconds float64) { + if c, ok := r.kekOpHandle[kekOpKey{provider, operation, result}]; ok { + c.Inc() + } else { + r.kekOps.WithLabelValues(provider, operation, result).Inc() + } + + if o, ok := r.kekDurHandle[kekDurKey{provider, operation}]; ok { + o.Observe(seconds) + } else { + r.kekOpDur.WithLabelValues(provider, operation).Observe(seconds) + } +} + +// CacheHit records a DEK cache hit and updates the cache-size gauge. +func (r *Reporter) CacheHit(e crypto.CacheEvent) { + r.cacheHits.Inc() + r.cacheSize.Set(float64(e.Size)) +} + +// CacheMiss records a DEK cache miss and updates the cache-size gauge. +func (r *Reporter) CacheMiss(e crypto.CacheEvent) { + r.cacheMisses.Inc() + r.cacheSize.Set(float64(e.Size)) +} diff --git a/internal/kms/reporter_test.go b/internal/kms/reporter_test.go new file mode 100644 index 0000000..83b45a5 --- /dev/null +++ b/internal/kms/reporter_test.go @@ -0,0 +1,86 @@ +package kms_test + +import ( + "reflect" + "testing" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + dto "github.com/prometheus/client_model/go" + "github.com/stretchr/testify/require" + + "github.com/temporalio/temporal-proxy/internal/kms" + "github.com/temporalio/temporal-proxy/internal/metrics" + "github.com/temporalio/temporal-proxy/pkg/crypto" +) + +func TestReporterKEKOp(t *testing.T) { + t.Parallel() + + r, reg := newTestReporter(t) + r.KEKOp("aws", "wrap", "success", 0.02) + + calls := gather(t, reg, "proxy_encryption_kek_ops_total") + require.NotNil(t, calls) + m := findMetric(t, calls, map[string]string{"provider": "aws", "operation": "wrap", "result": "success"}) + require.Equal(t, 1.0, m.GetCounter().GetValue()) + + dur := gather(t, reg, "proxy_encryption_kek_ops_duration_secs") + require.NotNil(t, dur) + d := findMetric(t, dur, map[string]string{"provider": "aws", "operation": "wrap"}) + require.Equal(t, uint64(1), d.GetHistogram().GetSampleCount()) +} + +func TestReporterCacheEvents(t *testing.T) { + t.Parallel() + + r, reg := newTestReporter(t) + r.CacheMiss(crypto.CacheEvent{Size: 0}) + r.CacheHit(crypto.CacheEvent{Size: 1}) + r.CacheHit(crypto.CacheEvent{Size: 2}) + + hits := gather(t, reg, "proxy_encryption_dek_cache_hits_total") + require.NotNil(t, hits) + require.Equal(t, 2.0, hits.GetMetric()[0].GetCounter().GetValue()) + + misses := gather(t, reg, "proxy_encryption_dek_cache_misses_total") + require.NotNil(t, misses) + require.Equal(t, 1.0, misses.GetMetric()[0].GetCounter().GetValue()) + + size := gather(t, reg, "proxy_encryption_dek_cache_size") + require.NotNil(t, size) + require.Equal(t, 2.0, size.GetMetric()[0].GetGauge().GetValue()) // last Set wins +} + +func gather(t *testing.T, reg *prometheus.Registry, name string) *dto.MetricFamily { + t.Helper() + mfs, err := reg.Gather() + require.NoError(t, err) + for _, mf := range mfs { + if mf.GetName() == name { + return mf + } + } + return nil +} + +func findMetric(t *testing.T, mf *dto.MetricFamily, labels map[string]string) *dto.Metric { + t.Helper() + for _, m := range mf.GetMetric() { + got := map[string]string{} + for _, l := range m.GetLabel() { + got[l.GetName()] = l.GetValue() + } + if reflect.DeepEqual(got, labels) { + return m + } + } + require.Failf(t, "metric not found", "no metric with labels %v in %s", labels, mf.GetName()) + return nil +} + +func newTestReporter(t *testing.T) (*kms.Reporter, *prometheus.Registry) { + t.Helper() + reg := prometheus.NewRegistry() + return kms.NewReporter(metrics.New("proxy", promauto.With(reg)).ForSubsystem("encryption")), reg +} diff --git a/internal/metrics/factory.go b/internal/metrics/factory.go index a8cb7e8..7c93c64 100644 --- a/internal/metrics/factory.go +++ b/internal/metrics/factory.go @@ -60,3 +60,15 @@ func (f *Factory) NewHistogram(opts prometheus.HistogramOpts, labelNames []strin return f.factory.NewHistogramVec(opts, labelNames) } + +// NewGauge creates and registers a label-less Gauge. It forces the bound +// namespace (and subsystem, when set) onto opts, overriding any the caller set, +// so every gauge shares the same prefix. +func (f *Factory) NewGauge(opts prometheus.GaugeOpts) prometheus.Gauge { + opts.Namespace = f.namespace + if f.subsystem != "" { + opts.Subsystem = f.subsystem + } + + return f.factory.NewGauge(opts) +} diff --git a/internal/metrics/factory_test.go b/internal/metrics/factory_test.go index def6320..d407288 100644 --- a/internal/metrics/factory_test.go +++ b/internal/metrics/factory_test.go @@ -7,6 +7,7 @@ import ( goprom "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promauto" "github.com/prometheus/client_golang/prometheus/testutil" + dto "github.com/prometheus/client_model/go" "github.com/stretchr/testify/require" "go.uber.org/fx" @@ -144,6 +145,29 @@ myns_sub_wait_seconds_count{kind="a"} 1 }) } +func TestNewGaugeIsNamespacedAndSubsystemed(t *testing.T) { + t.Parallel() + + reg := goprom.NewRegistry() + f := metrics.New("proxy", promauto.With(reg)).ForSubsystem("encryption") + + g := f.NewGauge(goprom.GaugeOpts{Name: "dek_cache_size", Help: "size"}) + g.Set(7) + + mfs, err := reg.Gather() + require.NoError(t, err) + + var found *dto.MetricFamily + for _, mf := range mfs { + if mf.GetName() == "proxy_encryption_dek_cache_size" { + found = mf + break + } + } + require.NotNil(t, found, "expected proxy_encryption_dek_cache_size to be registered") + require.Equal(t, 7.0, found.GetMetric()[0].GetGauge().GetValue()) +} + func TestModuleProvidesNamespacedMetrics(t *testing.T) { t.Parallel() diff --git a/internal/proxy/encryption.go b/internal/proxy/encryption.go index 972f3ea..c3f3091 100644 --- a/internal/proxy/encryption.go +++ b/internal/proxy/encryption.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "runtime" + "time" "go.temporal.io/api/common/v1" "go.temporal.io/api/proxy" @@ -41,16 +42,16 @@ type Vault interface { // never sees plaintext while local workers still exchange cleartext. On the way // back only payloads this interceptor sealed (identified by the // encryptionEncoding marker) are opened; anything else passes through -// untouched. Search attributes are skipped so they stay queryable upstream. It -// returns an error only if the underlying visitor interceptor cannot be -// constructed. -func EncryptionInterceptor(enabled bool, v Vault) (grpc.UnaryClientInterceptor, error) { +// untouched. Search attributes are skipped so they stay queryable upstream. r +// records the duration and result of every seal/open through DEKOp. It returns +// an error only if the underlying visitor interceptor cannot be constructed. +func EncryptionInterceptor(enabled bool, v Vault, r *Reporter) (grpc.UnaryClientInterceptor, error) { var outbound *proxy.VisitPayloadsOptions if enabled { outbound = &proxy.VisitPayloadsOptions{ ConcurrencyLimit: runtime.NumCPU(), SkipSearchAttributes: true, - Visitor: encryptPayloads(v), + Visitor: encryptPayloads(v, r), } } @@ -58,7 +59,7 @@ func EncryptionInterceptor(enabled bool, v Vault) (grpc.UnaryClientInterceptor, Inbound: &proxy.VisitPayloadsOptions{ ConcurrencyLimit: runtime.NumCPU(), SkipSearchAttributes: true, - Visitor: decryptPayloads(v), + Visitor: decryptPayloads(v, r), }, Outbound: outbound, }) @@ -68,8 +69,9 @@ func EncryptionInterceptor(enabled bool, v Vault) (grpc.UnaryClientInterceptor, // the bytes under v for the context's namespace, and replaces it with a payload // whose data is the ciphertext and whose metadata carries the wrapped DEK // needed to open it. The entire original payload (metadata included) is sealed, -// so decryptPayloads can restore it exactly. -func encryptPayloads(v Vault) func(*proxy.VisitPayloadsContext, []*common.Payload) ([]*common.Payload, error) { +// so decryptPayloads can restore it exactly. Each Seal call is timed and +// recorded via r.DEKOp, regardless of outcome. +func encryptPayloads(v Vault, r *Reporter) func(*proxy.VisitPayloadsContext, []*common.Payload) ([]*common.Payload, error) { return func(ctx *proxy.VisitPayloadsContext, payloads []*common.Payload) ([]*common.Payload, error) { ns := meta.NamespaceFrom(ctx) @@ -80,7 +82,9 @@ func encryptPayloads(v Vault) func(*proxy.VisitPayloadsContext, []*common.Payloa return nil, fmt.Errorf("failed to marshal payload: %w", err) } + start := time.Now() msg, err := v.Seal(ctx, ns, data) + r.DEKOp("encrypt", resultLabel(err), ns, time.Since(start).Seconds()) if err != nil { return nil, fmt.Errorf("failed to encrypt payload: %w", err) } @@ -102,9 +106,12 @@ func encryptPayloads(v Vault) func(*proxy.VisitPayloadsContext, []*common.Payloa // decryptPayloads returns a payload visitor that reverses encryptPayloads: // payloads carrying the encryptionEncoding marker are opened and unmarshaled // back into their original form, while any others pass through unchanged so -// payloads produced elsewhere survive the round trip. -func decryptPayloads(v Vault) func(*proxy.VisitPayloadsContext, []*common.Payload) ([]*common.Payload, error) { +// payloads produced elsewhere survive the round trip. Only payloads actually +// opened are timed and recorded via r.DEKOp; pass-through payloads are not. +func decryptPayloads(v Vault, r *Reporter) func(*proxy.VisitPayloadsContext, []*common.Payload) ([]*common.Payload, error) { return func(ctx *proxy.VisitPayloadsContext, payloads []*common.Payload) ([]*common.Payload, error) { + ns := meta.NamespaceFrom(ctx) + res := make([]*common.Payload, len(payloads)) for i, p := range payloads { // Only decrypt what we've encrypted @@ -113,6 +120,7 @@ func decryptPayloads(v Vault) func(*proxy.VisitPayloadsContext, []*common.Payloa continue } + start := time.Now() pt, err := v.Open(ctx, &crypto.Message{ Ciphertext: p.Data, KeyMaterial: &crypto.DEKMaterial{ @@ -120,6 +128,7 @@ func decryptPayloads(v Vault) func(*proxy.VisitPayloadsContext, []*common.Payloa EncryptedDEK: string(p.Metadata[metadataEncryptionDEK]), }, }) + r.DEKOp("decrypt", resultLabel(err), ns, time.Since(start).Seconds()) if err != nil { return nil, fmt.Errorf("failed to decrypt payload: %w", err) } @@ -135,3 +144,11 @@ func decryptPayloads(v Vault) func(*proxy.VisitPayloadsContext, []*common.Payloa return res, nil } } + +// resultLabel maps an error to the "result" metric label value. +func resultLabel(err error) string { + if err != nil { + return "error" + } + return "success" +} diff --git a/internal/proxy/encryption_test.go b/internal/proxy/encryption_test.go index 4630501..8bee213 100644 --- a/internal/proxy/encryption_test.go +++ b/internal/proxy/encryption_test.go @@ -8,13 +8,18 @@ import ( "sync" "testing" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" + dto "github.com/prometheus/client_model/go" "github.com/stretchr/testify/require" "go.temporal.io/api/common/v1" "go.temporal.io/api/proxy" workflowservice "go.temporal.io/api/workflowservice/v1" "google.golang.org/grpc" + "google.golang.org/grpc/metadata" "google.golang.org/protobuf/proto" + "github.com/temporalio/temporal-proxy/internal/metrics" "github.com/temporalio/temporal-proxy/internal/transport/meta" "github.com/temporalio/temporal-proxy/pkg/crypto" ) @@ -43,7 +48,7 @@ func TestEncryptionInterceptorRoundtrip(t *testing.T) { t.Parallel() v := &fakeVault{} - interceptor, err := EncryptionInterceptor(true, v) + interceptor, err := EncryptionInterceptor(true, v, newTestReporter(t)) require.NoError(t, err) input := &common.Payloads{Payloads: []*common.Payload{testPayload("json/plain", `"hi"`)}} @@ -77,7 +82,7 @@ func TestEncryptionInterceptorDisabledSkipsOutbound(t *testing.T) { t.Parallel() v := &fakeVault{} - interceptor, err := EncryptionInterceptor(false, v) + interceptor, err := EncryptionInterceptor(false, v, newTestReporter(t)) require.NoError(t, err) // A response payload sealed exactly as fakeVault.Seal would produce it, so @@ -121,10 +126,85 @@ func TestEncryptionInterceptorDisabledSkipsOutbound(t *testing.T) { require.Empty(t, v.namespaces, "interceptor must not seal when disabled") } +func TestEncryptionInterceptorRecordsDEKOps(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + reporter := NewReporter(metrics.New("proxy", promauto.With(reg)).ForSubsystem("encryption")) + + vault := &fakeVault{} + interceptor, err := EncryptionInterceptor(true, vault, reporter) + require.NoError(t, err) + + ctx := metadata.AppendToOutgoingContext(t.Context(), meta.NamespaceHeader, "ns1") + + req := &workflowservice.StartWorkflowExecutionRequest{ + Input: &common.Payloads{Payloads: []*common.Payload{{Data: []byte("hi")}}}, + } + resp := &workflowservice.StartWorkflowExecutionRequest{} + + // Echo the sealed request payload back so the inbound path opens it, + // exercising the decrypt metric path alongside encrypt. + invoker := func(_ context.Context, _ string, gotReq, gotResp any, _ *grpc.ClientConn, _ ...grpc.CallOption) error { + sent := gotReq.(*workflowservice.StartWorkflowExecutionRequest).Input.Payloads[0] + gotResp.(*workflowservice.StartWorkflowExecutionRequest).Input = &common.Payloads{ + Payloads: []*common.Payload{sent}, + } + return nil + } + require.NoError(t, interceptor(ctx, "/method", req, resp, nil, invoker)) + + ops := gatherFamily(t, reg, "proxy_encryption_dek_ops_total") + require.NotNil(t, ops) + require.True(t, hasLabels(ops, map[string]string{"operation": "encrypt", "result": "success", "namespace": "ns1"})) + require.True(t, hasLabels(ops, map[string]string{"operation": "decrypt", "result": "success", "namespace": "ns1"})) + + dur := gatherFamily(t, reg, "proxy_encryption_dek_ops_duration_secs") + require.NotNil(t, dur) + require.True(t, hasLabels(dur, map[string]string{"operation": "encrypt", "namespace": "ns1"})) + require.True(t, hasLabels(dur, map[string]string{"operation": "decrypt", "namespace": "ns1"})) +} + +func TestEncryptionInterceptorSkipsMetricsForPassThrough(t *testing.T) { + t.Parallel() + + reg := prometheus.NewRegistry() + reporter := NewReporter(metrics.New("proxy", promauto.With(reg)).ForSubsystem("encryption")) + + vault := &fakeVault{} + interceptor, err := EncryptionInterceptor(true, vault, reporter) + require.NoError(t, err) + + ctx := metadata.AppendToOutgoingContext(t.Context(), meta.NamespaceHeader, "ns1") + + req := &workflowservice.StartWorkflowExecutionRequest{ + Input: &common.Payloads{Payloads: []*common.Payload{{Data: []byte("hi")}}}, + } + resp := &workflowservice.StartWorkflowExecutionRequest{} + + // Return an unencrypted response payload; the inbound path must pass it + // through without opening it or recording a decrypt op. + invoker := func(_ context.Context, _ string, _, gotResp any, _ *grpc.ClientConn, _ ...grpc.CallOption) error { + gotResp.(*workflowservice.StartWorkflowExecutionRequest).Input = &common.Payloads{ + Payloads: []*common.Payload{testPayload("json/plain", `"plain"`)}, + } + return nil + } + require.NoError(t, interceptor(ctx, "/method", req, resp, nil, invoker)) + + require.Equal(t, 0, vault.opens) + + ops := gatherFamily(t, reg, "proxy_encryption_dek_ops_total") + require.NotNil(t, ops) + require.True(t, hasLabels(ops, map[string]string{"operation": "encrypt", "result": "success", "namespace": "ns1"})) + require.False(t, hasLabels(ops, map[string]string{"operation": "decrypt", "result": "success", "namespace": "ns1"})) +} + func TestEncryptDecryptPayloadsRoundtrip(t *testing.T) { t.Parallel() v := &fakeVault{} + r := newTestReporter(t) vc := visitCtx(meta.WithNamespace(t.Context(), "ns1")) original := []*common.Payload{ @@ -136,7 +216,7 @@ func TestEncryptDecryptPayloadsRoundtrip(t *testing.T) { proto.Clone(original[1]).(*common.Payload), } - sealed, err := encryptPayloads(v)(vc, original) + sealed, err := encryptPayloads(v, r)(vc, original) require.NoError(t, err) require.Len(t, sealed, len(original)) @@ -150,7 +230,7 @@ func TestEncryptDecryptPayloadsRoundtrip(t *testing.T) { require.Equal(t, []string{"ns1", "ns1"}, v.namespaces) - opened, err := decryptPayloads(v)(vc, sealed) + opened, err := decryptPayloads(v, r)(vc, sealed) require.NoError(t, err) require.Len(t, opened, len(want)) @@ -163,15 +243,16 @@ func TestDecryptPayloadsPassesThroughUnencrypted(t *testing.T) { t.Parallel() v := &fakeVault{} + r := newTestReporter(t) vc := visitCtx(meta.WithNamespace(t.Context(), "ns1")) orig := testPayload("json/plain", `"secret"`) - sealed, err := encryptPayloads(v)(vc, []*common.Payload{orig}) + sealed, err := encryptPayloads(v, r)(vc, []*common.Payload{orig}) require.NoError(t, err) plain := testPayload("json/plain", `"visible"`) - out, err := decryptPayloads(v)(vc, []*common.Payload{sealed[0], plain}) + out, err := decryptPayloads(v, r)(vc, []*common.Payload{sealed[0], plain}) require.NoError(t, err) require.Len(t, out, 2) require.True(t, proto.Equal(orig, out[0])) @@ -204,7 +285,7 @@ func TestEncryptPayloadsNamespace(t *testing.T) { t.Parallel() v := &fakeVault{} - _, err := encryptPayloads(v)(visitCtx(tc.ctx(t)), []*common.Payload{testPayload("json/plain", "x")}) + _, err := encryptPayloads(v, newTestReporter(t))(visitCtx(tc.ctx(t)), []*common.Payload{testPayload("json/plain", "x")}) require.NoError(t, err) require.Equal(t, []string{tc.want}, v.namespaces) }) @@ -218,7 +299,7 @@ func TestEncryptDecryptPayloadsErrors(t *testing.T) { t.Parallel() v := &fakeVault{sealErr: errors.New("kms unavailable")} - _, err := encryptPayloads(v)(visitCtx(t.Context()), []*common.Payload{testPayload("json/plain", "x")}) + _, err := encryptPayloads(v, newTestReporter(t))(visitCtx(t.Context()), []*common.Payload{testPayload("json/plain", "x")}) require.ErrorContains(t, err, "failed to encrypt payload") }) @@ -226,12 +307,13 @@ func TestEncryptDecryptPayloadsErrors(t *testing.T) { t.Parallel() v := &fakeVault{} + r := newTestReporter(t) vc := visitCtx(meta.WithNamespace(t.Context(), "ns1")) - sealed, err := encryptPayloads(v)(vc, []*common.Payload{testPayload("json/plain", "x")}) + sealed, err := encryptPayloads(v, r)(vc, []*common.Payload{testPayload("json/plain", "x")}) require.NoError(t, err) v.openErr = errors.New("unwrap failed") - _, err = decryptPayloads(v)(vc, sealed) + _, err = decryptPayloads(v, r)(vc, sealed) require.ErrorContains(t, err, "failed to decrypt payload") }) @@ -239,11 +321,12 @@ func TestEncryptDecryptPayloadsErrors(t *testing.T) { t.Parallel() v := &fakeVault{openReturn: []byte{0xFF, 0xFF, 0xFF}} + r := newTestReporter(t) vc := visitCtx(meta.WithNamespace(t.Context(), "ns1")) - sealed, err := encryptPayloads(v)(vc, []*common.Payload{testPayload("json/plain", "x")}) + sealed, err := encryptPayloads(v, r)(vc, []*common.Payload{testPayload("json/plain", "x")}) require.NoError(t, err) - _, err = decryptPayloads(v)(vc, sealed) + _, err = decryptPayloads(v, r)(vc, sealed) require.ErrorContains(t, err, "failed to unmarshal payload") }) } @@ -300,3 +383,56 @@ func mustMarshal(t *testing.T, p *common.Payload) []byte { func visitCtx(ctx context.Context) *proxy.VisitPayloadsContext { return &proxy.VisitPayloadsContext{Context: ctx} } + +// newTestReporter builds a Reporter backed by its own private registry, so +// call sites under test have a Reporter to record into without colliding with +// any other test's metrics. +func newTestReporter(t *testing.T) *Reporter { + t.Helper() + return NewReporter(metrics.New("test", promauto.With(prometheus.NewRegistry())).ForSubsystem("encryption")) +} + +// gatherFamily returns the metric family named name from reg, or nil if no +// such family has been registered/observed. +func gatherFamily(t *testing.T, reg *prometheus.Registry, name string) *dto.MetricFamily { + t.Helper() + + mfs, err := reg.Gather() + require.NoError(t, err) + + for _, mf := range mfs { + if mf.GetName() == name { + return mf + } + } + + return nil +} + +// hasLabels reports whether mf contains a metric whose label set matches +// labels exactly. +func hasLabels(mf *dto.MetricFamily, labels map[string]string) bool { + for _, m := range mf.GetMetric() { + got := make(map[string]string, len(m.GetLabel())) + for _, l := range m.GetLabel() { + got[l.GetName()] = l.GetValue() + } + + if len(got) != len(labels) { + continue + } + + match := true + for k, v := range labels { + if got[k] != v { + match = false + break + } + } + if match { + return true + } + } + + return false +} diff --git a/internal/proxy/fx.go b/internal/proxy/fx.go index 65d8152..927b70f 100644 --- a/internal/proxy/fx.go +++ b/internal/proxy/fx.go @@ -10,6 +10,7 @@ import ( "github.com/temporalio/temporal-proxy/internal/auth" "github.com/temporalio/temporal-proxy/internal/config" + "github.com/temporalio/temporal-proxy/internal/metrics" "github.com/temporalio/temporal-proxy/internal/protoutil" "github.com/temporalio/temporal-proxy/internal/transport/connect" "github.com/temporalio/temporal-proxy/pkg/crypto" @@ -26,6 +27,13 @@ var Module = fx.Options(fx.Invoke(func(p ProxyParams) error { return fmt.Errorf("encryption is enabled but no vault was provided") } + // Built once (not per upstream) so its collectors register with Prometheus + // exactly once; a per-upstream build would panic on duplicate registration. + var encReporter *Reporter + if p.Vault != nil { + encReporter = NewReporter(p.Factory.ForSubsystem("encryption")) + } + for i := range p.Config.Upstreams { up := &p.Config.Upstreams[i] if err := up.Validate(); err != nil { @@ -57,7 +65,7 @@ var Module = fx.Options(fx.Invoke(func(p ProxyParams) error { // interceptor, sealing outbound payloads last and opening inbound // payloads first. if p.Vault != nil { - enc, err := EncryptionInterceptor(p.Config.Encryption.Enabled, p.Vault) + enc, err := EncryptionInterceptor(p.Config.Encryption.Enabled, p.Vault, encReporter) if err != nil { return fmt.Errorf("failed to build encryption interceptor for upstream %q: %w", up.Name, err) } @@ -131,6 +139,7 @@ type ProxyParams struct { Translator *protoutil.Translator Pool *connect.Pool Vault *crypto.Vault + Factory *metrics.Factory // Optional values Logger logger.Logger `optional:"true"` diff --git a/internal/proxy/fx_test.go b/internal/proxy/fx_test.go index bc82804..4a8a76b 100644 --- a/internal/proxy/fx_test.go +++ b/internal/proxy/fx_test.go @@ -5,12 +5,15 @@ import ( "testing" "time" + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" "github.com/stretchr/testify/require" "go.uber.org/fx" "google.golang.org/grpc" "google.golang.org/grpc/health/grpc_health_v1" "github.com/temporalio/temporal-proxy/internal/config" + "github.com/temporalio/temporal-proxy/internal/metrics" "github.com/temporalio/temporal-proxy/internal/protoutil" "github.com/temporalio/temporal-proxy/internal/proxy" "github.com/temporalio/temporal-proxy/internal/transport/connect" @@ -233,6 +236,7 @@ func TestModuleRequiresTranslator(t *testing.T) { Upstreams: []config.Upstream{{Name: "primary", Listen: config.ListenConfig{HostPort: "127.0.0.1:47242"}}}, }), fx.Provide(func() *crypto.Vault { return nil }), + fx.Provide(func() *metrics.Factory { return metrics.New("tmprl_proxy", promauto.With(prometheus.NewRegistry())) }), connect.Module, proxy.Module, fx.NopLogger, @@ -251,6 +255,7 @@ func newProxyApp(t *testing.T, cfg *config.Config, opts ...fx.Option) *fx.App { // case, and the proxy skips the encryption interceptor. Tests that // exercise encryption override this with a real vault. fx.Provide(func() *crypto.Vault { return nil }), + fx.Provide(func() *metrics.Factory { return metrics.New("tmprl_proxy", promauto.With(prometheus.NewRegistry())) }), connect.Module, protoutil.Module, proxy.Module, diff --git a/internal/proxy/reporter.go b/internal/proxy/reporter.go new file mode 100644 index 0000000..e1b235d --- /dev/null +++ b/internal/proxy/reporter.go @@ -0,0 +1,38 @@ +package proxy + +import ( + "github.com/prometheus/client_golang/prometheus" + + "github.com/temporalio/temporal-proxy/internal/metrics" +) + +// Reporter records DEK payload-operation telemetry to Prometheus: each seal +// (encrypt) and open (decrypt) the encryption interceptor performs. The +// namespace label is unbounded, so handles are resolved per call via +// WithLabelValues rather than pre-computed. A Reporter is safe for concurrent +// use. +type Reporter struct { + ops *prometheus.CounterVec + duration *prometheus.HistogramVec +} + +// NewReporter builds the Prometheus-backed DEK-operation Reporter. f must +// already be scoped to the "encryption" subsystem by the caller. +func NewReporter(f *metrics.Factory) *Reporter { + return &Reporter{ + ops: f.NewCounter(prometheus.CounterOpts{ + Name: "dek_ops_total", + Help: "Total DEK payload operations (encrypt/decrypt), labeled by operation, result, and namespace.", + }, []string{"operation", "result", "namespace"}), + duration: f.NewHistogram(prometheus.HistogramOpts{ + Name: "dek_ops_duration_secs", + Help: "Duration of DEK payload operations in seconds, labeled by operation and namespace.", + }, []string{"operation", "namespace"}), + } +} + +// DEKOp records a single DEK payload operation and its duration. +func (r *Reporter) DEKOp(operation, result, namespace string, seconds float64) { + r.ops.WithLabelValues(operation, result, namespace).Inc() + r.duration.WithLabelValues(operation, namespace).Observe(seconds) +} diff --git a/pkg/crypto/observer.go b/pkg/crypto/observer.go new file mode 100644 index 0000000..bf8b3ed --- /dev/null +++ b/pkg/crypto/observer.go @@ -0,0 +1,25 @@ +package crypto + +type ( + // CacheEvent describes a DEK cache access. Fields may be added over time + // without breaking implementers, so callers accept the struct by value. + CacheEvent struct { + // Size is the number of entries in the DEK cache after the access. + Size int + } + + // Observer receives notifications about Vault-internal events for telemetry. + // Implementations must be safe for concurrent use and must not block: a + // Vault calls these on the Open path. A nil Observer is never used; the + // Vault substitutes a no-op (see WithObserver). + Observer interface { + CacheHit(CacheEvent) + CacheMiss(CacheEvent) + } + + // nopObserver is the default Observer; it drops every event. + nopObserver struct{} +) + +func (nopObserver) CacheHit(CacheEvent) {} +func (nopObserver) CacheMiss(CacheEvent) {} diff --git a/pkg/crypto/observer_test.go b/pkg/crypto/observer_test.go new file mode 100644 index 0000000..4d7369f --- /dev/null +++ b/pkg/crypto/observer_test.go @@ -0,0 +1,81 @@ +package crypto_test + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" + + "github.com/temporalio/temporal-proxy/pkg/crypto" +) + +type spyObserver struct { + hits []int + misses []int +} + +func TestObserverMissThenHit(t *testing.T) { + t.Parallel() + + spy := &spyObserver{} + v := newTestVault(t, crypto.WithObserver(spy)) + + msg, err := v.Seal(t.Context(), "ns1", []byte("hello")) + require.NoError(t, err) + + // First open: cache empty -> miss (size before add is 0). + pt, err := v.Open(t.Context(), msg) + require.NoError(t, err) + require.Equal(t, []byte("hello"), pt) + + // Second open of the same message: served from cache -> hit (size 1). + _, err = v.Open(t.Context(), msg) + require.NoError(t, err) + + require.Equal(t, []int{0}, spy.misses) + require.Equal(t, []int{1}, spy.hits) +} + +func TestObserverDisabledCache(t *testing.T) { + t.Parallel() + + spy := &spyObserver{} + v := newTestVault(t, crypto.WithObserver(spy), crypto.WithCacheSize(0)) + + msg, err := v.Seal(t.Context(), "ns1", []byte("hello")) + require.NoError(t, err) + _, err = v.Open(t.Context(), msg) + require.NoError(t, err) + _, err = v.Open(t.Context(), msg) + require.NoError(t, err) + + require.Empty(t, spy.hits) + require.Empty(t, spy.misses) +} + +func TestObserverDefaultNoPanic(t *testing.T) { + t.Parallel() + + v := newTestVault(t) // no WithObserver + + msg, err := v.Seal(t.Context(), "ns1", []byte("hello")) + require.NoError(t, err) + _, err = v.Open(t.Context(), msg) + require.NoError(t, err) // must not panic on the nil-default path +} + +func (s *spyObserver) CacheHit(e crypto.CacheEvent) { s.hits = append(s.hits, e.Size) } +func (s *spyObserver) CacheMiss(e crypto.CacheEvent) { s.misses = append(s.misses, e.Size) } + +func newTestVault(t *testing.T, opts ...crypto.VaultOption) *crypto.Vault { + t.Helper() + reg, err := crypto.NewKEKRegistry(crypto.WithDefaultKey(&fakeKEK{id: "k1"})) + require.NoError(t, err) + + base := []crypto.VaultOption{ + crypto.WithDefaultKeyConfig(crypto.KeyConfig{Duration: time.Hour, RenewBefore: time.Minute}), + } + v, err := crypto.NewVault(reg, append(base, opts...)...) + require.NoError(t, err) + return v +} diff --git a/pkg/crypto/vault.go b/pkg/crypto/vault.go index 38875e8..ffd71b6 100644 --- a/pkg/crypto/vault.go +++ b/pkg/crypto/vault.go @@ -25,6 +25,7 @@ type ( cache *lru.Cache[string, *DEK] defaultConfig *KeyConfig nowFn func() time.Time + observer Observer } // NamespacedVault is a [Vault] bound to a single namespace so callers can @@ -59,6 +60,7 @@ type ( defaultConfig *KeyConfig nowFn func() time.Time cacheSize int + observer Observer errs []error } @@ -84,6 +86,7 @@ func NewVault(r *KEKRegistry, opts ...VaultOption) (*Vault, error) { cacheSize: 100, config: make(map[string]KeyConfig), nowFn: time.Now, + observer: nopObserver{}, } for _, opt := range opts { @@ -99,6 +102,7 @@ func NewVault(r *KEKRegistry, opts ...VaultOption) (*Vault, error) { keys: make(map[string]*slidingDEK), defaultConfig: vopts.defaultConfig, nowFn: vopts.nowFn, + observer: vopts.observer, } if vopts.cacheSize > 0 { @@ -182,6 +186,19 @@ func WithCacheSize(n int) VaultOption { } } +// WithObserver sets the Observer notified of Vault-internal events such as DEK +// cache hits and misses. A nil Observer is replaced with a no-op, so Open never +// needs to nil-check. Without this option no events are emitted. +func WithObserver(o Observer) VaultOption { + return func(e *vaultOptions) { + if o == nil { + o = nopObserver{} + } + + e.observer = o + } +} + // Seal encrypts data for ns, returning the ciphertext together with the wrapped // DEK required to Open it. The active DEK for ns is created or rotated on demand. // Concurrent first-time seals holding the same DEK are coalesced into a single @@ -257,6 +274,9 @@ func (v *Vault) Open(ctx context.Context, msg *Message) ([]byte, error) { if v.cache != nil { if cached, ok := v.cache.Get(msg.KeyMaterial.EncryptedDEK); ok { dek = cached + v.observer.CacheHit(CacheEvent{Size: v.cache.Len()}) + } else { + v.observer.CacheMiss(CacheEvent{Size: v.cache.Len()}) } }