[10.4.x]: Live: Add ha_prefix (#93759) (#93927)

---------

Co-authored-by: Todd Treece <360020+toddtreece@users.noreply.github.com>

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.

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.

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
 

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
 }

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)
+		}
+	}
+}

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("")