diff --git a/conf/defaults.ini b/conf/defaults.ini index f4ee62c437f..03486a29d79 100644 --- a/conf/defaults.ini +++ b/conf/defaults.ini @@ -1638,6 +1638,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 eb9532aced9..88050edfddc 100644 --- a/conf/sample.ini +++ b/conf/sample.ini @@ -1500,6 +1500,9 @@ # 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 d81f7a99967..b2921ec6362 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..96e98dbe7d8 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,26 @@ 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 95a28baafa5..463edb22c00 100644 --- a/pkg/setting/setting.go +++ b/pkg/setting/setting.go @@ -428,6 +428,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 @@ -1984,6 +1986,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("")