syncing up to d15c0957749f70fb6b18287b85bb6d65b9380136

Co-authored-by: Carl Noel <[email protected]>

CHANGELOG.md

@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Add `last_updated` field to the `Resource` proto, and set last updated time in update signature requests to server
 - Fixed projection for findOne action in MongoDB integration
 - Added support for creating a worksheet in GSheets integration
+- Refactor `launchdarkly` client to separate wrapper around LaunchDarkly SDK into its own package separate from the interface for agent specific flags
 
 ## v1.16.0
 

cmd/orchestrator/main.go

@@ -28,6 +28,8 @@ import (
 	"github.com/spf13/viper"
 	"github.com/superblocksteam/agent/internal/auth"
 	"github.com/superblocksteam/agent/internal/fetch"
+	"github.com/superblocksteam/agent/internal/flags"
+	flagoptions "github.com/superblocksteam/agent/internal/flags/options"
 	"github.com/superblocksteam/agent/internal/kafka"
 	"github.com/superblocksteam/agent/internal/metadata"
 	"github.com/superblocksteam/agent/internal/metrics"
@@ -48,8 +50,7 @@ import (
 	sberrors "github.com/superblocksteam/agent/pkg/errors"
 	"github.com/superblocksteam/agent/pkg/events"
 	"github.com/superblocksteam/agent/pkg/executor/options"
-	"github.com/superblocksteam/agent/pkg/flags"
-	flagoptions "github.com/superblocksteam/agent/pkg/flags/options"
+	internalflagsclient "github.com/superblocksteam/agent/pkg/flagsclient"
 	grpcserver "github.com/superblocksteam/agent/pkg/grpc"
 	httpserver "github.com/superblocksteam/agent/pkg/http"
 	"github.com/superblocksteam/agent/pkg/httpretry"
@@ -483,13 +484,18 @@ func main() {
 			flagoptions.WithDefaultGoWorkerEnabled(viper.GetBool("worker.go.enabled")),
 			flagoptions.WithDefaultWorkflowPluginInheritanceEnabled(viper.GetBool("agent.plugins.workflow.inherit_parameters.enabled")),
 		}
+		internalOptions := []internalflagsclient.Option{
+			internalflagsclient.WithLogger(logger),
+		}
 
 		if viper.GetBool("launchdarkly.local") {
 			options = append(options, flagoptions.WithLocal(viper.GetString("launchdarkly.config")))
+			internalOptions = append(internalOptions, internalflagsclient.WithLocal(viper.GetString("launchdarkly.config")))
 		}
 
 		if viper.GetBool("quotas.enabled") {
-			flagsClient = flags.LaunchDarkly(viper.GetString("launchdarkly.apikey"), options...)
+			internalFlagsClient := internalflagsclient.NewLaunchDarklyClient(viper.GetString("launchdarkly.apikey"), internalOptions...)
+			flagsClient = flags.LaunchDarkly(internalFlagsClient, options...)
 		} else {
 			flagsClient = flags.NoopFlags(options...)
 		}

internal/flags/launchdarkly.go

@@ -0,0 +1,132 @@
+package flags
+
+import (
+	"github.com/superblocksteam/agent/internal/flags/options"
+	"github.com/superblocksteam/agent/pkg/flagsclient"
+	"github.com/superblocksteam/agent/pkg/observability"
+	"github.com/superblocksteam/agent/pkg/utils"
+	apiv1 "github.com/superblocksteam/agent/types/gen/go/api/v1"
+	"go.uber.org/zap"
+)
+
+type launchdarkly struct {
+	flagsclient.FlagsClient
+
+	logger  *zap.Logger
+	options options.Options
+}
+
+func LaunchDarkly(client flagsclient.FlagsClient, ops ...options.Option) Client {
+	settings := options.Apply(ops...)
+
+	return &launchdarkly{
+		FlagsClient: client,
+		logger:      settings.Logger,
+		options:     settings,
+	}
+}
+
+func (flags *launchdarkly) GetFlagSource() int {
+	return SourceLaunchDarkly
+}
+
+func (flags *launchdarkly) GetStepSizeByOrg(orgId string) int {
+	return flags.GetIntVariationByOrg("agent.quota.org.step.size.bytes", orgId, flags.options.DefaultStepSizeByOrg)
+}
+
+func (flags *launchdarkly) GetStepSizeV2(tier string, orgId string) int {
+	return flags.GetIntVariation("agent.quota.step.size.bytes", tier, orgId, flags.options.DefaultStepSizeByOrg)
+}
+
+func (flags *launchdarkly) GetStepRateByOrg(orgId string) int {
+	return flags.GetIntVariationByOrg("agent.quota.org.step.rate.seconds", orgId, flags.options.DefaultStepRateByOrg)
+}
+
+func (flags *launchdarkly) GetStepRateV2(tier string, orgId string) int {
+	return flags.GetIntVariation("agent.quota.org.step.rate.seconds", tier, orgId, flags.options.DefaultStepRateByOrg)
+}
+
+func (flags *launchdarkly) GetStepRatePerApiByOrg(workflowId string) int {
+	return flags.GetIntVariationByOrg("agent.quota.api.step.rate.seconds", workflowId, flags.options.DefaultStepRatePerApiByOrg)
+}
+
+func (flags *launchdarkly) GetStepRatePerApiV2(tier string, orgId string) int {
+	return flags.GetIntVariation("agent.quota.api.step.rate.seconds", tier, orgId, flags.options.DefaultStepRatePerApiByOrg)
+}
+
+func (flags *launchdarkly) GetStepRatePerPluginV2(tier string, orgId string, pluginName string) int {
+	return flags.GetIntVariationCustomDims("agent.quota.plugin.step.rate.seconds", orgId, map[string]string{"tier": tier, "plugin": pluginName}, flags.options.DefaultStepRatePerPluginByOrg)
+}
+
+func (flags *launchdarkly) GetStepRatePerUserByOrg(orgId string) int {
+	return flags.GetIntVariationByOrg("agent.quota.user.step.rate.seconds", orgId, flags.options.DefaultStepRatePerUserByOrg)
+}
+
+func (flags *launchdarkly) GetStepRatePerUserV2(tier string, orgId string) int {
+	return flags.GetIntVariation("agent.quota.user.step.rate.seconds", tier, orgId, flags.options.DefaultStepRatePerUserByOrg)
+}
+
+func (flags *launchdarkly) GetStepDurationByOrg(orgId string) int {
+	return flags.GetIntVariationByOrg("agent.quota.org.step.duration.milliseconds", orgId, flags.options.DefaultStepDurationByOrg)
+}
+
+func (flags *launchdarkly) GetStepDurationV2(tier string, orgId string) int {
+	return flags.GetIntVariation("agent.quota.org.step.duration.milliseconds", tier, orgId, flags.options.DefaultStepDurationByOrg)
+}
+
+func (flags *launchdarkly) GetMaxParallelPoolSizeByAPI(apiId string) int {
+	return flags.GetIntVariationByOrg("agent.quota.api.parallel.pool.size.max", apiId, flags.options.DefaultMaxParallelPoolSizeByAPI)
+}
+
+func (flags *launchdarkly) GetMaxParallelPoolSizeV2(tier string, orgId string) int {
+	return flags.GetIntVariation("agent.quota.api.parallel.pool.size.max", tier, orgId, flags.options.DefaultMaxParallelPoolSizeByAPI)
+}
+
+func (flags *launchdarkly) GetMaxStreamSendSizeByOrg(orgId string) int {
+	return flags.GetIntVariationByOrg("agent.quota.org.stream.send.size.max", orgId, flags.options.DefaultMaxStreamSendSizeByOrg)
+}
+
+func (flags *launchdarkly) GetMaxStreamSendSizeV2(tier string, orgId string) int {
+	return flags.GetIntVariation("agent.quota.org.stream.send.size.max", tier, orgId, flags.options.DefaultMaxStreamSendSizeByOrg)
+}
+
+func (flags *launchdarkly) GetComputeMinutesPerWeekV2(tier string, orgId string) float64 {
+	return flags.GetFloatVariation("agent.quota.minutes.rate.week", tier, orgId, flags.options.DefaultMaxComputeMinutesPerWeek)
+}
+
+func (flags *launchdarkly) GetGoWorkerEnabled(tier string, orgId string) bool {
+	return flags.GetBoolVariation("worker.go.enabled", tier, orgId, flags.options.DefaultGoWorkerEnabled)
+}
+
+func (flags *launchdarkly) GetWorkflowPluginInheritanceEnabled(orgId string) bool {
+	return flags.GetBoolVariationByOrg("agent.plugins.workflow.inherit_parameters.enabled", orgId, flags.options.DefaultWorkflowPluginInheritanceEnabled)
+}
+
+func (flags *launchdarkly) GetApiTimeoutV2(api *apiv1.Api, tier string) float64 {
+	fallback := flags.options.DefaultApiTimeout
+	orgId := api.GetMetadata().GetOrganization()
+
+	logger := flags.logger.With(
+		zap.String(observability.OBS_TAG_ORG_ID, orgId),
+		zap.String(observability.OBS_TAG_API_ID, api.GetMetadata().GetId()),
+		zap.String(observability.OBS_TAG_ORG_TIER, tier),
+		zap.String(observability.OBS_TAG_RESOURCE_TYPE, utils.ApiType(api)),
+	)
+
+	var flag string
+	{
+		switch api.GetTrigger().GetConfig().(type) {
+		case *apiv1.Trigger_Application_:
+			flag = "agent.quota.api.timeout.trigger.application.milliseconds"
+		case *apiv1.Trigger_Workflow_:
+			flag = "agent.quota.api.timeout.trigger.workflow.milliseconds"
+		case *apiv1.Trigger_Job_:
+			flag = "agent.quota.api.timeout.trigger.job.milliseconds"
+		default:
+			logger.Warn("could not retrieve flag value: unknown trigger; returning default value")
+			return fallback
+		}
+	}
+
+	return flags.GetFloatVariation(flag, tier, orgId, fallback)
+}

internal/flags/launchdarkly_test.go

@@ -0,0 +1,93 @@
+package flags
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/superblocksteam/agent/internal/flags/options"
+	"github.com/superblocksteam/agent/pkg/flagsclient"
+	apiv1 "github.com/superblocksteam/agent/types/gen/go/api/v1"
+	commonv1 "github.com/superblocksteam/agent/types/gen/go/common/v1"
+)
+
+func TestGetApiTimeoutByOrgTier(t *testing.T) {
+	t.Parallel()
+
+	for _, test := range []struct {
+		name                   string
+		tier                   string
+		api                    *apiv1.Api
+		shouldCallLaunchDarkly bool
+		expected               float64
+		expectedOrg            string
+		expectedFlag           string
+	}{
+		{
+			name:     "nil api",
+			tier:     "TRIAL",
+			api:      nil,
+			expected: 500,
+		},
+		{
+			name:     "nil trigger",
+			tier:     "TRIAL",
+			api:      &apiv1.Api{},
+			expected: 500,
+		},
+		{
+			name: "application",
+			tier: "TRIAL",
+			api: &apiv1.Api{
+				Trigger:  &apiv1.Trigger{Config: &apiv1.Trigger_Application_{}},
+				Metadata: &commonv1.Metadata{Organization: "asdf"},
+			},
+			shouldCallLaunchDarkly: true,
+			expected:               1000,
+			expectedOrg:            "asdf",
+			expectedFlag:           "agent.quota.api.timeout.trigger.application.milliseconds",
+		},
+		{
+			name: "workflow",
+			tier: "ENTERPRISE",
+			api: &apiv1.Api{
+				Trigger:  &apiv1.Trigger{Config: &apiv1.Trigger_Workflow_{}},
+				Metadata: &commonv1.Metadata{Organization: "asdf"},
+			},
+			shouldCallLaunchDarkly: true,
+			expected:               4000,
+			expectedOrg:            "asdf",
+			expectedFlag:           "agent.quota.api.timeout.trigger.workflow.milliseconds",
+		},
+		{
+			name: "job",
+			tier: "FREE",
+			api: &apiv1.Api{
+				Trigger:  &apiv1.Trigger{Config: &apiv1.Trigger_Job_{}},
+				Metadata: &commonv1.Metadata{Organization: "asdf"},
+			},
+			shouldCallLaunchDarkly: true,
+			expected:               6000,
+			expectedOrg:            "asdf",
+			expectedFlag:           "agent.quota.api.timeout.trigger.job.milliseconds",
+		},
+		{
+			name:     "malformed trigger",
+			tier:     "TRIAL",
+			api:      &apiv1.Api{Trigger: &apiv1.Trigger{}},
+			expected: 500,
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			anyDefaultTimeout := float64(500)
+
+			mockFlagsClient := flagsclient.NewMockFlagsClient(t)
+			if test.shouldCallLaunchDarkly {
+				mockFlagsClient.On("GetFloatVariation", test.expectedFlag, test.tier, test.expectedOrg, anyDefaultTimeout).Return(test.expected).Once()
+			}
+
+			client := LaunchDarkly(mockFlagsClient, options.WithDefaultApiTimeout(anyDefaultTimeout))
+
+			assert.Equal(t, test.expected, client.GetApiTimeoutV2(test.api, test.tier), test.name)
+		})
+	}
+}

pkg/flags/noop.go → internal/flags/noop.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"math"
 
-	"github.com/superblocksteam/agent/pkg/flags/options"
+	"github.com/superblocksteam/agent/internal/flags/options"
 	apiv1 "github.com/superblocksteam/agent/types/gen/go/api/v1"
 	"github.com/superblocksteam/run"
 )

pkg/flags/noop_test.go → internal/flags/noop_test.go

@@ -6,7 +6,7 @@ import (
 	"time"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/superblocksteam/agent/pkg/flags/options"
+	"github.com/superblocksteam/agent/internal/flags/options"
 	"go.uber.org/zap"
 )
 

internal/schedule/schedule.go

@@ -8,12 +8,12 @@ import (
 	structpb "github.com/golang/protobuf/ptypes/struct"
 	"github.com/superblocksteam/agent/internal/auth"
 	"github.com/superblocksteam/agent/internal/fetch"
+	"github.com/superblocksteam/agent/internal/flags"
 	"github.com/superblocksteam/agent/internal/metrics"
 	"github.com/superblocksteam/agent/pkg/clients"
 	"github.com/superblocksteam/agent/pkg/crypto/signature"
 	"github.com/superblocksteam/agent/pkg/executor"
 	"github.com/superblocksteam/agent/pkg/executor/options"
-	"github.com/superblocksteam/agent/pkg/flags"
 	"github.com/superblocksteam/agent/pkg/observability"
 	"github.com/superblocksteam/agent/pkg/observability/tracer"
 	"github.com/superblocksteam/agent/pkg/secrets"