fix(s3): Fix compactor copy metadata via sse (#5774)

Author: steffsas

CHANGELOG.md

@@ -13,6 +13,7 @@
 * [ENHANCEMENT] Unsafe search hints for frontend performance tests [#5723](https://github.com/grafana/tempo/pull/5723) (@ruslan-mikhailov)
 * [ENHANCEMENT] Add new livestore alert to the tempo-mixin [#5752](https://github.com/grafana/tempo/pull/5752) (@javiermolinar)
 * [ENHANCEMENT] Improve shutdown time in the first 30 seconds [#5725](https://github.com/grafana/tempo/pull/5725) (@ldufr)
+* [BUGFIX] Fix compactor to properly consider SSE-KMS information during metadata copy [#5774](https://github.com/grafana/tempo/pull/5774) (@steffsas)
 * [BUGFIX] Correctly track and reject too large traces in live stores. [#5757](https://github.com/grafana/tempo/pull/5757) (@joe-elliott)
 * [BUGFIX] Fix issues related to integer dedicated columns in vParquet5-preview2 [#5716](https://github.com/grafana/tempo/pull/5716) (@stoewer)
 * [FEATURE] Added validation mode and tests for tempo-vulture [#5605](https://github.com/grafana/tempo/pull/5605)

tempodb/backend/s3/compactor.go

@@ -21,18 +21,23 @@ func (rw *readerWriter) MarkBlockCompacted(blockID uuid.UUID, tenantID string) e
 	}
 
 	putObjectOptions := getPutObjectOptions(rw)
-
 	metaFileName := backend.MetaFileName(blockID, tenantID, rw.cfg.Prefix)
+
 	// copy meta.json to meta.compacted.json
-	_, err := rw.core.CopyObject(
-		context.TODO(),
-		rw.cfg.Bucket,
-		metaFileName,
-		rw.cfg.Bucket,
-		backend.CompactedMetaFileName(blockID, tenantID, rw.cfg.Prefix),
-		nil,
-		minio.CopySrcOptions{},
-		putObjectOptions,
+	// core.CopyObject does not support SSE on src object
+	_, err := rw.core.Client.CopyObject(context.TODO(),
+		minio.CopyDestOptions{
+			Bucket:     rw.cfg.Bucket,
+			Object:     backend.CompactedMetaFileName(blockID, tenantID, rw.cfg.Prefix),
+			Encryption: putObjectOptions.ServerSideEncryption,
+			UserTags:   putObjectOptions.UserTags,
+			// to set X-Amz-Tagging header, we need to set this flag if tags are present
+			ReplaceTags: len(putObjectOptions.UserTags) > 0,
+		}, minio.CopySrcOptions{
+			Bucket:     rw.cfg.Bucket,
+			Object:     metaFileName,
+			Encryption: putObjectOptions.ServerSideEncryption,
+		},
 	)
 	if err != nil {
 		return fmt.Errorf("error copying obj meta to compacted obj meta: %w", err)

tempodb/backend/s3/compactor_test.go

@@ -0,0 +1,108 @@
+package s3
+
+import (
+	"encoding/base64"
+	"net/http"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/grafana/dskit/flagext"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestMarkBlockCompacted(t *testing.T) {
+	sseConfig := SSEConfig{
+		Type:                 SSEKMS,
+		KMSKeyID:             "my-kms-key-id",
+		KMSEncryptionContext: "{}",
+	}
+
+	tags := map[string]string{"env": "prod", "app": "thing"}
+
+	testedHeaders := []string{
+		sseHeader,
+		sseKMSKeyIDHeader,
+		sseKMSContextHeader,
+		tagHeader,
+	}
+
+	tests := []struct {
+		name                 string
+		tags                 map[string]string
+		sse                  SSEConfig
+		expectedHeaderValues map[string]string
+	}{
+		{
+			"sse and tags",
+			tags,
+			sseConfig,
+			map[string]string{
+				sseHeader:           "aws:kms",
+				sseKMSKeyIDHeader:   sseConfig.KMSKeyID,
+				sseKMSContextHeader: base64.StdEncoding.EncodeToString([]byte(sseConfig.KMSEncryptionContext)),
+				tagHeader:           "app=thing&env=prod",
+			},
+		},
+		{
+			"tags without sse",
+			tags,
+			SSEConfig{},
+			map[string]string{
+				tagHeader: "app=thing&env=prod",
+			},
+		},
+		{
+			"sse without tags",
+			nil,
+			sseConfig,
+			map[string]string{
+				sseHeader:           "aws:kms",
+				sseKMSKeyIDHeader:   sseConfig.KMSKeyID,
+				sseKMSContextHeader: base64.StdEncoding.EncodeToString([]byte(sseConfig.KMSEncryptionContext)),
+			},
+		},
+		{
+			"no sse or tag headers",
+			nil,
+			SSEConfig{},
+			map[string]string{},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			// rawObject := raw.Object{}
+			var httpHeader http.Header
+
+			server := fakeServerWithHeader(t, &httpHeader)
+			_, _, c, err := New(&Config{
+				Region:    "blerg",
+				AccessKey: "test",
+				SecretKey: flagext.SecretWithValue("test"),
+				Bucket:    "blerg",
+				Insecure:  true,
+				Endpoint:  server.URL[7:], // [7:] -> strip http://
+				SSE:       tc.sse,
+				Tags:      tc.tags,
+			})
+			require.NoError(t, err)
+
+			_ = c.MarkBlockCompacted(uuid.New(), "tenant1")
+
+			// check expected headers to be set with expected values
+			for headerKey, expectedHeaderValue := range tc.expectedHeaderValues {
+				headerValue := httpHeader.Get(headerKey)
+				assert.Equal(t, expectedHeaderValue, headerValue, "expected header %s to have value %s", headerKey, expectedHeaderValue)
+			}
+
+			// check no unexpected headers are set
+			for _, testedHeader := range testedHeaders {
+				_, ok := tc.expectedHeaderValues[testedHeader]
+				if !ok {
+					require.Empty(t, httpHeader.Get(testedHeader), "expected header %s to be empty", testedHeader)
+				}
+			}
+		})
+	}
+}

tempodb/backend/s3/s3_test.go

@@ -31,10 +31,13 @@ import (
 )
 
 const (
-	getMethod          = "GET"
-	putMethod          = "PUT"
-	tagHeader          = "X-Amz-Tagging"
-	storageClassHeader = "X-Amz-Storage-Class"
+	getMethod           = "GET"
+	putMethod           = "PUT"
+	tagHeader           = "X-Amz-Tagging"
+	storageClassHeader  = "X-Amz-Storage-Class"
+	sseHeader           = "X-Amz-Server-Side-Encryption"
+	sseKMSKeyIDHeader   = "X-Amz-Server-Side-Encryption-Aws-Kms-Key-Id"
+	sseKMSContextHeader = "X-Amz-Server-Side-Encryption-Context"
 
 	defaultAccessKey = "AKIAIOSFODNN7EXAMPLE"
 	defaultSecretKey = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
@@ -334,21 +337,13 @@ func TestReadError(t *testing.T) {
 	assert.Equal(t, wups, errB)
 }
 
-func fakeServerWithHeader(t *testing.T, obj *url.Values, testedHeaderName string) *httptest.Server {
-	require.NotNil(t, obj)
+func fakeServerWithHeader(t *testing.T, httpHeader *http.Header) *httptest.Server {
+	require.NotNil(t, httpHeader)
 
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		switch method := r.Method; method {
 		case putMethod:
-			// https://docs.aws.amazon.com/AmazonS3/latest/API/API_PutObject.html
-			switch testedHeaderValue := r.Header.Get(testedHeaderName); testedHeaderValue {
-			case "":
-			default:
-
-				value, err := url.ParseQuery(testedHeaderValue)
-				require.NoError(t, err)
-				*obj = value
-			}
+			*httpHeader = r.Header
 		case getMethod:
 			// return fake list response b/c it's the only call that has to succeed
 			_, _ = w.Write([]byte(`<?xml version="1.0" encoding="UTF-8"?>
@@ -375,9 +370,9 @@ func TestObjectBlockTags(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			// rawObject := raw.Object{}
-			var obj url.Values
+			var httpHeaders http.Header
 
-			server := fakeServerWithHeader(t, &obj, tagHeader)
+			server := fakeServerWithHeader(t, &httpHeaders)
 			_, w, _, err := New(&Config{
 				Region:    "blerg",
 				AccessKey: "test",
@@ -392,8 +387,13 @@ func TestObjectBlockTags(t *testing.T) {
 			ctx := context.Background()
 			_ = w.Write(ctx, "object", backend.KeyPath{"test"}, bytes.NewReader([]byte{}), 0, nil)
 
+			testedHeaderValue := httpHeaders.Get(tagHeader)
+			require.NotEmpty(t, testedHeaderValue)
+			headerValue, err := url.ParseQuery(testedHeaderValue)
+			require.NoError(t, err)
+
 			for k, v := range tc.tags {
-				vv := obj.Get(k)
+				vv := headerValue.Get(k)
 				require.NotEmpty(t, vv)
 				require.Equal(t, v, vv)
 			}
@@ -604,9 +604,9 @@ func TestObjectStorageClass(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			// rawObject := raw.Object{}
-			var obj url.Values
+			var httpHeader http.Header
 
-			server := fakeServerWithHeader(t, &obj, storageClassHeader)
+			server := fakeServerWithHeader(t, &httpHeader)
 			_, w, _, err := New(&Config{
 				Region:       "blerg",
 				AccessKey:    "test",
@@ -620,7 +620,7 @@ func TestObjectStorageClass(t *testing.T) {
 
 			ctx := context.Background()
 			_ = w.Write(ctx, "object", backend.KeyPath{"test"}, bytes.NewReader([]byte{}), 0, nil)
-			require.Equal(t, obj.Has(tc.StorageClass), true)
+			require.Equal(t, tc.StorageClass, httpHeader.Get(storageClassHeader))
 		})
 	}
 }