Merge branch 'main' into feat/billing

ee/query-service/app/server.go

@@ -152,7 +152,7 @@ func NewServer(config signoz.Config, signoz *signoz.SigNoz) (*Server, error) {
 	}
 
 	// start the usagemanager
-	usageManager, err := usage.New(signoz.Licensing, signoz.TelemetryStore.ClickhouseDB(), signoz.Zeus, signoz.Modules.OrgGetter)
+	usageManager, err := usage.New(signoz.Licensing, signoz.TelemetryStore.ClickhouseDB(), signoz.Zeus, signoz.Modules.OrgGetter, signoz.Flagger)
 	if err != nil {
 		return nil, err
 	}

ee/query-service/usage/manager.go

@@ -16,9 +16,11 @@ import (
 
 	"github.com/SigNoz/signoz/ee/query-service/model"
 	"github.com/SigNoz/signoz/pkg/errors"
+	"github.com/SigNoz/signoz/pkg/flagger"
 	"github.com/SigNoz/signoz/pkg/licensing"
 	"github.com/SigNoz/signoz/pkg/modules/organization"
 	"github.com/SigNoz/signoz/pkg/query-service/utils/encryption"
+	"github.com/SigNoz/signoz/pkg/types/featuretypes"
 	"github.com/SigNoz/signoz/pkg/zeus"
 )
 
@@ -43,15 +45,18 @@ type Manager struct {
 	zeus zeus.Zeus
 
 	orgGetter organization.Getter
+
+	flagger flagger.Flagger
 }
 
-func New(licenseService licensing.Licensing, clickhouseConn clickhouse.Conn, zeus zeus.Zeus, orgGetter organization.Getter) (*Manager, error) {
+func New(licenseService licensing.Licensing, clickhouseConn clickhouse.Conn, zeus zeus.Zeus, orgGetter organization.Getter, flagger flagger.Flagger) (*Manager, error) {
 	m := &Manager{
 		clickhouseConn: clickhouseConn,
 		licenseService: licenseService,
 		scheduler:      gocron.NewScheduler(time.UTC).Every(1).Day().At("00:00"), // send usage every at 00:00 UTC
 		zeus:           zeus,
 		orgGetter:      orgGetter,
+		flagger:        flagger,
 	}
 	return m, nil
 }
@@ -168,7 +173,14 @@ func (lm *Manager) UploadUsage(ctx context.Context) {
 			return
 		}
 
-		errv2 = lm.zeus.PutMeters(ctx, payload.LicenseKey.String(), body)
+		evalCtx := featuretypes.NewFlaggerEvaluationContext(organization.ID)
+		useZeus := lm.flagger.BooleanOrEmpty(ctx, flagger.FeaturePutMetersInZeus, evalCtx)
+
+		if useZeus {
+			errv2 = lm.zeus.PutMetersV2(ctx, payload.LicenseKey.String(), body)
+		} else {
+			errv2 = lm.zeus.PutMeters(ctx, payload.LicenseKey.String(), body)
+		}
 		if errv2 != nil {
 			slog.ErrorContext(ctx, "failed to upload usage", errors.Attr(errv2))
 			// not returning error here since it is captured in the failed count

ee/zeus/httpzeus/provider.go

@@ -136,6 +136,18 @@ func (provider *Provider) PutMeters(ctx context.Context, key string, data []byte
 	return err
 }
 
+func (provider *Provider) PutMetersV2(ctx context.Context, key string, data []byte) error {
+	_, err := provider.do(
+		ctx,
+		provider.config.URL.JoinPath("/v1/meters"),
+		http.MethodPost,
+		key,
+		data,
+	)
+
+	return err
+}
+
 func (provider *Provider) PutProfile(ctx context.Context, key string, profile *zeustypes.PostableProfile) error {
 	body, err := json.Marshal(profile)
 	if err != nil {

pkg/flagger/registry.go

@@ -7,6 +7,7 @@ var (
 	FeatureKafkaSpanEval     = featuretypes.MustNewName("kafka_span_eval")
 	FeatureHideRootUser      = featuretypes.MustNewName("hide_root_user")
 	FeatureGetMetersFromZeus = featuretypes.MustNewName("get_meters_from_zeus")
+	FeaturePutMetersInZeus   = featuretypes.MustNewName("put_meters_in_zeus")
 )
 
 func MustNewRegistry() featuretypes.Registry {
@@ -43,6 +44,14 @@ func MustNewRegistry() featuretypes.Registry {
 			DefaultVariant: featuretypes.MustNewName("disabled"),
 			Variants:       featuretypes.NewBooleanVariants(),
 		},
+		&featuretypes.Feature{
+			Name:           FeaturePutMetersInZeus,
+			Kind:           featuretypes.KindBoolean,
+			Stage:          featuretypes.StageExperimental,
+			Description:    "Controls whether usage meters are sent to Zeus instead of the legacy subscriptions service",
+			DefaultVariant: featuretypes.MustNewName("disabled"),
+			Variants:       featuretypes.NewBooleanVariants(),
+		},
 	)
 	if err != nil {
 		panic(err)

pkg/zeus/noopzeus/provider.go

@@ -45,6 +45,10 @@ func (provider *provider) PutMeters(_ context.Context, _ string, _ []byte) error
 	return errors.New(errors.TypeUnsupported, zeus.ErrCodeUnsupported, "putting meters is not supported")
 }
 
+func (provider *provider) PutMetersV2(_ context.Context, _ string, _ []byte) error {
+	return errors.New(errors.TypeUnsupported, zeus.ErrCodeUnsupported, "putting meters v2 is not supported")
+}
+
 func (provider *provider) PutProfile(_ context.Context, _ string, _ *zeustypes.PostableProfile) error {
 	return errors.New(errors.TypeUnsupported, zeus.ErrCodeUnsupported, "putting profile is not supported")
 }

pkg/zeus/zeus.go

@@ -29,9 +29,12 @@ type Zeus interface {
 	// Returns the billing details for the given license key.
 	GetMeters(context.Context, string) ([]byte, error)
 
-	// Puts the meters for the given license key.
+	// Puts the meters for the given license key using the legacy subscriptions service.
 	PutMeters(context.Context, string, []byte) error
 
+	// Puts the meters for the given license key using Zeus.
+	PutMetersV2(context.Context, string, []byte) error
+
 	// Put profile for the given license key.
 	PutProfile(context.Context, string, *zeustypes.PostableProfile) error