From b4b410f5be65c9629c12a62a23bf73f2e62301ad Mon Sep 17 00:00:00 2001 From: Gabriel MABILLE Date: Thu, 13 Nov 2025 10:34:55 +0100 Subject: [PATCH] `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 --- pkg/services/apiserver/builder/helper.go | 72 ++++++++++++----------- pkg/services/apiserver/options/storage.go | 61 +++++++++++++++++++ 2 files changed, 99 insertions(+), 34 deletions(-) diff --git a/pkg/services/apiserver/builder/helper.go b/pkg/services/apiserver/builder/helper.go index cfcf3540b43..98a9c08ed25 100644 --- a/pkg/services/apiserver/builder/helper.go +++ b/pkg/services/apiserver/builder/helper.go @@ -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) } } diff --git a/pkg/services/apiserver/options/storage.go b/pkg/services/apiserver/options/storage.go index ee91ff8eb89..28f6e1046ab 100644 --- a/pkg/services/apiserver/options/storage.go +++ b/pkg/services/apiserver/options/storage.go @@ -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")