Merge pull request #64481 from immutableT/transormer-metrics-2

Automatic merge from submit-queue (batch tested with PRs 64481, 64569). If you want to cherry-pick this change to another branch, please follow the instructions <a href="https://github.com/kubernetes/community/blob/master/contributors/devel/cherry-picks.md">here</a>.

Instrument envelop transformer.

**What this PR does / why we need it**:
Add metrics for envelope transformer: 
   transformation_operation_count
   transformation_failures_count
   envelope_transformation_cache_misses_count
   data_key_generation_latencies_microseconds
   data_key_generation_failures_count

**Which issue(s) this PR fixes** *(optional, in `fixes #<issue number>(, fixes #<issue_number>, ...)` format, will close the issue(s) when PR gets merged)*:
Fixes #

**Special notes for your reviewer**:

**Release note**:

```release-note
NONE
```

staging/src/k8s.io/apiserver/pkg/storage/value/encrypt/envelope/envelope.go

@@ -24,6 +24,7 @@ import (
 	"encoding/base64"
 	"encoding/binary"
 	"fmt"
+	"time"
 
 	"k8s.io/apiserver/pkg/storage/value"
 
@@ -33,6 +34,10 @@ import (
 // defaultCacheSize is the number of decrypted DEKs which would be cached by the transformer.
 const defaultCacheSize = 1000
 
+func init() {
+	value.RegisterMetrics()
+}
+
 // Service allows encrypting and decrypting data using an external Key Management Service.
 type Service interface {
 	// Decrypt a given bytearray to obtain the original data as bytes.
@@ -85,6 +90,7 @@ func (t *envelopeTransformer) TransformFromStorage(data []byte, context value.Co
 	// Look up the decrypted DEK from cache or Envelope.
 	transformer := t.getTransformer(encKey)
 	if transformer == nil {
+		value.RecordCacheMiss()
 		key, err := t.envelopeService.Decrypt(encKey)
 		if err != nil {
 			return nil, false, fmt.Errorf("error while decrypting key: %q", err)
@@ -156,10 +162,12 @@ func (t *envelopeTransformer) getTransformer(encKey []byte) value.Transformer {
 }
 
 // generateKey generates a random key using system randomness.
-func generateKey(length int) ([]byte, error) {
-	key := make([]byte, length)
-	_, err := rand.Read(key)
-	if err != nil {
+func generateKey(length int) (key []byte, err error) {
+	defer func(start time.Time) {
+		value.RecordDataKeyGeneration(start, err)
+	}(time.Now())
+	key = make([]byte, length)
+	if _, err = rand.Read(key); err != nil {
 		return nil, err
 	}
 

staging/src/k8s.io/apiserver/pkg/storage/value/metrics.go

@@ -23,11 +23,16 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 )
 
+const (
+	namespace = "apiserver"
+	subsystem = "storage"
+)
+
 var (
 	transformerLatencies = prometheus.NewHistogramVec(
 		prometheus.HistogramOpts{
-			Namespace: "apiserver",
-			Subsystem: "storage",
+			Namespace: namespace,
+			Subsystem: subsystem,
 			Name:      "transformation_latencies_microseconds",
 			Help:      "Latencies in microseconds of value transformation operations.",
 			// In-process transformations (ex. AES CBC) complete on the order of 20 microseconds. However, when
@@ -36,6 +41,42 @@ var (
 		},
 		[]string{"transformation_type"},
 	)
+	transformerFailuresTotal = prometheus.NewCounterVec(
+		prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "transformation_failures_total",
+			Help:      "Total number of failed transformation operations.",
+		},
+		[]string{"transformation_type"},
+	)
+
+	envelopeTransformationCacheMissTotal = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "envelope_transformation_cache_misses_total",
+			Help:      "Total number of cache misses while accessing key decryption key(KEK).",
+		},
+	)
+
+	dataKeyGenerationLatencies = prometheus.NewHistogram(
+		prometheus.HistogramOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "data_key_generation_latencies_microseconds",
+			Help:      "Latencies in microseconds of data encryption key(DEK) generation operations.",
+			Buckets:   prometheus.ExponentialBuckets(5, 2, 14),
+		},
+	)
+	dataKeyGenerationFailuresTotal = prometheus.NewCounter(
+		prometheus.CounterOpts{
+			Namespace: namespace,
+			Subsystem: subsystem,
+			Name:      "data_key_generation_failures_total",
+			Help:      "Total number of failed data encryption key(DEK) generation operations.",
+		},
+	)
 )
 
 var registerMetrics sync.Once
@@ -43,14 +84,40 @@ var registerMetrics sync.Once
 func RegisterMetrics() {
 	registerMetrics.Do(func() {
 		prometheus.MustRegister(transformerLatencies)
+		prometheus.MustRegister(transformerFailuresTotal)
+		prometheus.MustRegister(envelopeTransformationCacheMissTotal)
+		prometheus.MustRegister(dataKeyGenerationLatencies)
+		prometheus.MustRegister(dataKeyGenerationFailuresTotal)
 	})
 }
 
-func RecordTransformation(transformationType string, start time.Time) {
+// RecordTransformation records latencies and count of TransformFromStorage and TransformToStorage operations.
+func RecordTransformation(transformationType string, start time.Time, err error) {
+	if err != nil {
+		transformerFailuresTotal.WithLabelValues(transformationType).Inc()
+		return
+	}
+
 	since := sinceInMicroseconds(start)
 	transformerLatencies.WithLabelValues(transformationType).Observe(float64(since))
 }
 
+// RecordCacheMiss records a miss on Key Encryption Key(KEK) - call to KMS was required to decrypt KEK.
+func RecordCacheMiss() {
+	envelopeTransformationCacheMissTotal.Inc()
+}
+
+// RecordDataKeyGeneration records latencies and count of Data Encryption Key generation operations.
+func RecordDataKeyGeneration(start time.Time, err error) {
+	if err != nil {
+		dataKeyGenerationFailuresTotal.Inc()
+		return
+	}
+
+	since := sinceInMicroseconds(start)
+	dataKeyGenerationLatencies.Observe(float64(since))
+}
+
 func sinceInMicroseconds(start time.Time) int64 {
 	elapsedNanoseconds := time.Since(start).Nanoseconds()
 	return elapsedNanoseconds / int64(time.Microsecond)

staging/src/k8s.io/apiserver/pkg/storage/value/transformer.go

@@ -85,14 +85,18 @@ func (t *MutableTransformer) Set(transformer Transformer) {
 }
 
 func (t *MutableTransformer) TransformFromStorage(data []byte, context Context) (out []byte, stale bool, err error) {
-	defer RecordTransformation("from_storage", time.Now())
+	defer func(start time.Time) {
+		RecordTransformation("from_storage", start, err)
+	}(time.Now())
 	t.lock.RLock()
 	transformer := t.transformer
 	t.lock.RUnlock()
 	return transformer.TransformFromStorage(data, context)
 }
 func (t *MutableTransformer) TransformToStorage(data []byte, context Context) (out []byte, err error) {
-	defer RecordTransformation("to_storage", time.Now())
+	defer func(start time.Time) {
+		RecordTransformation("to_storage", start, err)
+	}(time.Now())
 	t.lock.RLock()
 	transformer := t.transformer
 	t.lock.RUnlock()

test/integration/master/kms_transformation_test.go

@@ -144,6 +144,7 @@ func TestKMSProvider(t *testing.T) {
 	if secretVal != string(s.Data[secretKey]) {
 		t.Fatalf("expected %s from KubeAPI, but got %s", secretVal, string(s.Data[secretKey]))
 	}
+	test.printMetrics()
 }
 
 func getDEKFromKMSPlugin(pluginMock *base64Plugin) ([]byte, error) {

test/integration/master/transformation_testcase.go

@@ -48,7 +48,7 @@ const (
 	encryptionConfigFileName = "encryption.conf"
 	testNamespace            = "secret-encryption-test"
 	testSecret               = "test-secret"
-	latencySummaryMetricsFamily = "apiserver_storage_transformation_latencies_microseconds"
+	metricsPrefix            = "apiserver_storage_"
 )
 
 type unSealSecret func(cipherText []byte, ctx value.Context, config encryptionconfig.ProviderConfig) ([]byte, error)
@@ -247,9 +247,9 @@ func (e *transformTest) printMetrics() error {
 		return fmt.Errorf("failed to gather metrics: %s", err)
 	}
 
-	metricsOfInterest := []string{latencySummaryMetricsFamily}
 	for _, mf := range metrics {
-		if contains(metricsOfInterest, *mf.Name) {
+		if strings.HasPrefix(*mf.Name, metricsPrefix) {
+			e.logger.Logf("%s", *mf.Name)
 			for _, metric := range mf.GetMetric() {
 				e.logger.Logf("%v", metric)
 			}