grafana-iam: Register a flag to configure dualwrite modes (#113610)

* `grafana-iam`: Register a flag to configure dualwrite modes * Streamline helper code * Launch sync job only with mode 1 to 3
2025-11-13 10:34:55 +01:00
parent 6eac95f860
commit b4b410f5be
2 changed files with 99 additions and 34 deletions
@@ -308,6 +308,7 @@ func InstallAPIs(
 			var mode = grafanarest.DualWriterMode(0)

 			var (
+				err                                  error
 				dualWriterPeriodicDataSyncJobEnabled bool
 				dualWriterMigrationDataSyncDisabled  bool
 				dataSyncerInterval                   = time.Hour
@@ -329,29 +330,45 @@ func InstallAPIs(
 				return storage, nil
 			}

-			// TODO: inherited context from main Grafana process
-			ctx := context.Background()
+			currentMode := mode
+			if !dualWriterMigrationDataSyncDisabled || dualWriterPeriodicDataSyncJobEnabled {
+				// TODO: inherited context from main Grafana process
+				ctx := context.Background()

-			// Moving from one version to the next can only happen after the previous step has
-			// successfully synchronized.
-			requestInfo := getRequestInfo(gr, namespaceMapper)
+				// Moving from one version to the next can only happen after the previous step has
+				// successfully synchronized.
+				requestInfo := getRequestInfo(gr, namespaceMapper)

-			syncerCfg := &grafanarest.SyncerConfig{
-				Kind:                   key,
-				RequestInfo:            requestInfo,
-				Mode:                   mode,
-				SkipDataSync:           dualWriterMigrationDataSyncDisabled,
-				LegacyStorage:          legacy,
-				Storage:                storage,
-				ServerLockService:      serverLock,
-				DataSyncerInterval:     dataSyncerInterval,
-				DataSyncerRecordsLimit: dataSyncerRecordsLimit,
-			}
+				syncerCfg := &grafanarest.SyncerConfig{
+					Kind:                   key,
+					RequestInfo:            requestInfo,
+					Mode:                   mode,
+					SkipDataSync:           dualWriterMigrationDataSyncDisabled,
+					LegacyStorage:          legacy,
+					Storage:                storage,
+					ServerLockService:      serverLock,
+					DataSyncerInterval:     dataSyncerInterval,
+					DataSyncerRecordsLimit: dataSyncerRecordsLimit,
+				}

-			// This also sets the currentMode on the syncer config.
-			currentMode, err := grafanarest.SetDualWritingMode(ctx, kvStore, syncerCfg, dualWriterMetrics)
-			if err != nil {
-				return nil, err
+				// This also sets the currentMode on the syncer config.
+				currentMode, err = grafanarest.SetDualWritingMode(ctx, kvStore, syncerCfg, dualWriterMetrics)
+				if err != nil {
+					return nil, err
+				}
+
+				// when unable to use
+				if currentMode != mode {
+					klog.Warningf("Requested DualWrite mode: %d, but using %d for %+v", mode, currentMode, gr)
+				}
+
+				if dualWriterPeriodicDataSyncJobEnabled && (currentMode >= grafanarest.Mode1 && currentMode <= grafanarest.Mode3) {
+					// The mode might have changed in SetDualWritingMode, so apply current mode first.
+					syncerCfg.Mode = currentMode
+					if err := grafanarest.StartPeriodicDataSyncer(ctx, syncerCfg, dualWriterMetrics); err != nil {
+						return nil, err
+					}
+				}
 			}

 			builderMetrics.RecordDualWriterModes(gr.Resource, gr.Group, mode, currentMode)
@@ -362,21 +379,8 @@ func InstallAPIs(
 			case grafanarest.Mode4, grafanarest.Mode5:
 				return storage, nil
 			default:
+				return dualwrite.NewDualWriter(gr, currentMode, legacy, storage)
 			}
-
-			if dualWriterPeriodicDataSyncJobEnabled {
-				// The mode might have changed in SetDualWritingMode, so apply current mode first.
-				syncerCfg.Mode = currentMode
-				if err := grafanarest.StartPeriodicDataSyncer(ctx, syncerCfg, dualWriterMetrics); err != nil {
-					return nil, err
-				}
-			}
-
-			// when unable to use
-			if currentMode != mode {
-				klog.Warningf("Requested DualWrite mode: %d, but using %d for %+v", mode, currentMode, gr)
-			}
-			return dualwrite.NewDualWriter(gr, currentMode, legacy, storage)
 		}
 	}

@@ -4,6 +4,8 @@ import (
 	"context"
 	"fmt"
 	"net"
+	"strconv"
+	"strings"
 	"time"

 	"github.com/spf13/pflag"
@@ -14,6 +16,7 @@ import (
 	"k8s.io/apiserver/pkg/server/options"
 	"k8s.io/client-go/rest"

+	apiserverrest "github.com/grafana/grafana/pkg/apiserver/rest"
 	"github.com/grafana/grafana/pkg/infra/tracing"
 	secret "github.com/grafana/grafana/pkg/registry/apis/secret/contracts"
 	inlinesecurevalue "github.com/grafana/grafana/pkg/registry/apis/secret/inline"
@@ -87,6 +90,58 @@ type StorageOptions struct {
 	ConfigProvider RestConfigProvider
 }

+// unifiedStorageConfigValue implements pflag.Value for parsing unified storage config
+type unifiedStorageConfigValue struct {
+	config *map[string]setting.UnifiedStorageConfig
+}
+
+func (v *unifiedStorageConfigValue) String() string {
+	if v.config == nil || len(*v.config) == 0 {
+		return ""
+	}
+	parts := make([]string, 0, len(*v.config))
+	for key, cfg := range *v.config {
+		parts = append(parts, fmt.Sprintf("%s=%d", key, cfg.DualWriterMode))
+	}
+	return strings.Join(parts, ",")
+}
+
+func (v *unifiedStorageConfigValue) Set(val string) error {
+	if val == "" {
+		return nil
+	}
+
+	// Parse comma-separated key=value pairs
+	pairs := strings.Split(val, ",")
+	for _, pair := range pairs {
+		kv := strings.SplitN(pair, "=", 2)
+		if len(kv) != 2 {
+			return fmt.Errorf("invalid format: %s (expected key=value)", pair)
+		}
+
+		key := strings.TrimSpace(kv[0])
+		mode, err := strconv.Atoi(strings.TrimSpace(kv[1]))
+		if err != nil {
+			return fmt.Errorf("invalid mode value for %s: %w", key, err)
+		}
+
+		if mode < 0 || mode > 5 {
+			return fmt.Errorf("mode must be between 0 and 5, got %d for %s", mode, key)
+		}
+
+		(*v.config)[key] = setting.UnifiedStorageConfig{
+			DualWriterMode:                      apiserverrest.DualWriterMode(mode),
+			DualWriterMigrationDataSyncDisabled: true,
+		}
+	}
+
+	return nil
+}
+
+func (v *unifiedStorageConfigValue) Type() string {
+	return "stringToUnifiedStorageConfig"
+}
+
 func NewStorageOptions() *StorageOptions {
 	return &StorageOptions{
 		StorageType:                            StorageTypeUnified,
@@ -95,6 +150,7 @@ func NewStorageOptions() *StorageOptions {
 		GrpcClientAuthenticationAllowInsecure:  false,
 		GrpcClientKeepaliveTime:                0,
 		BlobThresholdBytes:                     BlobThresholdDefault,
+		UnifiedStorageConfig:                   make(map[string]setting.UnifiedStorageConfig),
 	}
 }

@@ -109,6 +165,11 @@ func (o *StorageOptions) AddFlags(fs *pflag.FlagSet) {
 	fs.BoolVar(&o.GrpcClientAuthenticationAllowInsecure, "grpc-client-authentication-allow-insecure", o.GrpcClientAuthenticationAllowInsecure, "Allow insecure grpc client authentication")
 	fs.DurationVar(&o.GrpcClientKeepaliveTime, "grpc-client-keepalive-time", o.GrpcClientKeepaliveTime, "gRPC client keep-alive ping interval (e.g., 6m).")

+	// Use custom flag value for unified storage config
+	fs.Var(&unifiedStorageConfigValue{config: &o.UnifiedStorageConfig},
+		"grafana-apiserver-unified-storage-config",
+		"Unified storage configuration per resource.group in the format resource.group=mode,... where mode is 0-5")
+
 	// Secrets Manager Configuration flags
 	fs.BoolVar(&o.SecretsManagerGrpcClientEnable, "grafana.secrets-manager.grpc-client-enable", false, "Enable gRPC client for secrets manager")
 	fs.StringVar(&o.SecretsManagerGrpcServerAddress, "grafana.secrets-manager.grpc-server-address", "", "gRPC server address for secrets manager")