DualWrite: Cleanup and centralize the dual write creation (#90013)
This commit is contained in:
@@ -22,8 +22,6 @@ import (
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
servicev0alpha1 "github.com/grafana/grafana/pkg/apis/service/v0alpha1"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/service"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"gopkg.in/yaml.v3"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
@@ -47,7 +45,9 @@ import (
|
||||
"k8s.io/kube-aggregator/pkg/controllers"
|
||||
"k8s.io/kube-aggregator/pkg/controllers/autoregister"
|
||||
|
||||
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
servicev0alpha1 "github.com/grafana/grafana/pkg/apis/service/v0alpha1"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/service"
|
||||
|
||||
servicev0alpha1applyconfiguration "github.com/grafana/grafana/pkg/generated/applyconfiguration/service/v0alpha1"
|
||||
serviceclientset "github.com/grafana/grafana/pkg/generated/clientset/versioned"
|
||||
informersv0alpha1 "github.com/grafana/grafana/pkg/generated/informers/externalversions"
|
||||
@@ -290,8 +290,7 @@ func CreateAggregatorServer(config *Config, delegateAPIServer genericapiserver.D
|
||||
aggregatorscheme.Scheme,
|
||||
aggregatorscheme.Codecs,
|
||||
aggregatorConfig.GenericConfig.RESTOptionsGetter,
|
||||
grafanarest.Mode0,
|
||||
reg,
|
||||
nil, // no dual writer
|
||||
)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
|
||||
@@ -3,8 +3,6 @@ package builder
|
||||
import (
|
||||
"net/http"
|
||||
|
||||
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/runtime/serializer"
|
||||
@@ -13,6 +11,8 @@ import (
|
||||
genericapiserver "k8s.io/apiserver/pkg/server"
|
||||
"k8s.io/kube-openapi/pkg/common"
|
||||
"k8s.io/kube-openapi/pkg/spec3"
|
||||
|
||||
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
)
|
||||
|
||||
// TODO: this (or something like it) belongs in grafana-app-sdk,
|
||||
@@ -29,8 +29,7 @@ type APIGroupBuilder interface {
|
||||
scheme *runtime.Scheme,
|
||||
codecs serializer.CodecFactory,
|
||||
optsGetter generic.RESTOptionsGetter,
|
||||
desiredMode grafanarest.DualWriterMode,
|
||||
reg prometheus.Registerer,
|
||||
dualWrite grafanarest.DualWriteBuilder,
|
||||
) (*genericapiserver.APIGroupInfo, error)
|
||||
|
||||
// Get OpenAPI definitions
|
||||
@@ -43,11 +42,6 @@ type APIGroupBuilder interface {
|
||||
// Standard namespace checking will happen before this is called, specifically
|
||||
// the namespace must matches an org|stack that the user belongs to
|
||||
GetAuthorizer() authorizer.Authorizer
|
||||
|
||||
// Get the desired dual writing mode. These are modes 1, 2, 3 and 4 if
|
||||
// the feature flag `unifiedStorage` is enabled and mode 0 if it is not enabled.
|
||||
// #TODO add type for map[string]grafanarest.DualWriterMode?
|
||||
GetDesiredDualWriterMode(dualWrite bool, toMode map[string]grafanarest.DualWriterMode) grafanarest.DualWriterMode
|
||||
}
|
||||
|
||||
// Builders that implement OpenAPIPostProcessor are given a chance to modify the schema directly
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package builder
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"regexp"
|
||||
@@ -10,11 +11,11 @@ import (
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/grafana/grafana/pkg/web"
|
||||
"github.com/prometheus/client_golang/prometheus"
|
||||
"golang.org/x/mod/semver"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/runtime/schema"
|
||||
"k8s.io/apimachinery/pkg/runtime/serializer"
|
||||
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
|
||||
"k8s.io/apimachinery/pkg/version"
|
||||
@@ -26,7 +27,10 @@ import (
|
||||
k8stracing "k8s.io/component-base/tracing"
|
||||
"k8s.io/kube-openapi/pkg/common"
|
||||
|
||||
"github.com/grafana/grafana/pkg/web"
|
||||
|
||||
"github.com/grafana/grafana/pkg/apiserver/endpoints/filters"
|
||||
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
|
||||
"github.com/grafana/grafana/pkg/services/apiserver/options"
|
||||
)
|
||||
|
||||
@@ -136,6 +140,10 @@ func SetupConfig(
|
||||
return nil
|
||||
}
|
||||
|
||||
type ServerLockService interface {
|
||||
LockExecuteAndRelease(ctx context.Context, actionName string, maxInterval time.Duration, fn func(ctx context.Context)) error
|
||||
}
|
||||
|
||||
func InstallAPIs(
|
||||
scheme *runtime.Scheme,
|
||||
codecs serializer.CodecFactory,
|
||||
@@ -144,15 +152,40 @@ func InstallAPIs(
|
||||
builders []APIGroupBuilder,
|
||||
storageOpts *options.StorageOptions,
|
||||
reg prometheus.Registerer,
|
||||
kvStore grafanarest.NamespacedKVStore,
|
||||
serverLock ServerLockService,
|
||||
) error {
|
||||
// dual writing is only enabled when the storage type is not legacy.
|
||||
// this is needed to support setting a default RESTOptionsGetter for new APIs that don't
|
||||
// support the legacy storage type.
|
||||
dualWriteEnabled := storageOpts.StorageType != options.StorageTypeLegacy
|
||||
var dualWrite grafanarest.DualWriteBuilder
|
||||
if storageOpts.StorageType != options.StorageTypeLegacy {
|
||||
dualWrite = func(gr schema.GroupResource, legacy grafanarest.LegacyStorage, storage grafanarest.Storage) (grafanarest.Storage, error) {
|
||||
key := gr.String() // ${resource}.{group} eg playlists.playlist.grafana.app
|
||||
|
||||
// Get the option from custom.ini/command line
|
||||
// when missing this will default to mode zero (legacy only)
|
||||
mode := storageOpts.DualWriterDesiredModes[key]
|
||||
|
||||
// Moving from one version to the next can only happen after the previous step has
|
||||
// successfully synchronized.
|
||||
currentMode, err := grafanarest.SetDualWritingMode(context.Background(), kvStore, legacy, storage, key, mode, reg)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
switch currentMode {
|
||||
case grafanarest.Mode0:
|
||||
return legacy, nil
|
||||
case grafanarest.Mode4:
|
||||
return storage, nil
|
||||
default:
|
||||
}
|
||||
return grafanarest.NewDualWriter(currentMode, legacy, storage, reg), nil
|
||||
}
|
||||
}
|
||||
|
||||
for _, b := range builders {
|
||||
mode := b.GetDesiredDualWriterMode(dualWriteEnabled, storageOpts.DualWriterDesiredModes)
|
||||
g, err := b.GetAPIGroupInfo(scheme, codecs, optsGetter, mode, reg)
|
||||
g, err := b.GetAPIGroupInfo(scheme, codecs, optsGetter, dualWrite)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -59,6 +59,11 @@ func applyGrafanaConfig(cfg *setting.Cfg, features featuremgmt.FeatureToggles, o
|
||||
o.StorageOptions.DualWriterDesiredModes = map[string]grafanarest.DualWriterMode{
|
||||
playlist.GROUPRESOURCE: grafanarest.DualWriterMode(unifiedStorageModeCfg.Key(playlist.GROUPRESOURCE).MustInt(0)),
|
||||
}
|
||||
|
||||
// TODO: ensure backwards compatibility with production
|
||||
// remove this after changing the unified_storage_mode key format in HGAPI
|
||||
o.StorageOptions.DualWriterDesiredModes[playlist.RESOURCE+"."+playlist.GROUP] = o.StorageOptions.DualWriterDesiredModes[playlist.GROUPRESOURCE]
|
||||
|
||||
o.ExtraOptions.DevMode = features.IsEnabledGlobally(featuremgmt.FlagGrafanaAPIServerEnsureKubectlAccess)
|
||||
o.ExtraOptions.ExternalAddress = host
|
||||
o.ExtraOptions.APIURL = apiURL
|
||||
|
||||
@@ -27,7 +27,9 @@ import (
|
||||
filestorage "github.com/grafana/grafana/pkg/apiserver/storage/file"
|
||||
"github.com/grafana/grafana/pkg/infra/appcontext"
|
||||
"github.com/grafana/grafana/pkg/infra/db"
|
||||
"github.com/grafana/grafana/pkg/infra/kvstore"
|
||||
"github.com/grafana/grafana/pkg/infra/metrics"
|
||||
"github.com/grafana/grafana/pkg/infra/serverlock"
|
||||
"github.com/grafana/grafana/pkg/infra/tracing"
|
||||
"github.com/grafana/grafana/pkg/middleware"
|
||||
"github.com/grafana/grafana/pkg/modules"
|
||||
@@ -119,7 +121,9 @@ type service struct {
|
||||
tracing *tracing.TracingService
|
||||
metrics prometheus.Registerer
|
||||
|
||||
authorizer *authorizer.GrafanaAuthorizer
|
||||
authorizer *authorizer.GrafanaAuthorizer
|
||||
serverLockService builder.ServerLockService
|
||||
kvStore kvstore.KVStore
|
||||
}
|
||||
|
||||
func ProvideService(
|
||||
@@ -128,7 +132,9 @@ func ProvideService(
|
||||
rr routing.RouteRegister,
|
||||
orgService org.Service,
|
||||
tracing *tracing.TracingService,
|
||||
serverLockService *serverlock.ServerLockService,
|
||||
db db.DB,
|
||||
kvStore kvstore.KVStore,
|
||||
) (*service, error) {
|
||||
s := &service{
|
||||
cfg: cfg,
|
||||
@@ -141,6 +147,7 @@ func ProvideService(
|
||||
tracing: tracing,
|
||||
db: db, // For Unified storage
|
||||
metrics: metrics.ProvideRegisterer(),
|
||||
kvStore: kvStore,
|
||||
}
|
||||
|
||||
// This will be used when running as a dskit service
|
||||
@@ -378,7 +385,10 @@ func (s *service) start(ctx context.Context) error {
|
||||
}
|
||||
|
||||
// Install the API group+version
|
||||
err = builder.InstallAPIs(Scheme, Codecs, server, serverConfig.RESTOptionsGetter, builders, o.StorageOptions, s.metrics)
|
||||
err = builder.InstallAPIs(Scheme, Codecs, server, serverConfig.RESTOptionsGetter, builders, o.StorageOptions,
|
||||
// Required for the dual writer initialization
|
||||
s.metrics, kvstore.WithNamespace(s.kvStore, 0, "storage.dualwriting"), s.serverLockService,
|
||||
)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -15,7 +15,6 @@ import (
|
||||
"github.com/grafana/grafana/pkg/infra/tracing"
|
||||
"github.com/grafana/grafana/pkg/plugins"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/datasource"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/featuretoggle"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/query"
|
||||
"github.com/grafana/grafana/pkg/registry/apis/query/client"
|
||||
"github.com/grafana/grafana/pkg/services/accesscontrol/actest"
|
||||
@@ -23,7 +22,6 @@ import (
|
||||
"github.com/grafana/grafana/pkg/services/apiserver/endpoints/request"
|
||||
"github.com/grafana/grafana/pkg/services/apiserver/options"
|
||||
"github.com/grafana/grafana/pkg/services/featuremgmt"
|
||||
"github.com/grafana/grafana/pkg/setting"
|
||||
testdatasource "github.com/grafana/grafana/pkg/tsdb/grafana-testdata-datasource"
|
||||
)
|
||||
|
||||
@@ -95,13 +93,6 @@ func (p *DummyAPIFactory) MakeAPIServer(_ context.Context, tracer tracing.Tracer
|
||||
tracer,
|
||||
)
|
||||
|
||||
case "featuretoggle.grafana.app":
|
||||
return featuretoggle.NewFeatureFlagAPIBuilder(
|
||||
featuremgmt.WithFeatureManager(setting.FeatureMgmtSettings{}, nil), // none... for now
|
||||
&actest.FakeAccessControl{ExpectedEvaluate: false},
|
||||
&setting.Cfg{},
|
||||
), nil
|
||||
|
||||
case "testdata.datasource.grafana.app":
|
||||
return datasource.NewDataSourceAPIBuilder(
|
||||
plugins.JSONData{
|
||||
|
||||
Reference in New Issue
Block a user