From 12dd28d04d4100a70e947f75d64168313f7d3d57 Mon Sep 17 00:00:00 2001 From: Stephanie Hingtgen Date: Fri, 27 Sep 2024 13:20:43 -0600 Subject: [PATCH] [v11.0.x]: Live: Add ha_prefix (#93759) (#93926) --------- Co-authored-by: Todd Treece <360020+toddtreece@users.noreply.github.com> --- conf/defaults.ini | 3 +++ conf/sample.ini | 3 +++ pkg/services/live/live.go | 13 +++++++++--- .../live/managedstream/cache_redis.go | 12 ++++++----- .../live/managedstream/cache_redis_test.go | 21 ++++++++++++++++++- pkg/setting/setting.go | 3 +++ 6 files changed, 46 insertions(+), 9 deletions(-) diff --git a/conf/defaults.ini b/conf/defaults.ini index 15df83bbc71..bd39bafef6f 100644 --- a/conf/defaults.ini +++ b/conf/defaults.ini @@ -1628,6 +1628,9 @@ ha_engine_address = "127.0.0.1:6379" # ha_engine_password allows setting an optional password to authenticate with the engine ha_engine_password = "" +# ha_prefix is a prefix for keys in the HA engine. It's used to separate keys for different Grafana instances. +ha_prefix = + #################################### Grafana Image Renderer Plugin ########################## [plugin.grafana-image-renderer] # Instruct headless browser instance to use a default timezone when not provided by Grafana, e.g. when rendering panel image of alert. diff --git a/conf/sample.ini b/conf/sample.ini index 7ae9dcc74b9..ea5ea404ca5 100644 --- a/conf/sample.ini +++ b/conf/sample.ini @@ -1480,6 +1480,9 @@ max_annotations_to_keep = # ha_engine_password allows setting an optional password to authenticate with the engine ;ha_engine_password = "" +# ha_prefix is a prefix for keys in the HA engine. It's used to separate keys for different Grafana instances. +;ha_prefix = + #################################### Grafana Image Renderer Plugin ########################## [plugin.grafana-image-renderer] # Instruct headless browser instance to use a default timezone when not provided by Grafana, e.g. when rendering panel image of alert. diff --git a/pkg/services/live/live.go b/pkg/services/live/live.go index f580caf6a3b..dba2e449481 100644 --- a/pkg/services/live/live.go +++ b/pkg/services/live/live.go @@ -98,6 +98,11 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r }, usageStatsService: usageStatsService, orgService: orgService, + keyPrefix: "gf_live", + } + + if cfg.LiveHAPrefix != "" { + g.keyPrefix = cfg.LiveHAPrefix + ".gf_live" } logger.Debug("GrafanaLive initialization", "ha", g.IsHA()) @@ -153,7 +158,7 @@ func ProvideService(plugCtxProvider *plugincontext.Provider, cfg *setting.Cfg, r managedStreamRunner = managedstream.NewRunner( g.Publish, channelLocalPublisher, - managedstream.NewRedisFrameCache(redisClient), + managedstream.NewRedisFrameCache(redisClient, g.keyPrefix), ) } else { managedStreamRunner = managedstream.NewRunner( @@ -343,7 +348,7 @@ func setupRedisLiveEngine(g *GrafanaLive, node *centrifuge.Node) error { } broker, err := centrifuge.NewRedisBroker(node, centrifuge.RedisBrokerConfig{ - Prefix: "gf_live", + Prefix: g.keyPrefix, Shards: redisShards, }) if err != nil { @@ -352,7 +357,7 @@ func setupRedisLiveEngine(g *GrafanaLive, node *centrifuge.Node) error { node.SetBroker(broker) presenceManager, err := centrifuge.NewRedisPresenceManager(node, centrifuge.RedisPresenceManagerConfig{ - Prefix: "gf_live", + Prefix: g.keyPrefix, Shards: redisShards, }) if err != nil { @@ -381,6 +386,8 @@ type GrafanaLive struct { queryDataService query.Service orgService org.Service + keyPrefix string + node *centrifuge.Node surveyCaller *survey.Caller diff --git a/pkg/services/live/managedstream/cache_redis.go b/pkg/services/live/managedstream/cache_redis.go index b547a11cd20..34b16991671 100644 --- a/pkg/services/live/managedstream/cache_redis.go +++ b/pkg/services/live/managedstream/cache_redis.go @@ -18,11 +18,13 @@ type RedisFrameCache struct { mu sync.RWMutex redisClient *redis.Client frames map[int64]map[string]data.FrameJSONCache + keyPrefix string } // NewRedisFrameCache ... -func NewRedisFrameCache(redisClient *redis.Client) *RedisFrameCache { +func NewRedisFrameCache(redisClient *redis.Client, keyPrefix string) *RedisFrameCache { return &RedisFrameCache{ + keyPrefix: keyPrefix, frames: map[int64]map[string]data.FrameJSONCache{}, redisClient: redisClient, } @@ -43,7 +45,7 @@ func (c *RedisFrameCache) GetActiveChannels(orgID int64) (map[string]json.RawMes } func (c *RedisFrameCache) GetFrame(ctx context.Context, orgID int64, channel string) (json.RawMessage, bool, error) { - key := getCacheKey(orgchannel.PrependOrgID(orgID, channel)) + key := c.getCacheKey(orgchannel.PrependOrgID(orgID, channel)) cmd := c.redisClient.HGetAll(ctx, key) result, err := cmd.Result() if err != nil { @@ -69,7 +71,7 @@ func (c *RedisFrameCache) Update(ctx context.Context, orgID int64, channel strin stringSchema := string(jsonFrame.Bytes(data.IncludeSchemaOnly)) - key := getCacheKey(orgchannel.PrependOrgID(orgID, channel)) + key := c.getCacheKey(orgchannel.PrependOrgID(orgID, channel)) pipe := c.redisClient.TxPipeline() defer func() { _ = pipe.Close() }() @@ -107,6 +109,6 @@ func (c *RedisFrameCache) Update(ctx context.Context, orgID int64, channel strin return true, nil } -func getCacheKey(channelID string) string { - return "gf_live.managed_stream." + channelID +func (c *RedisFrameCache) getCacheKey(channelID string) string { + return c.keyPrefix + ".managed_stream." + channelID } diff --git a/pkg/services/live/managedstream/cache_redis_test.go b/pkg/services/live/managedstream/cache_redis_test.go index d7c3f9fddbb..905684cb46c 100644 --- a/pkg/services/live/managedstream/cache_redis_test.go +++ b/pkg/services/live/managedstream/cache_redis_test.go @@ -5,6 +5,7 @@ import ( "testing" "github.com/go-redis/redis/v8" + "github.com/google/uuid" "github.com/stretchr/testify/require" ) @@ -30,7 +31,25 @@ func TestIntegrationRedisCacheStorage(t *testing.T) { Addr: addr, DB: db, }) - c := NewRedisFrameCache(redisClient) + prefix := uuid.New().String() + + t.Cleanup(redisCleanup(t, redisClient, prefix)) + + c := NewRedisFrameCache(redisClient, prefix) require.NotNil(t, c) testFrameCache(t, c) } + +func redisCleanup(t *testing.T, redisClient *redis.Client, prefix string) func() { + return func() { + keys, err := redisClient.Keys(redisClient.Context(), prefix+"*").Result() + if err != nil { + require.NoError(t, err) + } + + for _, key := range keys { + _, err := redisClient.Del(redisClient.Context(), key).Result() + require.NoError(t, err) + } + } +} diff --git a/pkg/setting/setting.go b/pkg/setting/setting.go index a76604b1975..430a1a9ce7b 100644 --- a/pkg/setting/setting.go +++ b/pkg/setting/setting.go @@ -424,6 +424,8 @@ type Cfg struct { // LiveHAEngine is a type of engine to use to achieve HA with Grafana Live. // Zero value means in-memory single node setup. LiveHAEngine string + // LiveHAPRefix is a prefix for HA engine keys. + LiveHAPrefix string // LiveHAEngineAddress is a connection address for Live HA engine. LiveHAEngineAddress string LiveHAEnginePassword string @@ -1972,6 +1974,7 @@ func (cfg *Cfg) readLiveSettings(iniFile *ini.File) error { default: return fmt.Errorf("unsupported live HA engine type: %s", cfg.LiveHAEngine) } + cfg.LiveHAPrefix = section.Key("ha_prefix").MustString("") cfg.LiveHAEngineAddress = section.Key("ha_engine_address").MustString("127.0.0.1:6379") cfg.LiveHAEnginePassword = section.Key("ha_engine_password").MustString("")