[release-11.6.7] Chore: Update Redis library to v9 (#112370)

pkg/services/live/live.go

@@ -14,11 +14,11 @@ import (
 	"time"
 
 	"github.com/centrifugal/centrifuge"
-	"github.com/go-redis/redis/v8"
 	"github.com/gobwas/glob"
 	"github.com/grafana/grafana-plugin-sdk-go/backend"
 	"github.com/grafana/grafana-plugin-sdk-go/live"
 	jsoniter "github.com/json-iterator/go"
+	"github.com/redis/go-redis/v9"
 	"golang.org/x/sync/errgroup"
 
 	"github.com/grafana/grafana/pkg/api/dtos"

pkg/services/live/managedstream/cache_redis.go

@@ -7,8 +7,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/go-redis/redis/v8"
 	"github.com/grafana/grafana-plugin-sdk-go/data"
+	"github.com/redis/go-redis/v9"
 
 	"github.com/grafana/grafana/pkg/services/live/orgchannel"
 )
@@ -73,40 +73,34 @@ func (c *RedisFrameCache) Update(ctx context.Context, orgID int64, channel strin
 
 	key := c.getCacheKey(orgchannel.PrependOrgID(orgID, channel))
 
-	pipe := c.redisClient.TxPipeline()
-	defer func() { _ = pipe.Close() }()
-
-	pipe.HGetAll(ctx, key)
-	pipe.HMSet(ctx, key, map[string]string{
-		"schema": stringSchema,
-		"frame":  string(jsonFrame.Bytes(data.IncludeAll)),
+	var mapReply *redis.MapStringStringCmd
+	replies, err := c.redisClient.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
+		mapReply = pipe.HGetAll(ctx, key)
+		pipe.HMSet(ctx, key, map[string]string{
+			"schema": stringSchema,
+			"frame":  string(jsonFrame.Bytes(data.IncludeAll)),
+		})
+		pipe.Expire(ctx, key, frameCacheTTL)
+		return nil
 	})
-	pipe.Expire(ctx, key, frameCacheTTL)
-
-	replies, err := pipe.Exec(ctx)
 	if err != nil {
 		return false, err
 	}
 	if len(replies) == 0 {
 		return false, errors.New("no replies in response")
 	}
-	reply := replies[0]
-
-	if reply.Err() != nil {
+	if mapReply.Err() != nil {
 		return false, err
 	}
 
-	if mapReply, ok := reply.(*redis.StringStringMapCmd); ok {
-		result, err := mapReply.Result()
-		if err != nil {
-			return false, err
-		}
-		if len(result) == 0 {
-			return true, nil
-		}
-		return result["schema"] != stringSchema, nil
+	result, err := mapReply.Result()
+	if err != nil {
+		return false, err
 	}
-	return true, nil
+	if len(result) == 0 {
+		return true, nil
+	}
+	return result["schema"] != stringSchema, nil
 }
 
 func (c *RedisFrameCache) getCacheKey(channelID string) string {

pkg/services/live/managedstream/cache_redis_test.go

@@ -1,12 +1,13 @@
 package managedstream
 
 import (
+	"context"
 	"os"
 	"strings"
 	"testing"
 
-	"github.com/go-redis/redis/v8"
 	"github.com/google/uuid"
+	"github.com/redis/go-redis/v9"
 	"github.com/stretchr/testify/require"
 )
 
@@ -40,7 +41,7 @@ func TestIntegrationRedisCacheStorage(t *testing.T) {
 	require.NotNil(t, c)
 	testFrameCache(t, c)
 
-	keys, err := redisClient.Keys(redisClient.Context(), "*").Result()
+	keys, err := redisClient.Keys(t.Context(), "*").Result()
 	if err != nil {
 		require.NoError(t, err)
 	}
@@ -54,13 +55,18 @@ func TestIntegrationRedisCacheStorage(t *testing.T) {
 
 func redisCleanup(t *testing.T, redisClient *redis.Client, prefix string) func() {
 	return func() {
-		keys, err := redisClient.Keys(redisClient.Context(), prefix+"*").Result()
+		ctx := t.Context()
+		ctx = context.WithoutCancel(ctx)
+		ctx, cancel := context.WithCancel(ctx)
+		defer cancel()
+
+		keys, err := redisClient.Keys(ctx, prefix+"*").Result()
 		if err != nil {
 			require.NoError(t, err)
 		}
 
 		for _, key := range keys {
-			_, err := redisClient.Del(redisClient.Context(), key).Result()
+			_, err := redisClient.Del(ctx, key).Result()
 			require.NoError(t, err)
 		}
 	}