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()}) } }