fix(remediation): retry on errors and throw errors to trigger retries

Signed-off-by: Igor Velichkovich <[email protected]>

Author: Igor Velichkovich

.gitignore

@@ -140,30 +140,7 @@ dist/
 ### JetBrains IDEs (GoLand, PyCharm, IntelliJ) ###
 ### JetBrains IDEs (GoLand, PyCharm, IntelliJ) ###
 # User-specific stuff
-.idea/**/workspace.xml
-.idea/**/tasks.xml
-.idea/**/usage.statistics.xml
-.idea/**/dictionaries
-.idea/**/shelf
-
-# AWS User-specific
-.idea/**/aws.xml
-
-# Generated files
-.idea/**/contentModel.xml
-
-# Sensitive or high-churn files
-.idea/**/dataSources/
-.idea/**/dataSources.ids
-.idea/**/dataSources.local.xml
-.idea/**/sqlDataSources.xml
-.idea/**/dynamic.xml
-.idea/**/uiDesigner.xml
-.idea/**/dbnavigator.xml
-
-# Gradle
-.idea/**/gradle.xml
-.idea/**/libraries
+.idea/
 
 # CMake
 cmake-build-*/

fault-remediation/main.go

@@ -20,6 +20,7 @@ import (
 	"flag"
 	"fmt"
 	"log/slog"
+	"net/http"
 	"os"
 	"os/signal"
 	"strconv"
@@ -100,28 +101,16 @@ func run() error {
 		TomlConfigPath:     tomlConfigPath,
 		DryRun:             dryRun,
 		EnableLogCollector: enableLogCollector,
+		UseCtrlRuntime:     enableControllerRuntime,
 	}
 
-	components, err := initializer.InitializeAll(ctx, params)
-	if err != nil {
-		return fmt.Errorf("initialization failed: %w", err)
-	}
-
-	reconciler := components.FaultRemediationReconciler
-
-	defer func() {
-		if err := reconciler.CloseAll(ctx); err != nil {
-			slog.Error("failed to close datastore components", "error", err)
-		}
-	}()
-
 	if enableControllerRuntime {
-		err = setupCtrlRuntimeManagement(ctx, components)
+		err := setupCtrlRuntimeManagement(ctx, params)
 		if err != nil {
 			return err
 		}
 	} else {
-		err = setupNonCtrlRuntimeManaged(ctx, components)
+		err := setupNonCtrlRuntimeManaged(ctx, params)
 		if err != nil {
 			return err
 		}
@@ -130,9 +119,22 @@ func run() error {
 	return nil
 }
 
-func setupNonCtrlRuntimeManaged(ctx context.Context, components *initializer.Components) error {
+func setupNonCtrlRuntimeManaged(ctx context.Context, params initializer.InitializationParams) error {
 	slog.Info("Running without controller runtime management")
 
+	components, err := initializer.InitializeAll(ctx, params, nil)
+	if err != nil {
+		return fmt.Errorf("initialization failed: %w", err)
+	}
+
+	reconciler := components.FaultRemediationReconciler
+
+	defer func() {
+		if err := reconciler.CloseAll(ctx); err != nil {
+			slog.Error("failed to close datastore components", "error", err)
+		}
+	}()
+
 	metricsAddr = strings.TrimPrefix(metricsAddr, ":")
 
 	portInt, err := strconv.Atoi(metricsAddr)
@@ -165,7 +167,19 @@ func setupNonCtrlRuntimeManaged(ctx context.Context, components *initializer.Com
 
 		for event := range components.FaultRemediationReconciler.Watcher.Events() {
 			slog.Info("Event received", "event", event)
-			_, _ = components.FaultRemediationReconciler.Reconcile(gCtx, &event)
+
+			for i := 1; i <= components.FaultRemediationReconciler.Config.UpdateMaxRetries; i++ {
+				_, err = components.FaultRemediationReconciler.Reconcile(gCtx, &event)
+				if err == nil {
+					break
+				}
+
+				slog.Error("Error processing event", "event", event, "error", err)
+
+				if i < components.FaultRemediationReconciler.Config.UpdateMaxRetries {
+					time.Sleep(components.FaultRemediationReconciler.Config.UpdateRetryDelay)
+				}
+			}
 		}
 
 		return nil
@@ -174,10 +188,16 @@ func setupNonCtrlRuntimeManaged(ctx context.Context, components *initializer.Com
 	return g.Wait()
 }
 
-func setupCtrlRuntimeManagement(ctx context.Context, components *initializer.Components) error {
+func setupCtrlRuntimeManagement(ctx context.Context, params initializer.InitializationParams) error {
 	slog.Info("Running in controller runtime managed mode")
 
-	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
+	cfg := ctrl.GetConfigOrDie()
+	cfg.Wrap(func(rt http.RoundTripper) http.RoundTripper {
+		return auditlogger.NewAuditingRoundTripper(rt)
+	})
+
+	//TODO: setup informers for node and job
+	mgr, err := ctrl.NewManager(cfg, ctrl.Options{
 		Scheme: scheme,
 		Metrics: metricsserver.Options{
 			BindAddress: metricsAddr,
@@ -205,6 +225,19 @@ func setupCtrlRuntimeManagement(ctx context.Context, components *initializer.Com
 		return err
 	}
 
+	components, err := initializer.InitializeAll(ctx, params, mgr.GetClient())
+	if err != nil {
+		return fmt.Errorf("initialization failed: %w", err)
+	}
+
+	reconciler := components.FaultRemediationReconciler
+
+	defer func() {
+		if err := reconciler.CloseAll(ctx); err != nil {
+			slog.Error("failed to close datastore components", "error", err)
+		}
+	}()
+
 	err = components.FaultRemediationReconciler.SetupWithManager(ctx, mgr)
 	if err != nil {
 		return fmt.Errorf("SetupWithManager failed: %w", err)
@@ -235,7 +268,7 @@ func parseFlags() {
 			" (otherwise metrics and health are on same port).",
 	)
 
-	flag.StringVar(&kubeconfigPath, "kubeconfig-path", "", "path to kubeconfig file")
+	flag.StringVar(&kubeconfigPath, "kubeconfig", "", "path to kubeconfig file")
 
 	flag.StringVar(&tomlConfigPath, "config-path", "/etc/config/config.toml",
 		"path where the fault remediation config file is present")

fault-remediation/pkg/annotation/annotation.go

@@ -0,0 +1,169 @@
+package annotation
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log/slog"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+)
+
+// CtrlRuntimeAnnotationManager manages node annotations for tracking remediation state.
+type CtrlRuntimeAnnotationManager struct {
+	client client.Client
+}
+
+// NewCtrlRuntimeAnnotationManager creates a new CtrlRuntimeAnnotationManager.
+func NewCtrlRuntimeAnnotationManager(client client.Client) *CtrlRuntimeAnnotationManager {
+	return &CtrlRuntimeAnnotationManager{
+		client: client,
+	}
+}
+
+// GetRemediationState retrieves the current remediation state from node annotation
+func (m *CtrlRuntimeAnnotationManager) GetRemediationState(
+	ctx context.Context,
+	nodeName string,
+) (*RemediationStateAnnotation, *corev1.Node, error) {
+	node := &corev1.Node{}
+
+	err := m.client.Get(ctx, types.NamespacedName{
+		Name: nodeName,
+	}, node)
+	if err != nil {
+		return nil, node, fmt.Errorf("failed to get node %s: %w", nodeName, err)
+	}
+	// TODO: maybe split this up so it's not returning both node and state
+
+	annotationValue, exists := node.Annotations[AnnotationKey]
+	if !exists {
+		// No annotation means no active remediations
+		return &RemediationStateAnnotation{
+			EquivalenceGroups: make(map[string]EquivalenceGroupState),
+		}, node, nil
+	}
+
+	var state RemediationStateAnnotation
+	if err = json.Unmarshal([]byte(annotationValue), &state); err != nil {
+		slog.Error("Failed to unmarshal annotation", "node", nodeName, "error", err)
+		// Return empty state if unmarshal fails
+		return &RemediationStateAnnotation{
+			EquivalenceGroups: make(map[string]EquivalenceGroupState),
+		}, node, nil
+	}
+
+	return &state, node, nil
+}
+
+// UpdateRemediationState updates the node annotation with new remediation state
+func (m *CtrlRuntimeAnnotationManager) UpdateRemediationState(ctx context.Context, nodeName string,
+	group string, crName string) error {
+	// Get current state
+	state, node, err := m.GetRemediationState(ctx, nodeName)
+	if err != nil {
+		// Log but continue with empty state
+		slog.Warn("Failed to get current remediation state", "node", nodeName, "error", err)
+
+		state = &RemediationStateAnnotation{
+			EquivalenceGroups: make(map[string]EquivalenceGroupState),
+		}
+	}
+
+	// Update state for the group
+	state.EquivalenceGroups[group] = EquivalenceGroupState{
+		MaintenanceCR: crName,
+		CreatedAt:     time.Now().UTC(),
+	}
+
+	// Marshal to JSON
+	stateJSON, err := json.Marshal(state)
+	if err != nil {
+		return fmt.Errorf("failed to marshal remediation state: %w", err)
+	}
+
+	patchBase := node.DeepCopy()
+	if node.Annotations == nil {
+		node.Annotations = map[string]string{}
+	}
+
+	node.Annotations[AnnotationKey] = string(stateJSON)
+
+	if err = m.client.Patch(ctx, node, client.MergeFrom(patchBase)); err != nil {
+		return fmt.Errorf("failed to patch node annotation: %w", err)
+	}
+
+	slog.Info("Updated remediation state annotation for node",
+		"node", nodeName,
+		"group", group,
+		"crName", crName)
+
+	return nil
+}
+
+// ClearRemediationState removes the remediation state annotation from a node
+func (m *CtrlRuntimeAnnotationManager) ClearRemediationState(ctx context.Context, nodeName string) error {
+	node := &corev1.Node{}
+
+	err := m.client.Get(ctx, types.NamespacedName{
+		Name: nodeName,
+	}, node)
+	if err != nil {
+		return fmt.Errorf("failed to get node %s: %w", nodeName, err)
+	}
+
+	if node.Annotations == nil {
+		return nil
+	}
+
+	patchBase := node.DeepCopy()
+	delete(node.Annotations, AnnotationKey)
+
+	if err = m.client.Patch(ctx, node, client.MergeFrom(patchBase)); err != nil {
+		return fmt.Errorf("failed to patch node annotation: %w", err)
+	}
+
+	slog.Info("Cleared remediation state annotation for node", "node", nodeName)
+
+	return nil
+}
+
+// RemoveGroupFromState removes a specific group from the remediation state
+func (m *CtrlRuntimeAnnotationManager) RemoveGroupFromState(ctx context.Context, nodeName string, group string) error {
+	state, node, err := m.GetRemediationState(ctx, nodeName)
+	if err != nil {
+		return fmt.Errorf("failed to get current remediation state: %w", err)
+	}
+
+	// Remove the group
+	delete(state.EquivalenceGroups, group)
+
+	// If no groups remain, clear the entire annotation
+	if len(state.EquivalenceGroups) == 0 {
+		return m.ClearRemediationState(ctx, nodeName)
+	}
+
+	// Marshal to JSON
+	stateJSON, err := json.Marshal(state)
+	if err != nil {
+		return fmt.Errorf("failed to marshal remediation state: %w", err)
+	}
+
+	patchBase := node.DeepCopy()
+	if node.Annotations == nil {
+		node.Annotations = map[string]string{}
+	}
+
+	node.Annotations[AnnotationKey] = string(stateJSON)
+
+	if err = m.client.Patch(ctx, node, client.MergeFrom(patchBase)); err != nil {
+		return fmt.Errorf("failed to patch node annotation: %w", err)
+	}
+
+	slog.Info("Removed group from remediation state for node", "node", nodeName, "group", group)
+
+	return nil
+}

fault-remediation/pkg/annotation/annotation_interface.go

@@ -0,0 +1,32 @@
+package annotation
+
+import (
+	"context"
+	"time"
+
+	corev1 "k8s.io/api/core/v1"
+)
+
+const (
+	// AnnotationKey is the key for the node annotation that tracks remediation state
+	AnnotationKey = "latestFaultRemediationState"
+)
+
+// NodeAnnotationManagerInterface defines the interface for managing node annotations
+type NodeAnnotationManagerInterface interface {
+	GetRemediationState(ctx context.Context, nodeName string) (*RemediationStateAnnotation, *corev1.Node, error)
+	UpdateRemediationState(ctx context.Context, nodeName string, group string, crName string) error
+	ClearRemediationState(ctx context.Context, nodeName string) error
+	RemoveGroupFromState(ctx context.Context, nodeName string, group string) error
+}
+
+// RemediationStateAnnotation represents the structure of the node annotation
+type RemediationStateAnnotation struct {
+	EquivalenceGroups map[string]EquivalenceGroupState `json:"equivalenceGroups"`
+}
+
+// EquivalenceGroupState represents the state of a single equivalence group
+type EquivalenceGroupState struct {
+	MaintenanceCR string    `json:"maintenanceCR"`
+	CreatedAt     time.Time `json:"createdAt"`
+}