user-configurable overrides: handle versioning, add PATCH endpoint

CHANGELOG.md

@@ -28,7 +28,7 @@ storage:
 * [CHANGE] **Breaking Change** Convert metrics generator from deployment to a statefulset in jsonnet. Refer to the PR for seamless migration instructions. [#2533](https://github.com/grafana/tempo/pull/2533) [#2467](https://github.com/grafana/tempo/pull/2647) (@zalegrala)
 * [FEATURE] New experimental API to derive on-demand RED metrics grouped by any attribute, and new metrics generator processor [#2368](https://github.com/grafana/tempo/pull/2368) [#2418](https://github.com/grafana/tempo/pull/2418) [#2424](https://github.com/grafana/tempo/pull/2424) [#2442](https://github.com/grafana/tempo/pull/2442) [#2480](https://github.com/grafana/tempo/pull/2480) [#2481](https://github.com/grafana/tempo/pull/2481) [#2501](https://github.com/grafana/tempo/pull/2501) [#2579](https://github.com/grafana/tempo/pull/2579) [#2582](https://github.com/grafana/tempo/pull/2582) (@mdisibio @zalegrala)
 * [FEATURE] New TraceQL structural operators descendant (>>), child (>), and sibling (~) [#2625](https://github.com/grafana/tempo/pull/2625) [#2660](https://github.com/grafana/tempo/pull/2660) (@mdisibio)
-* [FEATURE] Add user-configurable overrides module [#2543](https://github.com/grafana/tempo/pull/2543) [#2682](https://github.com/grafana/tempo/pull/2682) (@electron0zero @kvrhdn)
+* [FEATURE] Add user-configurable overrides module [#2543](https://github.com/grafana/tempo/pull/2543) [#2682](https://github.com/grafana/tempo/pull/2682) [#2681](https://github.com/grafana/tempo/pull/2681) (@electron0zero @kvrhdn)
 * [FEATURE] Add support for `q` query param in `/api/v2/search/<tag.name>/values` to filter results based on a TraceQL query [#2253](https://github.com/grafana/tempo/pull/2253) (@mapno)
 To make use of filtering, configure `autocomplete_filtering_enabled`.
 * [FEATURE] Add support for `by()` and `coalesce()` to TraceQL. [#2490](https://github.com/grafana/tempo/pull/2490) 

cmd/tempo/app/modules.go

@@ -216,6 +216,7 @@ func (t *App) initOverridesAPI() (services.Service, error) {
 
 	t.Server.HTTP.Path(overridesPath).Methods(http.MethodGet).Handler(wrapHandler(userConfigOverridesAPI.GetOverridesHandler))
 	t.Server.HTTP.Path(overridesPath).Methods(http.MethodPost).Handler(wrapHandler(userConfigOverridesAPI.PostOverridesHandler))
+	t.Server.HTTP.Path(overridesPath).Methods(http.MethodPatch).Handler(wrapHandler(userConfigOverridesAPI.PatchOverridesHandler))
 	t.Server.HTTP.Path(overridesPath).Methods(http.MethodDelete).Handler(wrapHandler(userConfigOverridesAPI.DeleteOverridesHandler))
 
 	return userConfigOverridesAPI, nil

go.mod

@@ -100,6 +100,7 @@ require (
 
 require (
 	github.com/Azure/go-autorest/autorest v0.11.28
+	github.com/evanphx/json-patch v4.12.0+incompatible
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da
 	github.com/googleapis/gax-go/v2 v2.7.0
 	github.com/grafana/gomemcache v0.0.0-20230316202710-a081dae0aba9

go.sum

@@ -587,6 +587,8 @@ github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7
 github.com/envoyproxy/protoc-gen-validate v0.6.7/go.mod h1:dyJXwwfPK2VSqiB9Klm1J6romD608Ba7Hij42vrOBCo=
 github.com/envoyproxy/protoc-gen-validate v0.9.1 h1:PS7VIOgmSVhWUEeZwTe7z7zouA22Cr590PzXKbZHOVY=
 github.com/envoyproxy/protoc-gen-validate v0.9.1/go.mod h1:OKNgG7TCp5pF4d6XftA0++PMirau2/yoOwVac3AbF2w=
+github.com/evanphx/json-patch v4.12.0+incompatible h1:4onqiflcdA9EOZ4RxV643DvftH5pOlLGNtQ5lPWQu84=
+github.com/evanphx/json-patch v4.12.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
 github.com/facette/natsort v0.0.0-20181210072756-2cd4dd1e2dcb h1:IT4JYU7k4ikYg1SCxNI1/Tieq/NFvh6dzLdgi7eu0tM=
 github.com/facette/natsort v0.0.0-20181210072756-2cd4dd1e2dcb/go.mod h1:bH6Xx7IW64qjjJq8M2u4dxNaBiDfKK+z/3eGDpXEQhc=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=

integration/e2e/config-all-in-one-gcs.yaml

@@ -45,8 +45,9 @@ overrides:
     poll_interval: 10s
     client:
       backend: gcs
+      # fsouza/fake-gcs-server does not support versioning
+      confirm_versioning: false
       gcs:
-        # TODO use separate bucket?
         bucket_name: tempo
         endpoint: https://tempo_e2e-gcs:4443/storage/v1/
         insecure: true

integration/e2e/overrides_test.go

@@ -63,12 +63,12 @@ func TestOverrides(t *testing.T) {
 
 			// Modify overrides
 			fmt.Println("* Setting overrides.forwarders")
-			err = apiClient.SetOverrides(&userconfigurableapi.UserConfigurableLimits{
+			_, err = apiClient.SetOverrides(&userconfigurableapi.UserConfigurableLimits{
 				Forwarders: &[]string{},
-			})
+			}, "0")
 			require.NoError(t, err)
 
-			limits, err := apiClient.GetOverrides()
+			limits, version, err := apiClient.GetOverrides()
 			printLimits(limits)
 			require.NoError(t, err)
 
@@ -81,10 +81,10 @@ func TestOverrides(t *testing.T) {
 
 			// Clear overrides
 			fmt.Println("* Deleting overrides")
-			err = apiClient.DeleteOverrides()
+			err = apiClient.DeleteOverrides(version)
 			require.NoError(t, err)
 
-			_, err = apiClient.GetOverrides()
+			_, _, err = apiClient.GetOverrides()
 			require.ErrorIs(t, err, httpclient.ErrNotFound)
 		})
 	}

modules/overrides/user_configurable_overrides.go

@@ -21,6 +21,7 @@ import (
 
 	api "github.com/grafana/tempo/modules/overrides/userconfigurableapi"
 	tempo_log "github.com/grafana/tempo/pkg/util/log"
+	"github.com/grafana/tempo/tempodb/backend"
 )
 
 type UserConfigurableOverridesConfig struct {
@@ -142,7 +143,11 @@ func (o *userConfigurableOverridesManager) reloadAllTenantLimits(ctx context.Con
 
 	// For every tenant with user-configurable overrides, download and cache them
 	for _, tenant := range tenants {
-		limits, err := o.client.Get(ctx, tenant)
+		limits, _, err := o.client.Get(ctx, tenant)
+		if err == backend.ErrDoesNotExist {
+			o.setTenantLimit(tenant, nil)
+			continue
+		}
 		if err != nil {
 			return errors.Wrapf(err, "failed to load tenant limits for tenant %v", tenant)
 		}

modules/overrides/user_configurable_overrides_test.go

@@ -38,7 +38,7 @@ func TestUserConfigOverridesManager(t *testing.T) {
 	userConfigurableLimits := &api.UserConfigurableLimits{
 		Forwarders: &[]string{"my-other-forwarder"},
 	}
-	err := mgr.client.Set(context.Background(), tenant2, userConfigurableLimits)
+	_, err := mgr.client.Set(context.Background(), tenant2, userConfigurableLimits, backend.VersionNew)
 	assert.NoError(t, err)
 
 	assert.NoError(t, mgr.reloadAllTenantLimits(context.Background()))
@@ -50,12 +50,12 @@ func TestUserConfigOverridesManager(t *testing.T) {
 	assert.Equal(t, []string{"my-other-forwarder"}, mgr.Forwarders(tenant2))
 
 	// Delete limits for tenant-1
-	err = mgr.client.Delete(context.Background(), tenant2)
+	err = mgr.client.Delete(context.Background(), tenant2, backend.VersionNew)
 	assert.NoError(t, err)
 
 	assert.NoError(t, mgr.reloadAllTenantLimits(context.Background()))
 
-	// Verify default limits are returned
+	// Verify default limits are returned again
 	assert.Equal(t, 1024, mgr.MaxBytesPerTrace(tenant1))
 	assert.Equal(t, []string{"my-forwarder"}, mgr.Forwarders(tenant1))
 	assert.Equal(t, 1024, mgr.MaxBytesPerTrace(tenant2))
@@ -92,7 +92,7 @@ func TestUserConfigOverridesManager_deletedFromBackend(t *testing.T) {
 	limits := &api.UserConfigurableLimits{
 		Forwarders: &[]string{"my-other-forwarder"},
 	}
-	err := mgr.client.Set(context.Background(), tenant1, limits)
+	_, err := mgr.client.Set(context.Background(), tenant1, limits, backend.VersionNew)
 	assert.NoError(t, err)
 
 	assert.NoError(t, mgr.reloadAllTenantLimits(context.Background()))
@@ -118,13 +118,13 @@ func TestUserConfigOverridesManager_backendUnavailable(t *testing.T) {
 	limits := &api.UserConfigurableLimits{
 		Forwarders: &[]string{"my-other-forwarder"},
 	}
-	err := mgr.client.Set(context.Background(), tenant1, limits)
+	_, err := mgr.client.Set(context.Background(), tenant1, limits, backend.VersionNew)
 	assert.NoError(t, err)
 
 	assert.NoError(t, mgr.reloadAllTenantLimits(context.Background()))
 
 	// replace reader by this uncooperative fella
-	mgr.client = badClient{}
+	mgr.client = &badClient{}
 
 	// reloading fails
 	assert.Error(t, mgr.reloadAllTenantLimits(context.Background()))
@@ -197,7 +197,7 @@ func writeUserConfigurableOverridesToDisk(t *testing.T, dir string, tenant strin
 	})
 	assert.NoError(t, err)
 
-	err = client.Set(context.Background(), tenant, limits)
+	_, err = client.Set(context.Background(), tenant, limits, backend.VersionNew)
 	assert.NoError(t, err)
 }
 
@@ -208,25 +208,27 @@ func deleteUserConfigurableOverridesFromDisk(t *testing.T, dir string, tenant st
 	})
 	assert.NoError(t, err)
 
-	err = client.Delete(context.Background(), tenant)
+	err = client.Delete(context.Background(), tenant, backend.VersionNew)
 	assert.NoError(t, err)
 }
 
 type badClient struct{}
 
-func (b badClient) List(context.Context) ([]string, error) {
+var _ api.Client = (*badClient)(nil)
+
+func (b *badClient) List(context.Context) ([]string, error) {
 	return nil, errors.New("no")
 }
 
-func (b badClient) Get(context.Context, string) (*api.UserConfigurableLimits, error) {
-	return nil, errors.New("no")
+func (b *badClient) Get(context.Context, string) (*api.UserConfigurableLimits, backend.Version, error) {
+	return nil, "", errors.New("no")
 }
 
-func (b badClient) Set(context.Context, string, *api.UserConfigurableLimits) error {
-	return errors.New("no")
+func (b *badClient) Set(context.Context, string, *api.UserConfigurableLimits, backend.Version) (backend.Version, error) {
+	return "", errors.New("no")
 }
 
-func (b badClient) Delete(context.Context, string) error {
+func (b *badClient) Delete(context.Context, string, backend.Version) error {
 	return errors.New("no")
 }
 

modules/overrides/userconfigurableapi/client.go

@@ -9,11 +9,13 @@ import (
 	"os"
 	"path"
 
+	"github.com/go-kit/log/level"
 	jsoniter "github.com/json-iterator/go"
 	"github.com/opentracing/opentracing-go"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 
+	"github.com/grafana/tempo/pkg/util/log"
 	"github.com/grafana/tempo/tempodb/backend"
 	"github.com/grafana/tempo/tempodb/backend/azure"
 	"github.com/grafana/tempo/tempodb/backend/gcs"
@@ -27,6 +29,11 @@ const (
 )
 
 var (
+	metricList = promauto.NewCounter(prometheus.CounterOpts{
+		Namespace: "tempo",
+		Name:      "overrides_user_configurable_overrides_list_total",
+		Help:      "How often the user-configurable overrides was listed",
+	})
 	metricFetch = promauto.NewCounterVec(prometheus.CounterOpts{
 		Namespace: "tempo",
 		Name:      "overrides_user_configurable_overrides_fetch_total",
@@ -42,79 +49,104 @@ var (
 type UserConfigurableOverridesClientConfig struct {
 	Backend string `yaml:"backend"`
 
+	// ConfirmVersioning is enabled when creating the backend client. If versioning is disabled no
+	// checks against concurrent writes will be performed.
+	ConfirmVersioning bool `yaml:"confirm_versioning"`
+
 	Local *local.Config `yaml:"local"`
 	GCS   *gcs.Config   `yaml:"gcs"`
 	S3    *s3.Config    `yaml:"s3"`
 	Azure *azure.Config `yaml:"azure"`
 }
 
 func (c *UserConfigurableOverridesClientConfig) RegisterFlagsAndApplyDefaults(*flag.FlagSet) {
+	c.ConfirmVersioning = true
+
 	c.Local = &local.Config{}
 	c.GCS = &gcs.Config{}
 	c.S3 = &s3.Config{}
 	c.Azure = &azure.Config{}
 }
 
+// Client is a collection of methods to manage overrides on a backend.
 type Client interface {
+	// List tenants that have user-configurable overrides.
 	List(ctx context.Context) ([]string, error)
-	Get(context.Context, string) (*UserConfigurableLimits, error)
-	Set(context.Context, string, *UserConfigurableLimits) error
-	Delete(context.Context, string) error
+	// Get the user-configurable overrides. Returns backend.ErrDoesNotExist if no limits are set.
+	Get(context.Context, string) (*UserConfigurableLimits, backend.Version, error)
+	// Set the user-configurable overrides. Returns backend.ErrVersionDoesNotMatch if the backend
+	// has a newer version.
+	Set(context.Context, string, *UserConfigurableLimits, backend.Version) (backend.Version, error)
+	// Delete the user-configurable overrides.
+	Delete(context.Context, string, backend.Version) error
+	// Shutdown the client.
 	Shutdown()
 }
 
 type userConfigOverridesClient struct {
-	r backend.RawReader
-	w backend.RawWriter
+	rw backend.VersionedReaderWriter
 }
 
 var _ Client = (*userConfigOverridesClient)(nil)
 
 func NewUserConfigOverridesClient(cfg *UserConfigurableOverridesClientConfig) (Client, error) {
-	r, w, err := initBackend(cfg)
+	rw, err := initBackend(cfg)
 	if err != nil {
 		return nil, err
 	}
-	return &userConfigOverridesClient{r, w}, nil
+	return &userConfigOverridesClient{rw}, nil
 }
 
 func (o *userConfigOverridesClient) Shutdown() {
-	o.r.Shutdown()
+	o.rw.Shutdown()
 }
 
-func initBackend(cfg *UserConfigurableOverridesClientConfig) (reader backend.RawReader, writer backend.RawWriter, err error) {
+func initBackend(cfg *UserConfigurableOverridesClientConfig) (rw backend.VersionedReaderWriter, err error) {
 	switch cfg.Backend {
 	case backend.Local:
-		reader, writer, _, err = local.New(cfg.Local)
+		r, w, _, err := local.New(cfg.Local)
 		if err != nil {
-			return
+			return nil, err
 		}
 		// Create overrides directory with necessary permissions
 		err = os.MkdirAll(path.Join(cfg.Local.Path, OverridesKeyPath), os.ModePerm)
+		if err != nil {
+			return nil, err
+		}
+		rw = backend.NewFakeVersionedReaderWriter(r, w)
 	case backend.GCS:
-		reader, writer, _, err = gcs.New(cfg.GCS)
+		rw, err = gcs.NewVersionedReaderWriter(cfg.GCS, cfg.ConfirmVersioning)
 	case backend.S3:
-		reader, writer, _, err = s3.New(cfg.S3)
+		rw, err = s3.NewVersionedReaderWriter(cfg.S3)
 	case backend.Azure:
-		reader, writer, _, err = azure.New(cfg.Azure)
+		rw, err = azure.NewVersionedReaderWriter(cfg.Azure)
 	default:
 		err = fmt.Errorf("unknown backend %s", cfg.Backend)
 	}
-	return
+	if err != nil {
+		return nil, err
+	}
+	if cfg.Backend == backend.Local || cfg.Backend == backend.S3 || cfg.Backend == backend.Azure {
+		level.Warn(log.Logger).Log(
+			"msg", "versioned backend requests are best-effort for the configured backend, concurrent requests modifying user-configurable overrides might cause data races",
+			"backend", cfg.Backend,
+		)
+	}
+	return rw, nil
 }
 
 func (o *userConfigOverridesClient) List(ctx context.Context) ([]string, error) {
 	span, ctx := opentracing.StartSpanFromContext(ctx, "userConfigOverridesClient.List")
 	defer span.Finish()
 
-	return o.r.List(ctx, []string{OverridesKeyPath})
+	metricList.Inc()
+
+	return o.rw.List(ctx, []string{OverridesKeyPath})
 }
 
-// Get downloads the tenant limits from the backend. Returns nil, nil if no limits are set.
-func (o *userConfigOverridesClient) Get(ctx context.Context, userID string) (tenantLimits *UserConfigurableLimits, err error) {
-	span, ctx := opentracing.StartSpanFromContext(ctx, "userConfigOverridesClient.Get")
+func (o *userConfigOverridesClient) Get(ctx context.Context, userID string) (tenantLimits *UserConfigurableLimits, version backend.Version, err error) {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "userConfigOverridesClient.Get", opentracing.Tag{Key: "tenant", Value: userID})
 	defer span.Finish()
-	span.SetTag("tenant", userID)
 
 	metricFetch.WithLabelValues(userID).Inc()
 	defer func() {
@@ -123,12 +155,9 @@ func (o *userConfigOverridesClient) Get(ctx context.Context, userID string) (ten
 		}
 	}()
 
-	reader, _, err := o.r.Read(ctx, OverridesFileName, []string{OverridesKeyPath, userID}, false)
-	if err == backend.ErrDoesNotExist {
-		return nil, nil
-	}
+	reader, version, err := o.rw.ReadVersioned(ctx, OverridesFileName, []string{OverridesKeyPath, userID})
 	if err != nil {
-		return
+		return nil, "", err
 	}
 	defer reader.Close()
 
@@ -137,26 +166,21 @@ func (o *userConfigOverridesClient) Get(ctx context.Context, userID string) (ten
 	return
 }
 
-// Set stores the user-configurable limits on the backend.
-func (o *userConfigOverridesClient) Set(ctx context.Context, userID string, limits *UserConfigurableLimits) error {
-	span, ctx := opentracing.StartSpanFromContext(ctx, "userConfigOverridesClient.Set")
+func (o *userConfigOverridesClient) Set(ctx context.Context, userID string, limits *UserConfigurableLimits, version backend.Version) (backend.Version, error) {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "userConfigOverridesClient.Set", opentracing.Tag{Key: "tenant", Value: userID})
 	defer span.Finish()
-	span.SetTag("tenant", userID)
 
 	data, err := jsoniter.Marshal(limits)
 	if err != nil {
-		return err
+		return "", err
 	}
 
-	// Store on the bucket
-	return o.w.Write(ctx, OverridesFileName, []string{OverridesKeyPath, userID}, bytes.NewReader(data), -1, false)
+	return o.rw.WriteVersioned(ctx, OverridesFileName, []string{OverridesKeyPath, userID}, bytes.NewReader(data), version)
 }
 
-// Delete will clear all user-configurable limits for the given tenant.
-func (o *userConfigOverridesClient) Delete(ctx context.Context, userID string) error {
-	span, ctx := opentracing.StartSpanFromContext(ctx, "userConfigOverridesClient.Delete")
+func (o *userConfigOverridesClient) Delete(ctx context.Context, userID string, version backend.Version) error {
+	span, ctx := opentracing.StartSpanFromContext(ctx, "userConfigOverridesClient.Delete", opentracing.Tag{Key: "tenant", Value: userID})
 	defer span.Finish()
-	span.SetTag("tenant", userID)
 
-	return o.w.Delete(ctx, OverridesFileName, []string{OverridesKeyPath, userID}, false)
+	return o.rw.DeleteVersioned(ctx, OverridesFileName, []string{OverridesKeyPath, userID}, version)
 }