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

Merge branch 'main' into mmandrus/secrets/dek-cache

Browse Source
This commit is contained in:
Michael Mandrus
2026-01-14 14:38:57 -05:00
parent b73869ea9c 72f7bd3900
commit ad7a6e9a7a
5163 changed files with 1042161 additions and 86929 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/server/ring.go
+53
View File
@@ -3,6 +3,7 @@ package server
import (
"context"
"fmt"
"strconv"
"time"
"github.com/grafana/dskit/flagext"
@@ -15,11 +16,15 @@ import (
"github.com/grafana/grafana/pkg/services/featuremgmt"
"github.com/grafana/grafana/pkg/setting"
"github.com/grafana/grafana/pkg/storage/unified/resource"
grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/retry"
"github.com/grpc-ecosystem/go-grpc-middleware/util/metautils"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/trace"
"google.golang.org/grpc"
"google.golang.org/grpc/backoff"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/health/grpc_health_v1"
)
@@ -111,14 +116,25 @@ func newClientPool(clientCfg grpcclient.Config, log log.Logger, reg prometheus.R
Help: "Time spent executing requests to resource server.",
Buckets: prometheus.ExponentialBuckets(0.008, 4, 7),
}, []string{"operation", "status_code"})
factoryRequestRetries := promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
Name: "resource_server_client_request_retries_total",
Help: "Total number of retries for requests to the resource server.",
}, []string{"operation"})
factory := ringclient.PoolInstFunc(func(inst ring.InstanceDesc) (ringclient.PoolClient, error) {
unaryInterceptors, streamInterceptors := grpcclient.Instrument(factoryRequestDuration)
// Add retry interceptors for transient connection issues
unaryInterceptors = append(unaryInterceptors, ringClientRetryInterceptor())
unaryInterceptors = append(unaryInterceptors, ringClientRetryInstrument(factoryRequestRetries))
opts, err := clientCfg.DialOption(unaryInterceptors, streamInterceptors, nil)
if err != nil {
return nil, err
}
opts = append(opts, connectionBackoffOptions())
conn, err := grpc.NewClient(inst.Addr, opts...)
if err != nil {
return nil, fmt.Errorf("failed to dial resource server %s %s: %s", inst.Id, inst.Addr, err)
@@ -135,3 +151,40 @@ func newClientPool(clientCfg grpcclient.Config, log log.Logger, reg prometheus.R
return ringclient.NewPool(resource.RingName, poolCfg, nil, factory, clientsCount, log)
}
// ringClientRetryInterceptor creates an interceptor to perform retries for unary methods.
// It retries on ResourceExhausted and Unavailable codes, which are typical for
// transient connection issues and rate limiting.
func ringClientRetryInterceptor() grpc.UnaryClientInterceptor {
return grpc_retry.UnaryClientInterceptor(
grpc_retry.WithMax(3),
grpc_retry.WithBackoff(grpc_retry.BackoffExponentialWithJitter(time.Second, 0.1)),
grpc_retry.WithCodes(codes.ResourceExhausted, codes.Unavailable),
)
}
// ringClientRetryInstrument creates an interceptor to count retry attempts for metrics.
func ringClientRetryInstrument(metric *prometheus.CounterVec) grpc.UnaryClientInterceptor {
return func(ctx context.Context, method string, req, resp interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
// We can tell if a call is a retry by checking the retry attempt metadata.
attempt, err := strconv.Atoi(metautils.ExtractOutgoing(ctx).Get(grpc_retry.AttemptMetadataKey))
if err == nil && attempt > 0 {
metric.WithLabelValues(method).Inc()
}
return invoker(ctx, method, req, resp, cc, opts...)
}
}
// connectionBackoffOptions configures connection backoff parameters for faster recovery from
// transient connection failures (e.g., during pod restarts).
func connectionBackoffOptions() grpc.DialOption {
return grpc.WithConnectParams(grpc.ConnectParams{
Backoff: backoff.Config{
BaseDelay: 100 * time.Millisecond,
Multiplier: 1.6,
Jitter: 0.2,
MaxDelay: 10 * time.Second,
},
MinConnectTimeout: 5 * time.Second,
})
}
pkg/server/search_server_distributor_test.go
+24 -1
View File
@@ -8,6 +8,7 @@ import (
"net"
"net/http"
"strconv"
"strings"
"sync"
"testing"
"time"
@@ -175,6 +176,28 @@ func TestIntegrationDistributor(t *testing.T) {
}
})
t.Run("RebuildIndexes", func(t *testing.T) {
instanceResponseCount := make(map[string]int)
// simulate RebuildIndexes for a single namespace
testNamespace := testNamespaces[0]
req := &resourcepb.RebuildIndexesRequest{
Namespace: testNamespace,
Keys: []*resourcepb.ResourceKey{{
Namespace: testNamespace,
Group: "folder.grafana.app",
Resource: "folders",
}},
}
distributorRes := getDistributorResponse(t, req, distributorServer.resourceClient.RebuildIndexes, instanceResponseCount)
require.Nil(t, distributorRes.Error)
// assert all instances got the response by looking at the merged details
count := strings.Count(distributorRes.Details, "{instance:")
require.Equal(t, len(testServers), count)
})
var wg sync.WaitGroup
for _, testServer := range testServers {
wg.Add(1)
@@ -366,7 +389,7 @@ func createBaselineServer(t *testing.T, dbType, dbConnStr string, testNamespaces
require.NoError(t, err)
tracer := noop.NewTracerProvider().Tracer("test-tracer")
require.NoError(t, err)
searchOpts, err := search.NewSearchOptions(features, cfg, tracer, docBuilders, nil, nil)
searchOpts, err := search.NewSearchOptions(features, cfg, docBuilders, nil, nil)
require.NoError(t, err)
server, err := sql.NewResourceServer(sql.ServerOptions{
DB: nil,
pkg/server/test_env.go
+33 -29
View File
@@ -3,6 +3,7 @@ package server
import (
"github.com/stretchr/testify/mock"
githubconnection "github.com/grafana/grafana/apps/provisioning/pkg/connection/github"
"github.com/grafana/grafana/apps/provisioning/pkg/repository/github"
"github.com/grafana/grafana/apps/secret/pkg/decrypt"
"github.com/grafana/grafana/pkg/infra/db"
@@ -34,24 +35,26 @@ func ProvideTestEnv(
featureMgmt featuremgmt.FeatureToggles,
resourceClient resource.ResourceClient,
idService auth.IDService,
githubFactory *github.Factory,
githubRepoFactory *github.Factory,
githubConnectionFactory githubconnection.GithubFactory,
decryptService decrypt.DecryptService,
) (*TestEnv, error) {
return &TestEnv{
TestingT: testingT,
Server: server,
SQLStore: db,
Cfg: cfg,
NotificationService: ns,
GRPCServer: grpcServer,
PluginRegistry: pluginRegistry,
HTTPClientProvider: httpClientProvider,
OAuthTokenService: oAuthTokenService,
FeatureToggles: featureMgmt,
ResourceClient: resourceClient,
IDService: idService,
GitHubFactory: githubFactory,
DecryptService: decryptService,
TestingT: testingT,
Server: server,
SQLStore: db,
Cfg: cfg,
NotificationService: ns,
GRPCServer: grpcServer,
PluginRegistry: pluginRegistry,
HTTPClientProvider: httpClientProvider,
OAuthTokenService: oAuthTokenService,
FeatureToggles: featureMgmt,
ResourceClient: resourceClient,
IDService: idService,
GithubRepoFactory: githubRepoFactory,
GithubConnectionFactory: githubConnectionFactory,
DecryptService: decryptService,
}, nil
}
@@ -60,18 +63,19 @@ type TestEnv struct {
mock.TestingT
Cleanup(func())
}
Server *Server
SQLStore db.DB
Cfg *setting.Cfg
NotificationService *notifications.NotificationServiceMock
GRPCServer grpcserver.Provider
PluginRegistry registry.Service
HTTPClientProvider httpclient.Provider
OAuthTokenService *oauthtokentest.Service
RequestMiddleware web.Middleware
FeatureToggles featuremgmt.FeatureToggles
ResourceClient resource.ResourceClient
IDService auth.IDService
GitHubFactory *github.Factory
DecryptService decrypt.DecryptService
Server *Server
SQLStore db.DB
Cfg *setting.Cfg
NotificationService *notifications.NotificationServiceMock
GRPCServer grpcserver.Provider
PluginRegistry registry.Service
HTTPClientProvider httpclient.Provider
OAuthTokenService *oauthtokentest.Service
RequestMiddleware web.Middleware
FeatureToggles featuremgmt.FeatureToggles
ResourceClient resource.ResourceClient
IDService auth.IDService
GithubRepoFactory *github.Factory
GithubConnectionFactory githubconnection.GithubFactory
DecryptService decrypt.DecryptService
}
pkg/server/wire.go
+11 -1
View File
@@ -15,6 +15,7 @@ import (
"go.opentelemetry.io/otel/trace"
sdkhttpclient "github.com/grafana/grafana-plugin-sdk-go/backend/httpclient"
ghconnection "github.com/grafana/grafana/apps/provisioning/pkg/connection/github"
"github.com/grafana/grafana/apps/provisioning/pkg/repository/github"
"github.com/grafana/grafana/pkg/api"
"github.com/grafana/grafana/pkg/api/avatar"
@@ -128,6 +129,7 @@ import (
"github.com/grafana/grafana/pkg/services/pluginsintegration/pluginaccesscontrol"
"github.com/grafana/grafana/pkg/services/preference/prefimpl"
promTypeMigration "github.com/grafana/grafana/pkg/services/promtypemigration"
"github.com/grafana/grafana/pkg/services/provisioning"
"github.com/grafana/grafana/pkg/services/publicdashboards"
publicdashboardsApi "github.com/grafana/grafana/pkg/services/publicdashboards/api"
publicdashboardsStore "github.com/grafana/grafana/pkg/services/publicdashboards/database"
@@ -181,6 +183,7 @@ import (
secretencryption "github.com/grafana/grafana/pkg/storage/secret/encryption"
secretmetadata "github.com/grafana/grafana/pkg/storage/secret/metadata"
secretmigrator "github.com/grafana/grafana/pkg/storage/secret/migrator"
unifiedmigrations "github.com/grafana/grafana/pkg/storage/unified/migrations"
"github.com/grafana/grafana/pkg/storage/unified/resource"
unifiedsearch "github.com/grafana/grafana/pkg/storage/unified/search"
"github.com/grafana/grafana/pkg/tsdb/azuremonitor"
@@ -237,7 +240,9 @@ var wireBasicSet = wire.NewSet(
uss.ProvideService,
wire.Bind(new(usagestats.Service), new(*uss.UsageStats)),
validator.ProvideService,
legacy.ProvideLegacyMigrator,
provisioning.ProvideStubProvisioningService,
legacy.ProvideMigratorDashboardAccessor,
unifiedmigrations.ProvideUnifiedMigrator,
pluginsintegration.WireSet,
pluginDashboards.ProvideFileStoreManager,
wire.Bind(new(pluginDashboards.FileStore), new(*pluginDashboards.FileStoreManager)),
@@ -276,6 +281,7 @@ var wireBasicSet = wire.NewSet(
store.ProvideService,
store.ProvideSystemUsersService,
live.ProvideService,
live.ProvideDashboardActivityChannel,
pushhttp.ProvideService,
contexthandler.ProvideService,
ldapservice.ProvideService,
@@ -292,6 +298,7 @@ var wireBasicSet = wire.NewSet(
notifications.ProvideService,
notifications.ProvideSmtpService,
github.ProvideFactory,
ghconnection.ProvideFactory,
tracing.ProvideService,
tracing.ProvideTracingConfig,
wire.Bind(new(tracing.Tracer), new(*tracing.TracingService)),
@@ -325,6 +332,7 @@ var wireBasicSet = wire.NewSet(
dashsnapstore.ProvideStore,
wire.Bind(new(dashboardsnapshots.Service), new(*dashsnapsvc.ServiceImpl)),
dashsnapsvc.ProvideService,
datasourceservice.ProvideDataSourceRetriever,
datasourceservice.ProvideService,
wire.Bind(new(datasources.DataSourceService), new(*datasourceservice.Service)),
datasourceservice.ProvideLegacyDataSourceLookup,
@@ -344,6 +352,7 @@ var wireBasicSet = wire.NewSet(
dashboardservice.ProvideDashboardService,
dashboardservice.ProvideDashboardProvisioningService,
dashboardservice.ProvideDashboardPluginService,
dashboardservice.ProvideDashboardAccessService,
dashboardstore.ProvideDashboardStore,
folderimpl.ProvideService,
wire.Bind(new(folder.Service), new(*folderimpl.Service)),
@@ -465,6 +474,7 @@ var wireBasicSet = wire.NewSet(
// Unified storage
resource.ProvideStorageMetrics,
resource.ProvideIndexMetrics,
unifiedmigrations.ProvideUnifiedStorageMigrationService,
// Kubernetes API server
grafanaapiserver.WireSet,
apiregistry.WireSet,
pkg/server/wire_gen.go
Generated
+160 -100
View File
File diff suppressed because one or more lines are too long
pkg/server/wireexts_oss.go
+6 -2
View File
@@ -63,14 +63,17 @@ import (
"github.com/grafana/grafana/pkg/services/user"
"github.com/grafana/grafana/pkg/services/validations"
"github.com/grafana/grafana/pkg/setting"
"github.com/grafana/grafana/pkg/storage/legacysql"
"github.com/grafana/grafana/pkg/storage/unified"
"github.com/grafana/grafana/pkg/storage/unified/resource"
search2 "github.com/grafana/grafana/pkg/storage/unified/search"
"github.com/grafana/grafana/pkg/storage/unified/search/builders"
"github.com/grafana/grafana/pkg/storage/unified/sql"
)
var provisioningExtras = wire.NewSet(
extras.ProvideProvisioningOSSRepositoryExtras,
extras.ProvideProvisioningOSSConnectionExtras,
)
var configProviderExtras = wire.NewSet(
@@ -99,6 +102,7 @@ var wireExtsBasicSet = wire.NewSet(
wire.Bind(new(validations.DataSourceRequestURLValidator), new(*validations.OSSDataSourceRequestURLValidator)),
provisioning.ProvideService,
wire.Bind(new(provisioning.ProvisioningService), new(*provisioning.ProvisioningServiceImpl)),
legacysql.NewDatabaseProvider,
backgroundsvcs.ProvideBackgroundServiceRegistry,
wire.Bind(new(registry.BackgroundServiceRegistry), new(*backgroundsvcs.BackgroundServiceRegistry)),
migrations.ProvideOSSMigrations,
@@ -137,8 +141,8 @@ var wireExtsBasicSet = wire.NewSet(
wire.Bind(new(auth.IDSigner), new(*idimpl.LocalSigner)),
manager.ProvideInstaller,
wire.Bind(new(plugins.Installer), new(*manager.PluginInstaller)),
search2.ProvideDashboardStats,
wire.Bind(new(search2.DashboardStats), new(*search2.OssDashboardStats)),
builders.ProvideDashboardStats,
wire.Bind(new(builders.DashboardStats), new(*builders.OssDashboardStats)),
search2.ProvideDocumentBuilders,
sandbox.ProvideService,
wire.Bind(new(sandbox.Sandbox), new(*sandbox.Service)),