Merge pull request #297 from hyperspike/operational_improvements

Operational improvements

Author: dmolik

Dockerfile.valkey

@@ -1,6 +1,6 @@
 FROM alpine:3.22.1 AS builder
 
-ARG VALKEY_VERSION=8.0.2
+ARG VALKEY_VERSION=8.1.3
 
 WORKDIR /home/valkey
 

cfg/config.go

@@ -4,7 +4,7 @@ var (
 	// Default Settings
 	DefaultSidecarImage string
 	DefaultValkeyImage  string
-	DefaultNodes        int = 3
+	DefaultNodes        int32 = 3
 )
 
 type Config struct {
@@ -13,7 +13,7 @@ type Config struct {
 	// The default clusterwide valkey image to use
 	ValkeyImage string `json:"valkeyImage"`
 	// The default number of nodes to use
-	Nodes int `json:"nodes"`
+	Nodes int32 `json:"nodes"`
 }
 
 func Defaults() *Config {

cmd/manager/main.go

@@ -191,7 +191,8 @@ func main() {
 			config.ValkeyImage = v
 		}
 		if k == "nodes" && v != "" {
-			config.Nodes, _ = strconv.Atoi(v)
+			n, _ := strconv.ParseInt(v, 10, 32)
+			config.Nodes = int32(n)
 		}
 	}
 

config/rbac/role.yaml

@@ -47,6 +47,14 @@ rules:
   - patch
   - update
   - watch
+- apiGroups:
+  - storage.k8s.io
+  resources:
+  - storageclasses
+  verbs:
+  - get
+  - list
+  - watch
 - apiGroups:
   - cert-manager.io
   resources:

internal/controller/cluster.go

@@ -0,0 +1,70 @@
+package controller
+
+import (
+	"fmt"
+	"regexp"
+	"slices"
+
+	valkeyClient "github.com/valkey-io/valkey-go"
+)
+
+// valkeyCluster represents a Valkey cluster. It contains a list of shards, each with its own nodes.
+type valkeyCluster struct {
+	shards []*valkeyShard
+}
+
+// valkeyShard represents a shard in the Valkey cluster. It contains slot information and a list of nodes.
+type valkeyShard struct {
+	id      int
+	slotMin int
+	slotMax int
+	nodes   []*valkeyNode
+}
+
+// valkeyNode represents a node in the Valkey cluster.
+type valkeyNode struct {
+	// id is the node id in the Valkey cluster
+	id string
+	// name is the pod name in Kubernetes
+	name string
+	// ip is the pod ip in Kubernetes
+	ip string
+	// port is the port of the Valkey service
+	port int
+	// flags are the Valkey flags for this pod
+	flags []string
+	// primary is the id of the primary node when this node is a replica
+	primary string
+	// connected is true when the pod is reachable from the operator
+	connected bool
+	// shard is the id of the shard this node belongs to
+	shard int
+	// client is the Valkey client for this node, if connected.
+	client valkeyClient.Client
+}
+
+// isPrimary checks if this node is a primary node for the shard in the Valkey cluster.
+func (vn *valkeyNode) isPrimary() bool {
+	return slices.Contains(vn.flags, "master")
+}
+
+// stsPodIndex extracts the pod index number from the pod name. The pod name is expected to be in
+// the format <name>-<number>. The pod index is the pod number from a StatefulSet.
+func stsPodIndex(podName string) (int, error) {
+	pattern := `.*-(\d+)$`
+	re := regexp.MustCompile(pattern)
+
+	matches := re.FindStringSubmatch(podName)
+	if len(matches) < 2 {
+		return 0, fmt.Errorf("no number found in pod name: %s", podName)
+	}
+
+	// Convert the captured group to an integer
+	var number int
+	_, err := fmt.Sscanf(matches[1], "%d", &number)
+	if err != nil {
+		return 0, fmt.Errorf("failed to parse number: %w", err)
+	}
+
+	return number, nil
+}