add support for AWS SSM parameterstore

open-telemetry#36383

confmap/provider/parameterstoreprovider/Makefile

@@ -0,0 +1 @@
+include ../../../Makefile.Common

confmap/provider/parameterstoreprovider/README.md

@@ -0,0 +1,12 @@
+## Summary
+This package provides a `ConfigMapProvider` implementation for AWS SSM ParameterStore (`parameterstore`) that gives 
+the Collector the ability to read data stored in AWS SSM ParameterStore.
+## How it works
+- Just use the placeholders with the following pattern `${parameterstore:<arn or name>}`
+- Make sure you have the `ssm:GetParameter` in the OTEL Collector Role
+- If your parameter is a json string, you can get the value for a json key using the following pattern `${parameterstore:<arn or name>#json-key}`
+- If your parameter is a SecureString, you can enable decryption of the value using following pattern `${parameterstore:<arn or name>?withDecryption=true}`
+
+Prerequisites:
+- Need to set up access keys from IAM console (aws_access_key_id and aws_secret_access_key) with permission to access AWS SSM ParameterStore
+- For details, can take a look at https://aws.github.io/aws-sdk-go-v2/docs/configuring-sdk/

confmap/provider/parameterstoreprovider/go.mod

@@ -0,0 +1,36 @@
+module github.com/open-telemetry/opentelemetry-collector-contrib/confmap/provider/parameterstoreprovider
+
+go 1.22.0
+
+require (
+	github.com/aws/aws-sdk-go-v2 v1.32.4
+	github.com/aws/aws-sdk-go-v2/config v1.28.3
+	github.com/aws/aws-sdk-go-v2/service/ssm v1.55.5
+	github.com/stretchr/testify v1.9.0
+	go.opentelemetry.io/collector/confmap v1.19.0
+)
+
+require (
+	github.com/aws/aws-sdk-go-v2/credentials v1.17.44 // indirect
+	github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.16.19 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/configsources v1.3.23 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.6.23 // indirect
+	github.com/aws/aws-sdk-go-v2/internal/ini v1.8.1 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.12.0 // indirect
+	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.12.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sso v1.24.5 // indirect
+	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.28.4 // indirect
+	github.com/aws/aws-sdk-go-v2/service/sts v1.32.4 // indirect
+	github.com/aws/smithy-go v1.22.0 // indirect
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/go-viper/mapstructure/v2 v2.2.1 // indirect
+	github.com/jmespath/go-jmespath v0.4.0 // indirect
+	github.com/knadh/koanf v1.5.0 // indirect
+	github.com/knadh/koanf/v2 v2.1.1 // indirect
+	github.com/mitchellh/copystructure v1.2.0 // indirect
+	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)

confmap/provider/parameterstoreprovider/metadata.yaml

@@ -0,0 +1,3 @@
+status:
+  codeowners:
+    active: [its-felix]

confmap/provider/parameterstoreprovider/provider.go

@@ -0,0 +1,122 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package parameterstoreprovider // import "github.com/open-telemetry/opentelemetry-collector-contrib/confmap/provider/parameterstoreprovider"
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/url"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/service/ssm"
+	"go.opentelemetry.io/collector/confmap"
+)
+
+type ssmClient interface {
+	GetParameter(ctx context.Context, params *ssm.GetParameterInput, optFns ...func(*ssm.Options)) (*ssm.GetParameterOutput, error)
+}
+
+const (
+	schemeName = "parameterstore"
+)
+
+type provider struct {
+	client ssmClient
+}
+
+// NewFactory returns a new confmap.ProviderFactory that creates a confmap.Provider
+// which reads configuration using the given AWS SSM ParameterStore Name or ARN.
+//
+// This Provider supports "parameterstore" scheme, and can be called with a selector:
+// `parameterstore:NAME_OR_ARN`
+func NewFactory() confmap.ProviderFactory {
+	return confmap.NewProviderFactory(newWithSettings)
+}
+
+func newWithSettings(_ confmap.ProviderSettings) confmap.Provider {
+	return &provider{client: nil}
+}
+
+func (provider *provider) Retrieve(ctx context.Context, rawURI string, _ confmap.WatcherFunc) (*confmap.Retrieved, error) {
+	uri, err := url.Parse(rawURI)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse uri %q: %w", rawURI, err)
+	}
+
+	if uri.Scheme != schemeName {
+		return nil, fmt.Errorf("%q uri is not supported by %q provider", rawURI, schemeName)
+	}
+
+	if err = provider.ensureClient(ctx); err != nil {
+		return nil, err
+	}
+
+	// extract relevant query and fragment values
+	jsonField := uri.EscapedFragment()
+	withDecryption := uri.Query().Get("withDecryption") == "true"
+
+	// reset scheme/query/fragment
+	uri.Scheme = ""
+	uri.RawQuery = ""
+	uri.Fragment = ""
+	uri.RawFragment = ""
+
+	parameterName := uri.String()
+
+	req := &ssm.GetParameterInput{
+		Name:           aws.String(parameterName),
+		WithDecryption: aws.Bool(withDecryption),
+	}
+
+	response, err := provider.client.GetParameter(ctx, req)
+	if err != nil {
+		return nil, fmt.Errorf("error getting parameter: %w", err)
+	}
+
+	if jsonField != "" {
+		return provider.retrieveJSONField(*response.Parameter.Value, jsonField)
+	}
+
+	return confmap.NewRetrieved(*response.Parameter.Value)
+}
+
+func (provider *provider) ensureClient(ctx context.Context) error {
+	// initialize the ssm client in the first call of Retrieve
+	if provider.client == nil {
+		cfg, err := config.LoadDefaultConfig(ctx)
+
+		if err != nil {
+			return fmt.Errorf("failed to load configurations to initialize an AWS SDK client, error: %w", err)
+		}
+
+		provider.client = ssm.NewFromConfig(cfg)
+	}
+
+	return nil
+}
+
+func (*provider) retrieveJSONField(rawJSON, field string) (*confmap.Retrieved, error) {
+	var fieldsMap map[string]any
+	err := json.Unmarshal([]byte(rawJSON), &fieldsMap)
+	if err != nil {
+		return nil, fmt.Errorf("error unmarshalling parameter string: %w", err)
+	}
+
+	fieldValue, ok := fieldsMap[field]
+	if !ok {
+		return nil, fmt.Errorf("field %q not found in fields map", field)
+	}
+
+	return confmap.NewRetrieved(fieldValue)
+}
+
+func (*provider) Scheme() string {
+	return schemeName
+}
+
+func (*provider) Shutdown(context.Context) error {
+	return nil
+}

confmap/provider/parameterstoreprovider/provider_test.go

@@ -0,0 +1,184 @@
+// Copyright The OpenTelemetry Authors
+// SPDX-License-Identifier: Apache-2.0
+
+package parameterstoreprovider
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/ssm"
+	"github.com/aws/aws-sdk-go-v2/service/ssm/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.opentelemetry.io/collector/confmap"
+)
+
+// Mock AWS SSM
+type testSSMClient struct {
+	name      string
+	value     string
+	encrypted bool
+}
+
+// Implement GetParameter()
+func (client *testSSMClient) GetParameter(_ context.Context, params *ssm.GetParameterInput, _ ...func(*ssm.Options)) (*ssm.GetParameterOutput, error) {
+	if client.encrypted && !*params.WithDecryption {
+		return nil, errors.New("attempt to read encrypted parameter without decryption")
+	}
+
+	if client.name != *params.Name {
+		return nil, fmt.Errorf("unexpected parameter name; expected: %q got: %q", client.name, *params.Name)
+	}
+
+	return &ssm.GetParameterOutput{
+		Parameter: &types.Parameter{
+			Value: aws.String(client.value),
+		},
+	}, nil
+}
+
+// Create a provider using mock ssm client
+func NewTestProvider(name, value string, encrypted bool) confmap.Provider {
+	return &provider{client: &testSSMClient{name: name, value: value, encrypted: encrypted}}
+}
+
+func TestFetchParameterStorePlain(t *testing.T) {
+	parameterValue := "BAR"
+
+	for testName, nameAndURI := range uriAndNameVariants("") {
+		t.Run(testName, func(t *testing.T) {
+			fp := NewTestProvider(nameAndURI[0], parameterValue, false)
+			result, err := fp.Retrieve(context.Background(), nameAndURI[1], nil)
+
+			assert.NoError(t, err)
+			assert.NoError(t, fp.Shutdown(context.Background()))
+
+			value, err := result.AsRaw()
+			assert.NoError(t, err)
+			assert.NotNil(t, value)
+			assert.Equal(t, parameterValue, value)
+		})
+	}
+}
+
+func TestFetchParameterStoreEncrypted(t *testing.T) {
+	parameterValue := "BAR"
+
+	for testName, nameAndURI := range uriAndNameVariants("?withDecryption=true") {
+		t.Run(testName, func(t *testing.T) {
+			fp := NewTestProvider(nameAndURI[0], parameterValue, false)
+			result, err := fp.Retrieve(context.Background(), nameAndURI[1], nil)
+
+			assert.NoError(t, err)
+			assert.NoError(t, fp.Shutdown(context.Background()))
+
+			value, err := result.AsRaw()
+			assert.NoError(t, err)
+			assert.NotNil(t, value)
+			assert.Equal(t, parameterValue, value)
+		})
+	}
+}
+
+func TestFetchParameterStoreEncryptedWithoutDecryption(t *testing.T) {
+	parameterValue := "BAR"
+
+	for testName, nameAndURI := range uriAndNameVariants("") {
+		t.Run(testName, func(t *testing.T) {
+			fp := NewTestProvider(nameAndURI[0], parameterValue, true)
+			_, err := fp.Retrieve(context.Background(), nameAndURI[1], nil)
+
+			assert.Error(t, err)
+			assert.NoError(t, fp.Shutdown(context.Background()))
+		})
+	}
+}
+
+func TestFetchParameterStoreFieldValidJson(t *testing.T) {
+	parameterValue := "BAR"
+	parameterJSON := fmt.Sprintf("{\"field1\": \"%s\"}", parameterValue)
+
+	for testName, nameAndURI := range uriAndNameVariants("#field1") {
+		t.Run(testName, func(t *testing.T) {
+			fp := NewTestProvider(nameAndURI[0], parameterJSON, false)
+			result, err := fp.Retrieve(context.Background(), nameAndURI[1], nil)
+
+			assert.NoError(t, err)
+			assert.NoError(t, fp.Shutdown(context.Background()))
+
+			value, err := result.AsRaw()
+			assert.NoError(t, err)
+			assert.NotNil(t, value)
+			assert.Equal(t, parameterValue, value)
+		})
+	}
+}
+
+func TestFetchParameterStoreFieldValidEncryptedJson(t *testing.T) {
+	parameterValue := "BAR"
+	parameterJSON := fmt.Sprintf("{\"field1\": \"%s\"}", parameterValue)
+
+	for testName, nameAndURI := range uriAndNameVariants("?withDecryption=true#field1") {
+		t.Run(testName, func(t *testing.T) {
+			fp := NewTestProvider(nameAndURI[0], parameterJSON, true)
+			result, err := fp.Retrieve(context.Background(), nameAndURI[1], nil)
+
+			assert.NoError(t, err)
+			assert.NoError(t, fp.Shutdown(context.Background()))
+
+			value, err := result.AsRaw()
+			assert.NoError(t, err)
+			assert.NotNil(t, value)
+			assert.Equal(t, parameterValue, value)
+		})
+	}
+}
+
+func TestFetchParameterStoreFieldInvalidJson(t *testing.T) {
+	parameterValue := "BAR"
+
+	for testName, nameAndURI := range uriAndNameVariants("#field1") {
+		t.Run(testName, func(t *testing.T) {
+			fp := NewTestProvider(nameAndURI[0], parameterValue, true)
+			_, err := fp.Retrieve(context.Background(), nameAndURI[1], nil)
+
+			assert.Error(t, err)
+			assert.NoError(t, fp.Shutdown(context.Background()))
+		})
+	}
+}
+
+func TestFetchParameterStoreFieldMissingInJson(t *testing.T) {
+	parameterValue := "BAR"
+	parameterJSON := fmt.Sprintf("{\"field0\": \"%s\"}", parameterValue)
+
+	for testName, nameAndURI := range uriAndNameVariants("#field1") {
+		t.Run(testName, func(t *testing.T) {
+			fp := NewTestProvider(nameAndURI[0], parameterJSON, false)
+			_, err := fp.Retrieve(context.Background(), nameAndURI[1], nil)
+
+			assert.Error(t, err)
+			assert.NoError(t, fp.Shutdown(context.Background()))
+		})
+	}
+}
+
+func TestFactory(t *testing.T) {
+	p := NewFactory().Create(confmap.ProviderSettings{})
+	_, ok := p.(*provider)
+	require.True(t, ok)
+}
+
+func uriAndNameVariants(uriSuffix string) map[string][2]string {
+	name := "/test/parameter"
+	arn := "arn:aws:ssm:us-east-1:123456789012:parameter:" + name
+
+	return map[string][2]string{
+		"name": {name, "parameterstore:" + name + uriSuffix},
+		"arn":  {arn, "parameterstore:" + arn + uriSuffix},
+	}
+}