Support SDK key on HTTP endpoints and RPCs / OTEL

.github/workflows/proxy-ci.yml

@@ -19,15 +19,6 @@ jobs:
 
   test:
     runs-on: ubuntu-latest
-    services:
-      mongodb:
-        image: mongodb/mongodb-community-server
-        ports:
-          - 27017:27017
-      dynamodb:
-        image: amazon/dynamodb-local
-        ports:
-          - 8000:8000
     steps:
     - uses: actions/checkout@v5
     - name: Setup Go
@@ -47,15 +38,6 @@ jobs:
   coverage:
     runs-on: ubuntu-latest
     needs: test
-    services:
-      mongodb:
-        image: mongodb/mongodb-community-server
-        ports:
-          - 27017:27017
-      dynamodb:
-        image: amazon/dynamodb-local
-        ports:
-          - 8000:8000
     steps:
       - uses: actions/checkout@v5
         with:

.github/workflows/proxy-release.yml

@@ -12,15 +12,6 @@ permissions:
 jobs:
   test:
     runs-on: ubuntu-latest
-    services:
-      mongodb:
-        image: mongodb/mongodb-community-server
-        ports:
-          - 27017:27017
-      dynamodb:
-        image: amazon/dynamodb-local
-        ports:
-          - 8000:8000
     steps:
     - uses: actions/checkout@v5
     - name: Setup Go

cache/cache.go

@@ -0,0 +1,51 @@
+package cache
+
+import (
+	"context"
+	"time"
+
+	"github.com/configcat/configcat-proxy/config"
+	"github.com/configcat/configcat-proxy/diag/telemetry"
+	"github.com/configcat/configcat-proxy/log"
+	configcat "github.com/configcat/go-sdk/v9"
+)
+
+const (
+	keyName     = "key"
+	payloadName = "payload"
+)
+
+type ReaderWriter = configcat.ConfigCache
+
+type External interface {
+	ReaderWriter
+	Shutdown()
+}
+
+func SetupExternalCache(conf *config.CacheConfig, telemetryReporter telemetry.Reporter, log log.Logger) (External, error) {
+	cacheLog := log.WithPrefix("cache")
+
+	ctx, cancel := context.WithTimeout(context.Background(), 15*time.Second) // give 15 sec to spin up the cache connection
+	defer cancel()
+
+	if conf.Redis.Enabled {
+		redis, err := newRedis(&conf.Redis, telemetryReporter, cacheLog)
+		if err != nil {
+			return nil, err
+		}
+		return redis, nil
+	} else if conf.MongoDb.Enabled {
+		mongoDb, err := newMongoDb(ctx, &conf.MongoDb, telemetryReporter, cacheLog)
+		if err != nil {
+			return nil, err
+		}
+		return mongoDb, nil
+	} else if conf.DynamoDb.Enabled {
+		dynamoDb, err := newDynamoDb(ctx, &conf.DynamoDb, telemetryReporter, cacheLog)
+		if err != nil {
+			return nil, err
+		}
+		return dynamoDb, nil
+	}
+	return nil, nil
+}

cache/cache_test.go

@@ -0,0 +1,64 @@
+package cache
+
+import (
+	"testing"
+
+	"github.com/alicebob/miniredis/v2"
+	"github.com/configcat/configcat-proxy/config"
+	"github.com/configcat/configcat-proxy/diag/telemetry"
+	"github.com/configcat/configcat-proxy/log"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSetupExternalCache_OnlyOneSelected(t *testing.T) {
+	s := miniredis.RunT(t)
+	store, err := SetupExternalCache(&config.CacheConfig{
+		Redis: config.RedisConfig{Addresses: []string{s.Addr()}, Enabled: true},
+		MongoDb: config.MongoDbConfig{
+			Enabled:    true,
+			Url:        "mongodb://localhost:27017",
+			Database:   "test_db",
+			Collection: "coll",
+		},
+	}, telemetry.NewEmptyReporter(), log.NewNullLogger())
+	assert.NoError(t, err)
+	defer store.Shutdown()
+	assert.IsType(t, &redisStore{}, store)
+}
+
+func (s *mongoTestSuite) TestSetupExternalCache() {
+	store, err := SetupExternalCache(&config.CacheConfig{MongoDb: config.MongoDbConfig{
+		Enabled:    true,
+		Url:        s.addr,
+		Database:   "test_db",
+		Collection: "coll",
+	}}, telemetry.NewEmptyReporter(), log.NewNullLogger())
+	assert.NoError(s.T(), err)
+	defer store.Shutdown()
+	assert.IsType(s.T(), &mongoDbStore{}, store)
+}
+
+func (s *redisTestSuite) TestSetupExternalCache() {
+	store, err := SetupExternalCache(&config.CacheConfig{Redis: config.RedisConfig{Addresses: []string{"localhost:" + s.dbPort}, Enabled: true}}, telemetry.NewEmptyReporter(), log.NewNullLogger())
+	assert.NoError(s.T(), err)
+	defer store.Shutdown()
+	assert.IsType(s.T(), &redisStore{}, store)
+}
+
+func (s *valkeyTestSuite) TestSetupExternalCache() {
+	store, err := SetupExternalCache(&config.CacheConfig{Redis: config.RedisConfig{Addresses: []string{"localhost:" + s.dbPort}, Enabled: true}}, telemetry.NewEmptyReporter(), log.NewNullLogger())
+	assert.NoError(s.T(), err)
+	defer store.Shutdown()
+	assert.IsType(s.T(), &redisStore{}, store)
+}
+
+func (s *dynamoDbTestSuite) TestSetupExternalCache() {
+	store, err := SetupExternalCache(&config.CacheConfig{DynamoDb: config.DynamoDbConfig{
+		Enabled: true,
+		Table:   tableName,
+		Url:     s.addr,
+	}}, telemetry.NewEmptyReporter(), log.NewNullLogger())
+	assert.NoError(s.T(), err)
+	defer store.Shutdown()
+	assert.IsType(s.T(), &dynamoDbStore{}, store)
+}

sdk/store/cache/dynamodb.go → cache/dynamodb.go

@@ -9,6 +9,7 @@ import (
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
 	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
 	"github.com/configcat/configcat-proxy/config"
+	"github.com/configcat/configcat-proxy/diag/telemetry"
 	"github.com/configcat/configcat-proxy/log"
 )
 
@@ -18,9 +19,10 @@ type dynamoDbStore struct {
 	log      log.Logger
 }
 
-func newDynamoDb(ctx context.Context, conf *config.DynamoDbConfig, log log.Logger) (External, error) {
+func newDynamoDb(ctx context.Context, conf *config.DynamoDbConfig, telemetryReporter telemetry.Reporter, log log.Logger) (External, error) {
 	dynamoLog := log.WithPrefix("dynamodb")
 	awsCtx, err := awsconfig.LoadDefaultConfig(ctx)
+	telemetryReporter.InstrumentAws(&awsCtx)
 	if err != nil {
 		dynamoLog.Errorf("couldn't read aws config for DynamoDB: %s", err)
 		return nil, err

cache/dynamodb_test.go

@@ -0,0 +1,174 @@
+package cache
+
+import (
+	"context"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	awsconfig "github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
+	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
+	"github.com/configcat/configcat-proxy/config"
+	"github.com/configcat/configcat-proxy/diag/telemetry"
+	"github.com/configcat/configcat-proxy/log"
+	"github.com/configcat/go-sdk/v9/configcatcache"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/suite"
+	"github.com/testcontainers/testcontainers-go"
+	tcdynamodb "github.com/testcontainers/testcontainers-go/modules/dynamodb"
+)
+
+const (
+	tableName = "test-table"
+)
+
+type dynamoDbTestSuite struct {
+	suite.Suite
+
+	db   *tcdynamodb.DynamoDBContainer
+	addr string
+}
+
+func (s *dynamoDbTestSuite) SetupSuite() {
+	dynamoDbContainer, err := tcdynamodb.Run(s.T().Context(), "amazon/dynamodb-local")
+	if err != nil {
+		panic("failed to start container: " + err.Error() + "")
+	}
+	s.db = dynamoDbContainer
+	str, _ := s.db.ConnectionString(s.T().Context())
+	s.addr = "http://" + str
+
+	s.T().Setenv("AWS_ACCESS_KEY_ID", "key")
+	s.T().Setenv("AWS_SECRET_ACCESS_KEY", "secret")
+	s.T().Setenv("AWS_SESSION_TOKEN", "session")
+	s.T().Setenv("AWS_DEFAULT_REGION", "us-east-1")
+}
+
+func (s *dynamoDbTestSuite) TearDownSuite() {
+	if err := testcontainers.TerminateContainer(s.db); err != nil {
+		panic("failed to terminate container: " + err.Error() + "")
+	}
+}
+
+func TestRunDynamoDbSuite(t *testing.T) {
+	suite.Run(t, new(dynamoDbTestSuite))
+}
+
+func (s *dynamoDbTestSuite) TestDynamoDbStore() {
+	assert.NoError(s.T(), createTableIfNotExist(s.T().Context(), tableName, s.addr))
+
+	s.Run("ok", func() {
+		store, err := newDynamoDb(s.T().Context(), &config.DynamoDbConfig{
+			Enabled: true,
+			Table:   tableName,
+			Url:     s.addr,
+		}, telemetry.NewEmptyReporter(), log.NewNullLogger())
+		assert.NoError(s.T(), err)
+		defer store.Shutdown()
+
+		cacheEntry := configcatcache.CacheSegmentsToBytes(time.Now(), "etag", []byte(`test`))
+
+		err = store.Set(s.T().Context(), "k1", cacheEntry)
+		assert.NoError(s.T(), err)
+
+		res, err := store.Get(s.T().Context(), "k1")
+		assert.NoError(s.T(), err)
+		assert.Equal(s.T(), cacheEntry, res)
+
+		cacheEntry = configcatcache.CacheSegmentsToBytes(time.Now(), "etag", []byte(`test2`))
+
+		err = store.Set(s.T().Context(), "k1", cacheEntry)
+		assert.NoError(s.T(), err)
+
+		res, err = store.Get(s.T().Context(), "k1")
+		assert.NoError(s.T(), err)
+		assert.Equal(s.T(), cacheEntry, res)
+	})
+
+	s.Run("empty", func() {
+		store, err := newDynamoDb(s.T().Context(), &config.DynamoDbConfig{
+			Enabled: true,
+			Table:   tableName,
+			Url:     s.addr,
+		}, telemetry.NewEmptyReporter(), log.NewNullLogger())
+		assert.NoError(s.T(), err)
+		defer store.Shutdown()
+
+		_, err = store.Get(s.T().Context(), "k2")
+		assert.Error(s.T(), err)
+	})
+
+	s.Run("no-table", func() {
+		store, err := newDynamoDb(s.T().Context(), &config.DynamoDbConfig{
+			Enabled: true,
+			Table:   "nonexisting",
+			Url:     s.addr,
+		}, telemetry.NewEmptyReporter(), log.NewNullLogger())
+		assert.NoError(s.T(), err)
+		defer store.Shutdown()
+
+		_, err = store.Get(s.T().Context(), "k3")
+		assert.Error(s.T(), err)
+	})
+}
+
+func createTableIfNotExist(ctx context.Context, table string, addr string) error {
+	awsCtx, err := awsconfig.LoadDefaultConfig(ctx)
+	if err != nil {
+		return err
+	}
+	var opts []func(*dynamodb.Options)
+	opts = append(opts, func(options *dynamodb.Options) {
+		options.BaseEndpoint = aws.String(addr)
+	})
+
+	client := dynamodb.NewFromConfig(awsCtx, opts...)
+
+	_, err = client.DescribeTable(ctx, &dynamodb.DescribeTableInput{
+		TableName: aws.String(table),
+	})
+	if err == nil {
+		return nil
+	}
+	_, err = client.CreateTable(ctx, &dynamodb.CreateTableInput{
+		TableName: aws.String(table),
+		AttributeDefinitions: []types.AttributeDefinition{
+			{
+				AttributeName: aws.String(keyName),
+				AttributeType: types.ScalarAttributeTypeS,
+			},
+		},
+		KeySchema: []types.KeySchemaElement{
+			{
+				AttributeName: aws.String(keyName),
+				KeyType:       types.KeyTypeHash,
+			},
+		},
+		ProvisionedThroughput: &types.ProvisionedThroughput{
+			ReadCapacityUnits:  aws.Int64(1),
+			WriteCapacityUnits: aws.Int64(1),
+		},
+	})
+	if err != nil {
+		return err
+	}
+
+	timeout := time.After(5 * time.Second)
+	ticker := time.NewTicker(100 * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-timeout:
+			return fmt.Errorf("table creation timed out")
+		case <-ticker.C:
+			res, err := client.DescribeTable(ctx, &dynamodb.DescribeTableInput{
+				TableName: aws.String(table),
+			})
+			if err == nil && res.Table.TableStatus == types.TableStatusActive {
+				return nil
+			}
+		}
+	}
+}

sdk/store/cache/mongodb.go → cache/mongodb.go

@@ -5,10 +5,11 @@ import (
 	"time"
 
 	"github.com/configcat/configcat-proxy/config"
+	"github.com/configcat/configcat-proxy/diag/telemetry"
 	"github.com/configcat/configcat-proxy/log"
-	"go.mongodb.org/mongo-driver/bson"
-	"go.mongodb.org/mongo-driver/mongo"
-	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
 )
 
 type mongoDbStore struct {
@@ -22,8 +23,9 @@ type entry struct {
 	Payload []byte
 }
 
-func newMongoDb(ctx context.Context, conf *config.MongoDbConfig, log log.Logger) (External, error) {
+func newMongoDb(ctx context.Context, conf *config.MongoDbConfig, telemetryReporter telemetry.Reporter, log log.Logger) (External, error) {
 	opts := options.Client().ApplyURI(conf.Url)
+	telemetryReporter.InstrumentMongoDb(opts)
 	if conf.Tls.Enabled {
 		t, err := conf.Tls.LoadTlsOptions()
 		if err != nil {
@@ -32,7 +34,7 @@ func newMongoDb(ctx context.Context, conf *config.MongoDbConfig, log log.Logger)
 		}
 		opts.SetTLSConfig(t)
 	}
-	client, err := mongo.Connect(ctx, opts)
+	client, err := mongo.Connect(opts)
 	if err != nil {
 		log.Errorf("couldn't connect to MongoDB: %s", err)
 		return nil, err