feat: add redis agent sidecar for role probe

internal/agent/events/k8s_event.go

@@ -0,0 +1,108 @@
+package events
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"os"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+const (
+	RoleChangedReason    = "RoleChanged"
+	InvolvedObjectPod    = "Pod"
+	EventSourceComponent = "redis-role-detector"
+)
+
+// RoleChangeEventData represents the structured data for role change events
+type RoleChangeEventData struct {
+	PodName      string    `json:"podName"`
+	Namespace    string    `json:"namespace"`
+	PreviousRole string    `json:"previousRole"`
+	CurrentRole  string    `json:"currentRole"`
+	Timestamp    time.Time `json:"timestamp"`
+}
+
+// K8sEventSender sends Kubernetes events for role changes
+type K8sEventSender struct {
+	clientset *kubernetes.Clientset
+	podName   string
+	namespace string
+}
+
+// NewK8sEventSender creates a new Kubernetes event sender
+func NewK8sEventSender() (*K8sEventSender, error) {
+	config, err := rest.InClusterConfig()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get in-cluster config: %w", err)
+	}
+
+	clientset, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create clientset: %w", err)
+	}
+
+	podName := os.Getenv("POD_NAME")
+	namespace := os.Getenv("POD_NAMESPACE")
+
+	if podName == "" || namespace == "" {
+		return nil, fmt.Errorf("POD_NAME and POD_NAMESPACE environment variables must be set")
+	}
+
+	return &K8sEventSender{
+		clientset: clientset,
+		podName:   podName,
+		namespace: namespace,
+	}, nil
+}
+
+// SendRoleChangeEvent sends a Kubernetes event for role change
+func (s *K8sEventSender) SendRoleChangeEvent(ctx context.Context, previousRole, currentRole string) error {
+	// Create structured event data
+	eventData := RoleChangeEventData{
+		PodName:      s.podName,
+		Namespace:    s.namespace,
+		PreviousRole: previousRole,
+		CurrentRole:  currentRole,
+		Timestamp:    time.Now().UTC(),
+	}
+
+	// Marshal to JSON for structured message
+	messageBytes, err := json.Marshal(eventData)
+	if err != nil {
+		return fmt.Errorf("failed to marshal event data: %w", err)
+	}
+
+	event := &corev1.Event{
+		ObjectMeta: metav1.ObjectMeta{
+			GenerateName: "redis-role-change-",
+			Namespace:    s.namespace,
+		},
+		InvolvedObject: corev1.ObjectReference{
+			Kind:      InvolvedObjectPod,
+			Name:      s.podName,
+			Namespace: s.namespace,
+		},
+		Reason:  RoleChangedReason,
+		Message: string(messageBytes),
+		Type:    corev1.EventTypeNormal,
+		Source: corev1.EventSource{
+			Component: EventSourceComponent,
+		},
+		FirstTimestamp: metav1.Now(),
+		LastTimestamp:  metav1.Now(),
+		Count:          1,
+	}
+
+	_, err = s.clientset.CoreV1().Events(s.namespace).Create(ctx, event, metav1.CreateOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to create event: %w", err)
+	}
+
+	return nil
+}

internal/agent/server/probe/role_detector.go

@@ -0,0 +1,100 @@
+package probe
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"time"
+
+	"github.com/OT-CONTAINER-KIT/redis-operator/internal/agent/events"
+	"github.com/redis/go-redis/v9"
+)
+
+// RoleDetector periodically checks the Redis role using INFO replication.
+// It prints a log each time the role changes.
+// All configuration is supplied through constructor parameters so that the caller
+// can decide how to source configuration (for example, via environment variables).
+type RoleDetector struct {
+	client      *redis.Client
+	interval    time.Duration
+	currentRole string
+	eventSender *events.K8sEventSender
+}
+
+// NewRoleDetector creates a detector that connects to the given Redis instance.
+func NewRoleDetector(addr, password string, interval time.Duration) *RoleDetector {
+	opts := &redis.Options{
+		Addr:     addr,
+		Password: password,
+	}
+
+	// Try to create event sender, but don't fail if it's not available
+	eventSender, err := events.NewK8sEventSender()
+	if err != nil {
+		fmt.Printf("warning: failed to create event sender: %v\n", err)
+	}
+
+	return &RoleDetector{
+		client:      redis.NewClient(opts),
+		interval:    interval,
+		eventSender: eventSender,
+	}
+}
+
+// Run starts the detection loop and blocks until the context is cancelled.
+func (d *RoleDetector) Run(ctx context.Context) error {
+	// Perform an initial detection immediately.
+	if err := d.detect(ctx); err != nil {
+		fmt.Printf("role detection failed: %v\n", err)
+	}
+
+	ticker := time.NewTicker(d.interval)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-ticker.C:
+			if err := d.detect(ctx); err != nil {
+				fmt.Printf("role detection failed: %v\n", err)
+			}
+		}
+	}
+}
+
+// detect queries Redis and logs a message if the role has changed.
+func (d *RoleDetector) detect(ctx context.Context) error {
+	res, err := d.client.Info(ctx, "replication").Result()
+	if err != nil {
+		return err
+	}
+	role := parseRole(res)
+	if role != d.currentRole {
+		previousRole := d.currentRole
+		if previousRole == "" {
+			previousRole = "unknown"
+		}
+		fmt.Printf("detected role change: %s -> %s\n", previousRole, role)
+
+		// Send Kubernetes event if event sender is available
+		if d.eventSender != nil {
+			if err := d.eventSender.SendRoleChangeEvent(ctx, previousRole, role); err != nil {
+				fmt.Printf("warning: failed to send role change event: %v\n", err)
+			}
+		}
+
+		d.currentRole = role
+	}
+	return nil
+}
+
+// parseRole extracts the role from the output of `INFO replication`.
+func parseRole(info string) string {
+	for _, line := range strings.Split(info, "\n") {
+		if strings.HasPrefix(line, "role:") {
+			return strings.TrimSpace(strings.TrimPrefix(line, "role:"))
+		}
+	}
+	return "unknown"
+}

internal/agent/server/server.go

@@ -0,0 +1,33 @@
+package server
+
+import (
+	"context"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/OT-CONTAINER-KIT/redis-operator/internal/agent/server/probe"
+)
+
+// Start launches a role detector that periodically checks the Redis role.
+// Configuration is provided via function parameters.
+// The call blocks until the context is canceled (SIGINT/SIGTERM) or the
+// detector exits. Any error returned by detector.Run is propagated to caller.
+func Start(addr, password string, interval time.Duration) error {
+	detector := probe.NewRoleDetector(addr, password, interval)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	go handleSignals(cancel)
+
+	return detector.Run(ctx)
+}
+
+func handleSignals(cancel context.CancelFunc) {
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
+	<-sigCh
+	cancel()
+}

internal/cmd/agent/cmd.go

@@ -2,6 +2,7 @@
 
 import (
 	"github.com/OT-CONTAINER-KIT/redis-operator/internal/cmd/agent/bootstrap"
+	"github.com/OT-CONTAINER-KIT/redis-operator/internal/cmd/agent/server"
 	"github.com/spf13/cobra"
 )
 
@@ -12,5 +13,6 @@
 		Short: "Agent is a tool which run as a init/sidecar container along with redis/sentinel",
 	}
 	agentCmd.AddCommand(bootstrap.CMD())
+	agentCmd.AddCommand(server.CMD())
 	return agentCmd
 }

internal/cmd/agent/server/cmd.go

@@ -0,0 +1,30 @@
+package server
+
+import (
+	"time"
+
+	"github.com/OT-CONTAINER-KIT/redis-operator/internal/agent/server"
+	"github.com/spf13/cobra"
+)
+
+func CMD() *cobra.Command {
+	var (
+		redisAddr      string
+		redisPassword  string
+		detectInterval time.Duration
+	)
+
+	cmd := &cobra.Command{
+		Use:   "server",
+		Short: "Start the Redis role detector agent",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			return server.Start(redisAddr, redisPassword, detectInterval)
+		},
+	}
+
+	cmd.Flags().StringVar(&redisAddr, "redis-addr", "127.0.0.1:6379", "Redis address in host:port format")
+	cmd.Flags().StringVar(&redisPassword, "redis-password", "", "Redis password for authentication (optional)")
+	cmd.Flags().DurationVar(&detectInterval, "detect-interval", 10*time.Second, "Role detection interval")
+
+	return cmd
+}

internal/cmd/manager/cmd.go

@@ -25,6 +25,7 @@
 	rsvb2 "github.com/OT-CONTAINER-KIT/redis-operator/api/redissentinel/v1beta2"
 	"github.com/OT-CONTAINER-KIT/redis-operator/internal/controller/common/redis"
 	"github.com/OT-CONTAINER-KIT/redis-operator/internal/controller/common/scheme"
+	eventcontroller "github.com/OT-CONTAINER-KIT/redis-operator/internal/controller/event"
 	rediscontroller "github.com/OT-CONTAINER-KIT/redis-operator/internal/controller/redis"
 	redisclustercontroller "github.com/OT-CONTAINER-KIT/redis-operator/internal/controller/rediscluster"
 	redisreplicationcontroller "github.com/OT-CONTAINER-KIT/redis-operator/internal/controller/redisreplication"
@@ -242,6 +243,12 @@
 		setupLog.Error(err, "unable to create controller", "controller", "RedisSentinel")
 		return err
 	}
+	if err := (&eventcontroller.EventReconciler{
+		Client: mgr.GetClient(),
+	}).SetupWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "Event")
+		return err
+	}
 
 	return nil
 }