parse SSEKMSKeyId from store url and pass to store config (#323)

broker/fragment/store_s3.go

@@ -41,6 +41,9 @@ type S3StoreConfig struct {
 	// SSE is the server-side encryption type to be applied (eg, "AES256").
 	// By default, encryption is not used.
 	SSE string
+	// SSEKMSKeyId specifies the ID for the AWS KMS symmetric customer managed key
+	// By default, not used.
+	SSEKMSKeyId string
 }
 
 type s3Backend struct {
@@ -128,6 +131,9 @@ func (s *s3Backend) Persist(ctx context.Context, ep *url.URL, spool Spool) error
 	if cfg.SSE != "" {
 		putObj.ServerSideEncryption = aws.String(cfg.SSE)
 	}
+	if cfg.SSEKMSKeyId != "" {
+		putObj.SSEKMSKeyId = aws.String(cfg.SSEKMSKeyId)
+	}
 	if spool.CompressionCodec == pb.CompressionCodec_GZIP_OFFLOAD_DECOMPRESSION {
 		putObj.ContentEncoding = aws.String("gzip")
 	}

broker/fragment/stores_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"flag"
 	"io/ioutil"
+	"net/url"
 	"os"
 	"sort"
 	"testing"
@@ -124,6 +125,17 @@ func TestStoreInteractions(t *testing.T) {
 		func(f pb.Fragment) { panic("not called") }))
 }
 
+func TestParseStoreArgsS3(t *testing.T) {
+	storeURL, _ := url.Parse("s3://bucket/prefix/?endpoint=https://s3.region.amazonaws.com&SSE=kms&SSEKMSKeyId=123")
+	var s3Cfg S3StoreConfig
+	parseStoreArgs(storeURL, &s3Cfg)
+	require.Equal(t, "bucket", storeURL.Host)
+	require.Equal(t, "prefix/", storeURL.Path[1:])
+	require.Equal(t, "https://s3.region.amazonaws.com", s3Cfg.Endpoint)
+	require.Equal(t, "kms", s3Cfg.SSE)
+	require.Equal(t, "123", s3Cfg.SSEKMSKeyId)
+}
+
 func readFrag(t *testing.T, f pb.Fragment) string {
 	var rc, err = Open(context.Background(), f)
 	require.NoError(t, err)