mirror of
https://github.com/hanzokv/go.git
synced 2026-09-24 00:31:51 +00:00
zeekayandHanzo Dev
ee3ef1a41a
The tree carried two compose files. Docker Compose v2 resolves by a fixed
precedence -- compose.yaml, compose.yml, docker-compose.yaml,
docker-compose.yml -- so compose.yml won every time and docker-compose.yml,
the one upstream merges actually keep current, was never read:
warning Found multiple config files with supported names:
.../compose.yml, .../docker-compose.yml
warning Using /Users/z/work/hanzo/kv-go/compose.yml
compose.yml was a 106-line snapshot taken before several upstream updates.
It pinned CLIENT_LIBS_TEST_IMAGE at 8.4.0 while the Makefile asks for 8.8.0,
and it was missing three services and a whole profile. Two things were
therefore broken and quiet about it:
* `make docker.e2e.start` runs `docker compose --profile e2e up`. There is
no e2e profile in the stale file, so it started nothing;
.github/workflows/test-e2e.yml then sat waiting for cae-resp-proxy on
:18100 until the timeout.
* `make docker.start` (--profile all) never started osscluster-tls, so
tls_cluster_test.go -- whose own comment says the ports come from the
compose file -- had no server on 5430.
Deleting the stale copy and giving the live one the canonical name (this
workspace uses compose.yml, never docker-compose.yml) leaves exactly one
compose file. `docker compose --profile e2e config --services` now returns
redis, cae-resp-proxy and proxy-fault-injector; --profile all also brings up
osscluster-tls. The three comments that named the old path now name the new
one. go build ./... clean.
Co-authored-by: Hanzo Dev <[email protected]>
283 lines
8.0 KiB
Go
283 lines
8.0 KiB
Go
package redis_test
|
|
|
|
import (
|
|
"context"
|
|
"crypto/tls"
|
|
"fmt"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
. "github.com/bsm/ginkgo/v2"
|
|
. "github.com/bsm/gomega"
|
|
"github.com/hanzoai/kv-go/v9"
|
|
)
|
|
|
|
var _ = Describe("TLS Cluster", Label("NonRedisEnterprise"), func() {
|
|
var tlsConfig *tls.Config
|
|
var clusterAddrs []string
|
|
|
|
BeforeEach(func() {
|
|
var err error
|
|
certDir := "dockers/osscluster-tls/tls"
|
|
tlsConfig, err = loadClusterTLSConfig(certDir)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
|
|
// TLS cluster ports from compose.yml
|
|
// TLS_PORT=5430 to avoid conflict with sentinel-cluster (which uses 4430-4432)
|
|
clusterAddrs = []string{
|
|
"localhost:5430",
|
|
"localhost:5431",
|
|
"localhost:5432",
|
|
"localhost:5433",
|
|
"localhost:5434",
|
|
"localhost:5435",
|
|
}
|
|
})
|
|
|
|
Describe("Cluster Client with TLS", func() {
|
|
It("should connect to TLS cluster", func() {
|
|
ctx := context.Background()
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: clusterAddrs,
|
|
TLSConfig: tlsConfig,
|
|
})
|
|
defer client.Close()
|
|
|
|
// Wait for cluster to be ready
|
|
Eventually(func() error {
|
|
return client.Ping(ctx).Err()
|
|
}, 30*time.Second, 1*time.Second).Should(Succeed())
|
|
|
|
val, err := client.Ping(ctx).Result()
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(val).To(Equal("PONG"))
|
|
})
|
|
|
|
It("should perform basic operations over TLS cluster", func() {
|
|
ctx := context.Background()
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: clusterAddrs,
|
|
TLSConfig: tlsConfig,
|
|
})
|
|
defer client.Close()
|
|
|
|
// Wait for cluster to be ready
|
|
Eventually(func() error {
|
|
return client.Ping(ctx).Err()
|
|
}, 30*time.Second, 1*time.Second).Should(Succeed())
|
|
|
|
// SET
|
|
err := client.Set(ctx, "tls_cluster_key", "tls_cluster_value", 0).Err()
|
|
Expect(err).NotTo(HaveOccurred())
|
|
|
|
// GET
|
|
val, err := client.Get(ctx, "tls_cluster_key").Result()
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(val).To(Equal("tls_cluster_value"))
|
|
|
|
// DEL
|
|
deleted, err := client.Del(ctx, "tls_cluster_key").Result()
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(deleted).To(Equal(int64(1)))
|
|
})
|
|
|
|
It("should support pipelining over TLS cluster", func() {
|
|
ctx := context.Background()
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: clusterAddrs,
|
|
TLSConfig: tlsConfig,
|
|
})
|
|
defer client.Close()
|
|
|
|
// Wait for cluster to be ready
|
|
Eventually(func() error {
|
|
return client.Ping(ctx).Err()
|
|
}, 30*time.Second, 1*time.Second).Should(Succeed())
|
|
|
|
// Use hash tags to ensure keys go to same slot
|
|
pipe := client.Pipeline()
|
|
setCmd := pipe.Set(ctx, "{tls}pipe_key1", "value1", 0)
|
|
getCmd := pipe.Get(ctx, "{tls}pipe_key1")
|
|
delCmd := pipe.Del(ctx, "{tls}pipe_key1")
|
|
|
|
_, err := pipe.Exec(ctx)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
|
|
Expect(setCmd.Err()).NotTo(HaveOccurred())
|
|
Expect(getCmd.Val()).To(Equal("value1"))
|
|
Expect(delCmd.Val()).To(Equal(int64(1)))
|
|
})
|
|
|
|
It("should support cluster commands over TLS", func() {
|
|
ctx := context.Background()
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: clusterAddrs,
|
|
TLSConfig: tlsConfig,
|
|
})
|
|
defer client.Close()
|
|
|
|
// Wait for cluster to be ready
|
|
Eventually(func() error {
|
|
return client.Ping(ctx).Err()
|
|
}, 30*time.Second, 1*time.Second).Should(Succeed())
|
|
|
|
// CLUSTER INFO
|
|
info, err := client.ClusterInfo(ctx).Result()
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(info).To(ContainSubstring("cluster_state:ok"))
|
|
|
|
// CLUSTER NODES
|
|
nodes, err := client.ClusterNodes(ctx).Result()
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(nodes).NotTo(BeEmpty())
|
|
|
|
// CLUSTER SLOTS
|
|
slots, err := client.ClusterSlots(ctx).Result()
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(len(slots)).To(BeNumerically(">", 0))
|
|
})
|
|
|
|
It("should support pub/sub over TLS cluster", func() {
|
|
ctx := context.Background()
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: clusterAddrs,
|
|
TLSConfig: tlsConfig,
|
|
})
|
|
defer client.Close()
|
|
|
|
// Wait for cluster to be ready
|
|
Eventually(func() error {
|
|
return client.Ping(ctx).Err()
|
|
}, 30*time.Second, 1*time.Second).Should(Succeed())
|
|
|
|
pubsub := client.Subscribe(ctx, "tls_cluster_channel")
|
|
defer pubsub.Close()
|
|
|
|
// Wait for subscription confirmation
|
|
_, err := pubsub.Receive(ctx)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
|
|
// Publish a message
|
|
err = client.Publish(ctx, "tls_cluster_channel", "tls_cluster_message").Err()
|
|
Expect(err).NotTo(HaveOccurred())
|
|
|
|
// Receive the message
|
|
msg, err := pubsub.ReceiveMessage(ctx)
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(msg.Channel).To(Equal("tls_cluster_channel"))
|
|
Expect(msg.Payload).To(Equal("tls_cluster_message"))
|
|
})
|
|
|
|
It("should handle cluster redirects over TLS", func() {
|
|
ctx := context.Background()
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: clusterAddrs,
|
|
TLSConfig: tlsConfig,
|
|
})
|
|
defer client.Close()
|
|
|
|
// Wait for cluster to be ready
|
|
Eventually(func() error {
|
|
return client.Ping(ctx).Err()
|
|
}, 30*time.Second, 1*time.Second).Should(Succeed())
|
|
|
|
// Set multiple keys that will be distributed across cluster
|
|
for i := 0; i < 100; i++ {
|
|
key := fmt.Sprintf("tls_redirect_key_%d", i)
|
|
err := client.Set(ctx, key, fmt.Sprintf("value_%d", i), 0).Err()
|
|
Expect(err).NotTo(HaveOccurred())
|
|
}
|
|
|
|
// Verify all keys can be retrieved
|
|
for i := 0; i < 100; i++ {
|
|
key := fmt.Sprintf("tls_redirect_key_%d", i)
|
|
val, err := client.Get(ctx, key).Result()
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(val).To(Equal(fmt.Sprintf("value_%d", i)))
|
|
}
|
|
|
|
// Cleanup
|
|
for i := 0; i < 100; i++ {
|
|
key := fmt.Sprintf("tls_redirect_key_%d", i)
|
|
client.Del(ctx, key)
|
|
}
|
|
})
|
|
|
|
It("should connect with InsecureSkipVerify to TLS cluster", func() {
|
|
ctx := context.Background()
|
|
insecureTLSConfig := &tls.Config{
|
|
InsecureSkipVerify: true,
|
|
}
|
|
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: clusterAddrs,
|
|
TLSConfig: insecureTLSConfig,
|
|
})
|
|
defer client.Close()
|
|
|
|
// Wait for cluster to be ready
|
|
Eventually(func() error {
|
|
return client.Ping(ctx).Err()
|
|
}, 30*time.Second, 1*time.Second).Should(Succeed())
|
|
|
|
val, err := client.Ping(ctx).Result()
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(val).To(Equal("PONG"))
|
|
})
|
|
|
|
It("should support ForEachShard over TLS cluster", func() {
|
|
ctx := context.Background()
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: clusterAddrs,
|
|
TLSConfig: tlsConfig,
|
|
})
|
|
defer client.Close()
|
|
|
|
// Wait for cluster to be ready
|
|
Eventually(func() error {
|
|
return client.Ping(ctx).Err()
|
|
}, 30*time.Second, 1*time.Second).Should(Succeed())
|
|
|
|
var shardCount int64
|
|
err := client.ForEachShard(ctx, func(ctx context.Context, shard *redis.Client) error {
|
|
atomic.AddInt64(&shardCount, 1)
|
|
return shard.Ping(ctx).Err()
|
|
})
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(atomic.LoadInt64(&shardCount)).To(BeNumerically(">", 0))
|
|
})
|
|
|
|
It("should support ForEachMaster over TLS cluster", func() {
|
|
ctx := context.Background()
|
|
client := redis.NewClusterClient(&redis.ClusterOptions{
|
|
Addrs: clusterAddrs,
|
|
TLSConfig: tlsConfig,
|
|
})
|
|
defer client.Close()
|
|
|
|
// Wait for cluster to be ready
|
|
Eventually(func() error {
|
|
return client.Ping(ctx).Err()
|
|
}, 30*time.Second, 1*time.Second).Should(Succeed())
|
|
|
|
var masterCount int64
|
|
err := client.ForEachMaster(ctx, func(ctx context.Context, master *redis.Client) error {
|
|
atomic.AddInt64(&masterCount, 1)
|
|
return master.Ping(ctx).Err()
|
|
})
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(atomic.LoadInt64(&masterCount)).To(BeNumerically(">", 0))
|
|
})
|
|
})
|
|
|
|
Describe("TLS Cluster URL Parsing", func() {
|
|
It("should parse rediss:// URL for cluster", func() {
|
|
opt, err := redis.ParseClusterURL("rediss://localhost:16800?addr=localhost:16801&addr=localhost:16802")
|
|
Expect(err).NotTo(HaveOccurred())
|
|
Expect(opt.Addrs).To(ContainElements("localhost:16800", "localhost:16801", "localhost:16802"))
|
|
Expect(opt.TLSConfig).NotTo(BeNil())
|
|
Expect(opt.TLSConfig.ServerName).To(Equal("localhost"))
|
|
})
|
|
})
|
|
})
|