public/grafana
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

UniStore/Large Objects: Make threshold configurable (#101774)

Browse Source 
* Make blob threshold configurable
* Test condition for deconstructing large objects
* Refactor blob threshold naming
This commit is contained in:
Arati R.
2025-04-08 10:50:35 +02:00
committed by GitHub
parent ba653c22c3
commit df537d6f0f
13 changed files with 125 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/registry/apis/dashboard/large.go
+4 -4
View File
@@ -14,15 +14,15 @@ import (
"github.com/grafana/grafana/pkg/storage/unified/apistore"
)
func NewDashboardLargeObjectSupport(scheme *runtime.Scheme) *apistore.BasicLargeObjectSupport {
func NewDashboardLargeObjectSupport(scheme *runtime.Scheme, threshold int) *apistore.BasicLargeObjectSupport {
return &apistore.BasicLargeObjectSupport{
TheGroupResource: dashboardV0.DashboardResourceInfo.GroupResource(),
// byte size, while testing lets do almost everything (10bytes)
ThresholdSize: 10,
// Byte size above which an object is considered large.
ThresholdBytes: threshold,
// 10mb -- we should check what the largest ones are... might be bigger
MaxByteSize: 10 * 1024 * 1024,
MaxBytes: 10 * 1024 * 1024,
ReduceSpec: func(obj runtime.Object) error {
meta, err := utils.MetaAccessor(obj)
pkg/registry/apis/dashboard/large_test.go
+1 -1
View File
@@ -41,7 +41,7 @@ func TestLargeDashboardSupport(t *testing.T) {
err = dashboardv1alpha1.AddToScheme(scheme)
require.NoError(t, err)
largeObject := NewDashboardLargeObjectSupport(scheme)
largeObject := NewDashboardLargeObjectSupport(scheme, 0)
// Convert the dashboard to a small value
err = largeObject.ReduceSpec(dash)
pkg/registry/apis/dashboard/register.go
+2 -2
View File
@@ -390,10 +390,10 @@ func (b *DashboardsAPIBuilder) UpdateAPIGroupInfo(apiGroupInfo *genericapiserver
// Split dashboards when they are large
var largeObjects apistore.LargeObjectSupport
if b.features.IsEnabledGlobally(featuremgmt.FlagUnifiedStorageBigObjectsSupport) {
largeObjects = NewDashboardLargeObjectSupport(opts.Scheme)
largeObjects = NewDashboardLargeObjectSupport(opts.Scheme, opts.StorageOpts.BlobThresholdBytes)
storageOpts.LargeObjectSupport = largeObjects
}
opts.StorageOptions(v0alpha1.DashboardResourceInfo.GroupResource(), storageOpts)
opts.StorageOptsRegister(v0alpha1.DashboardResourceInfo.GroupResource(), storageOpts)
// v0alpha1
if err := b.storageForVersion(apiGroupInfo, opts, largeObjects,
pkg/registry/apis/folders/register.go
+1 -1
View File
@@ -155,7 +155,7 @@ func (b *FolderAPIBuilder) UpdateAPIGroupInfo(apiGroupInfo *genericapiserver.API
cfg: b.cfg,
}
opts.StorageOptions(resourceInfo.GroupResource(), apistore.StorageOptions{
opts.StorageOptsRegister(resourceInfo.GroupResource(), apistore.StorageOptions{
EnableFolderSupport: true,
RequireDeprecatedInternalID: true})
pkg/services/apiserver/builder/common.go
+7 -5
View File
@@ -16,6 +16,7 @@ import (
"k8s.io/kube-openapi/pkg/spec3"
grafanarest "github.com/grafana/grafana/pkg/apiserver/rest"
"github.com/grafana/grafana/pkg/services/apiserver/options"
"github.com/grafana/grafana/pkg/storage/unified/apistore"
)
@@ -73,11 +74,12 @@ type APIGroupPostStartHookProvider interface {
}
type APIGroupOptions struct {
Scheme *runtime.Scheme
OptsGetter generic.RESTOptionsGetter
DualWriteBuilder grafanarest.DualWriteBuilder
MetricsRegister prometheus.Registerer
StorageOptions apistore.StorageOptionsRegister
Scheme *runtime.Scheme
OptsGetter generic.RESTOptionsGetter
DualWriteBuilder grafanarest.DualWriteBuilder
MetricsRegister prometheus.Registerer
StorageOptsRegister apistore.StorageOptionsRegister
StorageOpts *options.StorageOptions
}
// Builders that implement OpenAPIPostProcessor are given a chance to modify the schema directly
pkg/services/apiserver/builder/helper.go
+6 -5
View File
@@ -396,11 +396,12 @@ func InstallAPIs(
g := genericapiserver.NewDefaultAPIGroupInfo(group, scheme, metav1.ParameterCodec, codecs)
for _, b := range buildersForGroup {
if err := b.UpdateAPIGroupInfo(&g, APIGroupOptions{
Scheme: scheme,
OptsGetter: optsGetter,
DualWriteBuilder: dualWrite,
MetricsRegister: reg,
StorageOptions: optsregister,
Scheme: scheme,
OptsGetter: optsGetter,
DualWriteBuilder: dualWrite,
MetricsRegister: reg,
StorageOptsRegister: optsregister,
StorageOpts: storageOpts,
}); err != nil {
return err
}
pkg/services/apiserver/config.go
+1
View File
@@ -56,6 +56,7 @@ func applyGrafanaConfig(cfg *setting.Cfg, features featuremgmt.FeatureToggles, o
o.StorageOptions.DataPath = apiserverCfg.Key("storage_path").MustString(filepath.Join(cfg.DataPath, "grafana-apiserver"))
o.StorageOptions.Address = apiserverCfg.Key("address").MustString(o.StorageOptions.Address)
o.StorageOptions.BlobStoreURL = apiserverCfg.Key("blob_url").MustString(o.StorageOptions.BlobStoreURL)
o.StorageOptions.BlobThresholdBytes = apiserverCfg.Key("blob_threshold_bytes").MustInt(o.StorageOptions.BlobThresholdBytes)
// unified storage configs look like
// [unified_storage.<group>.<resource>]
pkg/services/apiserver/options/storage.go
+6
View File
@@ -28,6 +28,8 @@ const (
// Deprecated: legacy is a shim that is no longer necessary
StorageTypeLegacy StorageType = "legacy"
BlobThresholdDefault int = 0
)
type StorageOptions struct {
@@ -50,6 +52,9 @@ type StorageOptions struct {
// s3://my-bucket?region=us-west-1 (using default credentials)
// azblob://my-container
BlobStoreURL string
// Optional blob storage field. When an object's size in bytes exceeds the threshold
// value, it is considered large and gets partially stored in blob storage.
BlobThresholdBytes int
// {resource}.{group} = 1|2|3|4
UnifiedStorageConfig map[string]setting.UnifiedStorageConfig
@@ -61,6 +66,7 @@ func NewStorageOptions() *StorageOptions {
Address: "localhost:10000",
GrpcClientAuthenticationTokenNamespace: "*",
GrpcClientAuthenticationAllowInsecure: false,
BlobThresholdBytes: BlobThresholdDefault,
}
}
pkg/storage/unified/apistore/fake_large.go
+38
View File
@@ -0,0 +1,38 @@
package apistore
import (
"context"
dashboardv1alpha1 "github.com/grafana/grafana/apps/dashboard/pkg/apis/dashboard/v1alpha1"
"github.com/grafana/grafana/pkg/apimachinery/utils"
"github.com/grafana/grafana/pkg/storage/unified/resource"
"k8s.io/apimachinery/pkg/runtime/schema"
)
type LargeObjectSupportFake struct {
threshold int
deconstructed bool
reconstructed bool
}
func (s *LargeObjectSupportFake) GroupResource() schema.GroupResource {
return dashboardv1alpha1.DashboardResourceInfo.GroupResource()
}
func (s *LargeObjectSupportFake) Threshold() int {
return s.threshold
}
func (s *LargeObjectSupportFake) MaxSize() int {
return 10 * 1024 * 1024
}
func (s *LargeObjectSupportFake) Deconstruct(ctx context.Context, key *resource.ResourceKey, client resource.BlobStoreClient, obj utils.GrafanaMetaAccessor, raw []byte) error {
s.deconstructed = true
return nil
}
func (s *LargeObjectSupportFake) Reconstruct(ctx context.Context, key *resource.ResourceKey, client resource.BlobStoreClient, obj utils.GrafanaMetaAccessor) error {
s.reconstructed = true
return nil
}
pkg/storage/unified/apistore/large.go
+4 -4
View File
@@ -38,8 +38,8 @@ var _ LargeObjectSupport = (*BasicLargeObjectSupport)(nil)
type BasicLargeObjectSupport struct {
TheGroupResource schema.GroupResource
ThresholdSize int
MaxByteSize int
ThresholdBytes int
MaxBytes int
// Mutate the spec so it only has the small properties
ReduceSpec func(obj runtime.Object) error
@@ -55,12 +55,12 @@ func (s *BasicLargeObjectSupport) GroupResource() schema.GroupResource {
// Threshold implements LargeObjectSupport.
func (s *BasicLargeObjectSupport) Threshold() int {
return s.ThresholdSize
return s.ThresholdBytes
}
// MaxSize implements LargeObjectSupport.
func (s *BasicLargeObjectSupport) MaxSize() int {
return s.MaxByteSize
return s.MaxBytes
}
// Deconstruct implements LargeObjectSupport.
pkg/storage/unified/apistore/prepare.go
+4 -6
View File
@@ -175,12 +175,10 @@ func (s *Storage) prepareObjectForUpdate(ctx context.Context, updateObject runti
func (s *Storage) handleLargeResources(ctx context.Context, obj utils.GrafanaMetaAccessor, buf bytes.Buffer) ([]byte, error) {
support := s.opts.LargeObjectSupport
if support != nil {
size := buf.Len()
if size > support.Threshold() {
if support.MaxSize() > 0 && size > support.MaxSize() {
return nil, fmt.Errorf("request object is too big (%s > %s)", formatBytes(size), formatBytes(support.MaxSize()))
}
size := buf.Len()
if support != nil && size > support.Threshold() {
if support.MaxSize() > 0 && size > support.MaxSize() {
return nil, fmt.Errorf("request object is too big (%s > %s)", formatBytes(size), formatBytes(support.MaxSize()))
}
key := &resource.ResourceKey{
pkg/storage/unified/apistore/prepare_test.go
+46
View File
@@ -297,3 +297,49 @@ func getPreparedObject(t *testing.T, ctx context.Context, s *Storage, obj runtim
require.NoError(t, err)
return meta
}
func TestPrepareLargeObjectForStorage(t *testing.T) {
_ = v1alpha1.AddToScheme(scheme)
node, err := snowflake.NewNode(rand.Int63n(1024))
require.NoError(t, err)
ctx := authtypes.WithAuthInfo(context.Background(), &identity.StaticRequester{UserID: 1, UserUID: "user-uid", Type: authtypes.TypeUser})
dashboard := v1alpha1.Dashboard{}
dashboard.Name = "test-name"
t.Run("Should deconstruct object if size is over threshold", func(t *testing.T) {
los := LargeObjectSupportFake{
threshold: 0,
}
f := &Storage{
codec: apitesting.TestCodec(codecs, v1alpha1.DashboardResourceInfo.GroupVersion()),
snowflake: node,
opts: StorageOptions{
LargeObjectSupport: &los,
},
}
_, err := f.prepareObjectForStorage(ctx, dashboard.DeepCopyObject())
require.Nil(t, err)
require.True(t, los.deconstructed)
})
t.Run("Should not deconstruct object if size is under threshold", func(t *testing.T) {
los := LargeObjectSupportFake{
threshold: 1000,
}
f := &Storage{
codec: apitesting.TestCodec(codecs, v1alpha1.DashboardResourceInfo.GroupVersion()),
snowflake: node,
opts: StorageOptions{
LargeObjectSupport: &los,
},
}
_, err := f.prepareObjectForStorage(ctx, dashboard.DeepCopyObject())
require.Nil(t, err)
require.False(t, los.deconstructed)
})
}
pkg/storage/unified/client.go
+5 -4
View File
@@ -55,10 +55,11 @@ func ProvideUnifiedStorageClient(opts *Options, storageMetrics *resource.Storage
// See: apiserver.ApplyGrafanaConfig(cfg, features, o)
apiserverCfg := opts.Cfg.SectionWithEnvOverrides("grafana-apiserver")
client, err := newClient(options.StorageOptions{
StorageType: options.StorageType(apiserverCfg.Key("storage_type").MustString(string(options.StorageTypeUnified))),
DataPath: apiserverCfg.Key("storage_path").MustString(filepath.Join(opts.Cfg.DataPath, "grafana-apiserver")),
Address: apiserverCfg.Key("address").MustString(""), // client address
BlobStoreURL: apiserverCfg.Key("blob_url").MustString(""),
StorageType: options.StorageType(apiserverCfg.Key("storage_type").MustString(string(options.StorageTypeUnified))),
DataPath: apiserverCfg.Key("storage_path").MustString(filepath.Join(opts.Cfg.DataPath, "grafana-apiserver")),
Address: apiserverCfg.Key("address").MustString(""), // client address
BlobStoreURL: apiserverCfg.Key("blob_url").MustString(""),
BlobThresholdBytes: apiserverCfg.Key("blob_threshold_bytes").MustInt(options.BlobThresholdDefault),
}, opts.Cfg, opts.Features, opts.DB, opts.Tracer, opts.Reg, opts.Authzc, opts.Docs, storageMetrics, indexMetrics)
if err == nil {
// Used to get the folder stats