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

unified-storage: setup ring to shard requests (#103783)

Browse Source 
* Updates the instrumentation_server service to use mux instead of the builtin router, and have it store the router in the module server: this is so we can register the /ring endpoint to check the status of the ring
* Create a new Ring service that depends on the instrumentation server and declares it as a dependency for the storage server
* Create standalone MemberlistKV service for Ring service to use
* Update the storage server Search and GetStats handler to distribute requests if applicable
This commit is contained in:
Will Assis
2025-04-25 13:08:44 -04:00
committed by GitHub
parent ff7b923d33
commit 4adebd6058
19 changed files with 654 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/server/instrumentation_service.go
+9 -8
View File
@@ -6,6 +6,7 @@ import (
"net/http"
"time"
"github.com/gorilla/mux"
"github.com/grafana/dskit/services"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/setting"
@@ -22,14 +23,14 @@ type instrumentationService struct {
promGatherer prometheus.Gatherer
}
func NewInstrumentationService(log log.Logger, cfg *setting.Cfg, promGatherer prometheus.Gatherer) (*instrumentationService, error) {
s := &instrumentationService{log: log, cfg: cfg, promGatherer: promGatherer}
func (ms *ModuleServer) initInstrumentationServer() (*instrumentationService, error) {
s := &instrumentationService{log: ms.log, cfg: ms.cfg, promGatherer: ms.promGatherer}
s.httpServ, ms.httpServerRouter = s.newInstrumentationServer()
s.BasicService = services.NewBasicService(s.start, s.running, s.stop)
return s, nil
}
func (s *instrumentationService) start(ctx context.Context) error {
s.httpServ = s.newInstrumentationServer(ctx)
s.errChan = make(chan error)
go func() {
s.errChan <- s.httpServ.ListenAndServe()
@@ -56,17 +57,17 @@ func (s *instrumentationService) stop(failureReason error) error {
return nil
}
func (s *instrumentationService) newInstrumentationServer(ctx context.Context) *http.Server {
router := http.NewServeMux()
func (s *instrumentationService) newInstrumentationServer() (*http.Server, *mux.Router) {
router := mux.NewRouter()
router.Handle("/metrics", promhttp.HandlerFor(s.promGatherer, promhttp.HandlerOpts{EnableOpenMetrics: true}))
addr := net.JoinHostPort(s.cfg.HTTPAddr, s.cfg.HTTPPort)
srv := &http.Server{
// 5s timeout for header reads to avoid Slowloris attacks (https://thetooth.io/blog/slowloris-attack/)
ReadHeaderTimeout: 5 * time.Second,
Addr: ":" + s.cfg.HTTPPort,
Addr: addr,
Handler: router,
BaseContext: func(_ net.Listener) context.Context { return ctx },
}
return srv
return srv, router
}
pkg/server/instrumentation_service_test.go
+6 -1
View File
@@ -18,7 +18,12 @@ import (
func TestRunInstrumentationService(t *testing.T) {
cfg := setting.NewCfg()
cfg.HTTPPort = "3001"
s, err := NewInstrumentationService(log.New("test-logger"), cfg, prometheus.DefaultGatherer)
ms := ModuleServer{
log: log.New("test-logger"),
cfg: cfg,
promGatherer: prometheus.DefaultGatherer,
}
s, err := ms.initInstrumentationServer()
require.NoError(t, err)
ctx, cancel := context.WithTimeout(context.Background(), 300*time.Second)
pkg/server/memberlist.go
+53
View File
@@ -0,0 +1,53 @@
package server
import (
"github.com/grafana/dskit/dns"
"github.com/grafana/dskit/flagext"
"github.com/grafana/dskit/kv"
"github.com/grafana/dskit/kv/codec"
"github.com/grafana/dskit/kv/memberlist"
"github.com/grafana/dskit/ring"
"github.com/grafana/dskit/services"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/setting"
"github.com/prometheus/client_golang/prometheus"
)
func (ms *ModuleServer) initMemberlistKV() (services.Service, error) {
logger := log.New("memberlist")
dnsProviderReg := prometheus.WrapRegistererWithPrefix(
"grafana",
prometheus.WrapRegistererWith(
prometheus.Labels{"component": "memberlist"},
ms.registerer,
),
)
dnsProvider := dns.NewProvider(logger, dnsProviderReg, dns.GolangResolverType)
KVStore := kv.Config{Store: "memberlist"}
memberlistKVsvc := memberlist.NewKVInitService(toMemberlistConfig(ms.cfg), logger, dnsProvider, ms.registerer)
KVStore.MemberlistKV = memberlistKVsvc.GetMemberlistKV
ms.MemberlistKVConfig = KVStore
return memberlistKVsvc, nil
}
func toMemberlistConfig(cfg *setting.Cfg) *memberlist.KVConfig {
memberlistKVcfg := &memberlist.KVConfig{}
flagext.DefaultValues(memberlistKVcfg)
memberlistKVcfg.Codecs = []codec.Codec{
ring.GetCodec(),
}
if cfg.MemberlistBindAddr != "" {
memberlistKVcfg.TCPTransport.BindAddrs = []string{cfg.MemberlistBindAddr}
}
if cfg.MemberlistAdvertiseAddr != "" {
memberlistKVcfg.AdvertiseAddr = cfg.MemberlistAdvertiseAddr
}
memberlistKVcfg.JoinMembers = []string{cfg.MemberlistJoinMember}
return memberlistKVcfg
}
pkg/server/module_server.go
+16 -4
View File
@@ -9,6 +9,8 @@ import (
"strconv"
"sync"
"github.com/gorilla/mux"
"github.com/grafana/dskit/kv"
"github.com/prometheus/client_golang/prometheus"
"github.com/grafana/dskit/services"
@@ -32,10 +34,11 @@ func NewModule(opts Options,
cfg *setting.Cfg,
storageMetrics *resource.StorageMetrics,
indexMetrics *resource.BleveIndexMetrics,
reg prometheus.Registerer,
promGatherer prometheus.Gatherer,
license licensing.Licensing,
) (*ModuleServer, error) {
s, err := newModuleServer(opts, apiOpts, features, cfg, storageMetrics, indexMetrics, promGatherer, license)
s, err := newModuleServer(opts, apiOpts, features, cfg, storageMetrics, indexMetrics, reg, promGatherer, license)
if err != nil {
return nil, err
}
@@ -47,7 +50,7 @@ func NewModule(opts Options,
return s, nil
}
func newModuleServer(opts Options, apiOpts api.ServerOptions, features featuremgmt.FeatureToggles, cfg *setting.Cfg, storageMetrics *resource.StorageMetrics, indexMetrics *resource.BleveIndexMetrics, promGatherer prometheus.Gatherer, license licensing.Licensing) (*ModuleServer, error) {
func newModuleServer(opts Options, apiOpts api.ServerOptions, features featuremgmt.FeatureToggles, cfg *setting.Cfg, storageMetrics *resource.StorageMetrics, indexMetrics *resource.BleveIndexMetrics, reg prometheus.Registerer, promGatherer prometheus.Gatherer, license licensing.Licensing) (*ModuleServer, error) {
rootCtx, shutdownFn := context.WithCancel(context.Background())
s := &ModuleServer{
@@ -66,6 +69,7 @@ func newModuleServer(opts Options, apiOpts api.ServerOptions, features featuremg
storageMetrics: storageMetrics,
indexMetrics: indexMetrics,
promGatherer: promGatherer,
registerer: reg,
license: license,
}
@@ -98,6 +102,11 @@ type ModuleServer struct {
buildBranch string
promGatherer prometheus.Gatherer
registerer prometheus.Registerer
MemberlistKVConfig kv.Config
httpServerRouter *mux.Router
distributor *resource.Distributor
}
// init initializes the server and its services.
@@ -136,9 +145,12 @@ func (s *ModuleServer) Run() error {
if m.IsModuleEnabled(modules.All) || m.IsModuleEnabled(modules.Core) || m.IsModuleEnabled(modules.FrontendServer) {
return services.NewBasicService(nil, nil, nil).WithName(modules.InstrumentationServer), nil
}
return NewInstrumentationService(s.log, s.cfg, s.promGatherer)
return s.initInstrumentationServer()
})
m.RegisterModule(modules.MemberlistKV, s.initMemberlistKV)
m.RegisterModule(modules.StorageRing, s.initRing)
m.RegisterModule(modules.Core, func() (services.Service, error) {
return NewService(s.cfg, s.opts, s.apiOpts)
})
@@ -157,7 +169,7 @@ func (s *ModuleServer) Run() error {
if err != nil {
return nil, err
}
return sql.ProvideUnifiedStorageGrpcService(s.cfg, s.features, nil, s.log, nil, docBuilders, s.storageMetrics, s.indexMetrics)
return sql.ProvideUnifiedStorageGrpcService(s.cfg, s.features, nil, s.log, nil, docBuilders, s.storageMetrics, s.indexMetrics, s.distributor)
})
m.RegisterModule(modules.ZanzanaServer, func() (services.Service, error) {
pkg/server/ring.go
+222
View File
@@ -0,0 +1,222 @@
package server
import (
"context"
"fmt"
"net"
"os"
"strconv"
"time"
"github.com/grafana/dskit/flagext"
"github.com/grafana/dskit/grpcclient"
"github.com/grafana/dskit/kv"
"github.com/grafana/dskit/netutil"
"github.com/grafana/dskit/ring"
ringclient "github.com/grafana/dskit/ring/client"
"github.com/grafana/dskit/services"
"github.com/grafana/grafana/pkg/infra/log"
"github.com/grafana/grafana/pkg/setting"
"github.com/grafana/grafana/pkg/storage/unified/resource"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"google.golang.org/grpc"
"google.golang.org/grpc/health/grpc_health_v1"
)
const ringKey = "storage-ring"
const ringName = "unified_storage"
const numTokens = 128
const heartbeatTimeout = time.Minute
var metricsPrefix = ringName + "_"
func (ms *ModuleServer) initRing() (services.Service, error) {
if !ms.cfg.EnableSharding {
return nil, nil
}
logger := log.New("resource-server-ring")
reg := prometheus.WrapRegistererWithPrefix(metricsPrefix, ms.registerer)
grpcclientcfg := &grpcclient.Config{}
flagext.DefaultValues(grpcclientcfg)
pool := newClientPool(*grpcclientcfg, logger, reg)
ringStore, err := kv.NewClient(
ms.MemberlistKVConfig,
ring.GetCodec(),
kv.RegistererWithKVName(reg, ringName),
logger,
)
if err != nil {
return nil, fmt.Errorf("failed to create KV store client: %s", err)
}
lifecyclerCfg, err := toLifecyclerConfig(ms.cfg, logger)
if err != nil {
return nil, fmt.Errorf("failed to initialize storage-ring lifecycler config: %s", err)
}
// Define lifecycler delegates in reverse order (last to be called defined first because they're
// chained via "next delegate").
delegate := ring.BasicLifecyclerDelegate(ring.NewInstanceRegisterDelegate(ring.JOINING, numTokens))
delegate = ring.NewLeaveOnStoppingDelegate(delegate, logger)
delegate = ring.NewAutoForgetDelegate(heartbeatTimeout*2, delegate, logger)
lifecycler, err := ring.NewBasicLifecycler(
lifecyclerCfg,
ringName,
ringKey,
ringStore,
delegate,
logger,
reg,
)
if err != nil {
return nil, fmt.Errorf("failed to initialize storage-ring lifecycler: %s", err)
}
storageRing, err := ring.NewWithStoreClientAndStrategy(
toRingConfig(ms.cfg, ms.MemberlistKVConfig),
ringName,
ringKey,
ringStore,
ring.NewIgnoreUnhealthyInstancesReplicationStrategy(),
reg,
logger,
)
if err != nil {
return nil, fmt.Errorf("failed to initialize storage-ring ring: %s", err)
}
startFn := func(ctx context.Context) error {
err = storageRing.StartAsync(ctx)
if err != nil {
return fmt.Errorf("failed to start the ring: %s", err)
}
err = lifecycler.StartAsync(ctx)
if err != nil {
return fmt.Errorf("failed to start the lifecycler: %s", err)
}
err = pool.StartAsync(ctx)
if err != nil {
return fmt.Errorf("failed to start the ring client pool: %s", err)
}
logger.Info("waiting until resource server is JOINING in the ring")
ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
defer cancel()
if err := ring.WaitInstanceState(ctx, storageRing, lifecycler.GetInstanceID(), ring.JOINING); err != nil {
return fmt.Errorf("error switching to JOINING in the ring: %s", err)
}
logger.Info("resource server is JOINING in the ring")
if err := lifecycler.ChangeState(ctx, ring.ACTIVE); err != nil {
return fmt.Errorf("error switching to ACTIVE in the ring: %s", err)
}
logger.Info("resource server is ACTIVE in the ring")
return nil
}
ms.distributor = &resource.Distributor{
ClientPool: pool,
Ring: storageRing,
Lifecycler: lifecycler,
}
ms.httpServerRouter.Path("/ring").Methods("GET", "POST").Handler(storageRing)
svc := services.NewIdleService(startFn, nil)
return svc, nil
}
func toLifecyclerConfig(cfg *setting.Cfg, logger log.Logger) (ring.BasicLifecyclerConfig, error) {
instanceAddr, err := ring.GetInstanceAddr(cfg.MemberlistBindAddr, netutil.PrivateNetworkInterfacesWithFallback([]string{"eth0", "en0"}, logger), logger, true)
if err != nil {
return ring.BasicLifecyclerConfig{}, err
}
instanceId := cfg.InstanceID
if instanceId == "" {
hostname, err := os.Hostname()
if err != nil {
return ring.BasicLifecyclerConfig{}, err
}
instanceId = hostname
}
_, grpcPortStr, err := net.SplitHostPort(cfg.GRPCServer.Address)
if err != nil {
return ring.BasicLifecyclerConfig{}, fmt.Errorf("could not get grpc port from grpc server address: %s", err)
}
grpcPort, err := strconv.Atoi(grpcPortStr)
if err != nil {
return ring.BasicLifecyclerConfig{}, fmt.Errorf("error converting grpc address port to int: %s", err)
}
return ring.BasicLifecyclerConfig{
Addr: fmt.Sprintf("%s:%d", instanceAddr, grpcPort),
ID: instanceId,
HeartbeatPeriod: 15 * time.Second,
HeartbeatTimeout: heartbeatTimeout,
TokensObservePeriod: 0,
NumTokens: numTokens,
}, nil
}
func toRingConfig(cfg *setting.Cfg, KVStore kv.Config) ring.Config {
rc := ring.Config{}
flagext.DefaultValues(&rc)
rc.KVStore = KVStore
rc.HeartbeatTimeout = heartbeatTimeout
rc.ReplicationFactor = 1
return rc
}
func newClientPool(clientCfg grpcclient.Config, log log.Logger, reg prometheus.Registerer) *ringclient.Pool {
poolCfg := ringclient.PoolConfig{
CheckInterval: 10 * time.Second,
HealthCheckEnabled: true,
HealthCheckTimeout: 10 * time.Second,
}
clientsCount := promauto.With(reg).NewGauge(prometheus.GaugeOpts{
Name: "resource_server_clients",
Help: "The current number of resource server clients in the pool.",
})
factoryRequestDuration := promauto.With(reg).NewHistogramVec(prometheus.HistogramOpts{
Name: "resource_server_client_request_duration_seconds",
Help: "Time spent executing requests to resource server.",
Buckets: prometheus.ExponentialBuckets(0.008, 4, 7),
}, []string{"operation", "status_code"})
factory := ringclient.PoolInstFunc(func(inst ring.InstanceDesc) (ringclient.PoolClient, error) {
opts, err := clientCfg.DialOption(grpcclient.Instrument(factoryRequestDuration))
if err != nil {
return nil, err
}
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)
}
// TODO only use this if FlagAppPlatformGrpcClientAuth is not enabled
client := resource.NewLegacyResourceClient(conn)
return &resource.RingClient{
Client: client,
HealthClient: grpc_health_v1.NewHealthClient(conn),
Conn: conn,
}, nil
})
return ringclient.NewPool(ringName, poolCfg, nil, factory, clientsCount, log)
}