update peer timeout logic:

- Update the default PeerTimeout value to a higher, safer value (e.g., 7 seconds).
- added a validating admission webhook that issues a warning if the configured PeerTimeout is less than APIServerTimeout + MinimumBuffer (e.g., 2 seconds).
- if the configured PeerTimeout is less than APIServerTimeout + MinimumBuffer, the operator will internally use APIServerTimeout + MinimumBuffer for the peer check.
- added Unit tests

Signed-off-by: Michael Shitrit <[email protected]>

Author: mshitrit

api/v1alpha1/selfnoderemediationconfig_types.go

@@ -94,7 +94,7 @@ type SelfNodeRemediationConfigSpec struct {
 
 	// Timeout for each peer request.
 	// Valid time units are "ms", "s", "m", "h".
-	// +kubebuilder:default:="5s"
+	// +kubebuilder:default:="7s"
 	// +kubebuilder:validation:Pattern="^([0-9]+(\\.[0-9]+)?(ns|us|µs|ms|s|m|h))+$"
 	// +kubebuilder:validation:Type:=string
 	// +optional

api/v1alpha1/selfnoderemediationconfig_webhook.go

@@ -49,6 +49,10 @@ const (
 	minDurPeerRequestTimeout   = 10 * time.Millisecond
 	minDurApiCheckInterval     = 1 * time.Second
 	minDurPeerUpdateInterval   = 10 * time.Second
+
+	// MinimumBuffer is the minimum buffer time between APIServerTimeout and PeerRequestTimeout
+	// It is required to make sure there is enough time for network communication between the peers in case the API Server is out
+	MinimumBuffer = 2 * time.Second
 )
 
 type field struct {
@@ -74,7 +78,9 @@ var _ webhook.Validator = &SelfNodeRemediationConfig{}
 func (r *SelfNodeRemediationConfig) ValidateCreate() (warning admission.Warnings, err error) {
 	selfNodeRemediationConfigLog.Info("validate create", "name", r.Name)
 
-	return admission.Warnings{}, errors.NewAggregate([]error{
+	warnings := r.validatePeerTimeoutSafety()
+
+	return warnings, errors.NewAggregate([]error{
 		r.validateTimes(),
 		r.validateCustomTolerations(),
 		r.validateSingleton(),
@@ -86,7 +92,9 @@ func (r *SelfNodeRemediationConfig) ValidateCreate() (warning admission.Warnings
 func (r *SelfNodeRemediationConfig) ValidateUpdate(_ runtime.Object) (warning admission.Warnings, err error) {
 	selfNodeRemediationConfigLog.Info("validate update", "name", r.Name)
 
-	return admission.Warnings{}, errors.NewAggregate([]error{
+	warnings := r.validatePeerTimeoutSafety()
+
+	return warnings, errors.NewAggregate([]error{
 		r.validateTimes(),
 		r.validateCustomTolerations(),
 	})
@@ -184,6 +192,43 @@ func validateToleration(toleration v1.Toleration) error {
 	return nil
 }
 
+// validatePeerTimeoutSafety checks if PeerRequestTimeout is safe relative to ApiServerTimeout
+// and returns warnings if the configuration might be unsafe
+func (r *SelfNodeRemediationConfig) validatePeerTimeoutSafety() admission.Warnings {
+	var warnings admission.Warnings
+
+	spec := r.Spec
+	if spec.PeerRequestTimeout == nil || spec.ApiServerTimeout == nil {
+		// Use defaults if not specified
+		return warnings
+	}
+
+	peerRequestTimeoutDuration := spec.PeerRequestTimeout.Duration
+	apiServerTimeoutDuration := spec.ApiServerTimeout.Duration
+	minimumSafePeerTimeout := apiServerTimeoutDuration + MinimumBuffer
+
+	if peerRequestTimeoutDuration < minimumSafePeerTimeout {
+		warningMsg := fmt.Sprintf(
+			"PeerRequestTimeout (%s) is less than ApiServerTimeout + MinimumBuffer (%s + %s = %s). "+
+				"This configuration may lead to race conditions where peer health checks time out "+
+				"before API server checks complete, potentially causing premature remediation. "+
+				"Consider increasing PeerRequestTimeout to at least %s for safer operation.",
+			peerRequestTimeoutDuration,
+			apiServerTimeoutDuration,
+			MinimumBuffer,
+			minimumSafePeerTimeout,
+			minimumSafePeerTimeout,
+		)
+		warnings = append(warnings, warningMsg)
+		selfNodeRemediationConfigLog.Info("PeerRequestTimeout safety warning",
+			"peerRequestTimeout", peerRequestTimeoutDuration,
+			"apiServerTimeout", apiServerTimeoutDuration,
+			"minimumSafeTimeout", minimumSafePeerTimeout)
+	}
+
+	return warnings
+}
+
 func (r *SelfNodeRemediationConfig) validateSingleton() error {
 	if r.Name != ConfigCRName {
 		return fmt.Errorf("to enforce only one SelfNodeRemediationConfig in the cluster, a name other than %s is not allowed", ConfigCRName)

api/v1alpha1/selfnoderemediationconfig_webhook_test.go

@@ -18,7 +18,7 @@ const (
 	peerApiServerTimeoutDefault = 5 * time.Second
 	apiServerTimeoutDefault     = 5 * time.Second
 	peerDialTimeoutDefault      = 5 * time.Second
-	peerRequestTimeoutDefault   = 5 * time.Second
+	peerRequestTimeoutDefault   = 7 * time.Second
 	apiCheckIntervalDefault     = 15 * time.Second
 	peerUpdateIntervalDefault   = 15 * time.Minute
 )
@@ -139,6 +139,42 @@ var _ = Describe("SelfNodeRemediationConfig Validation", func() {
 		})
 	})
 
+	Describe("PeerRequestTimeout Safety Validation", func() {
+		var snrc *SelfNodeRemediationConfig
+		BeforeEach(func() {
+			snrc = createTestSelfNodeRemediationConfigCR()
+			snrc.Name = ConfigCRName
+			_ = os.Setenv("DEPLOYMENT_NAMESPACE", snrc.Namespace)
+		})
+		It("should produce warning when PeerRequestTimeout is too low", func() {
+			// Set ApiServerTimeout to 5s and PeerRequestTimeout to 6s (less than 5s + 2s buffer)
+			snrc.Spec.ApiServerTimeout = &metav1.Duration{Duration: 5 * time.Second}
+			snrc.Spec.PeerRequestTimeout = &metav1.Duration{Duration: 6 * time.Second}
+
+			warnings, err := snrc.ValidateCreate()
+			Expect(err).To(BeNil())
+			Expect(len(warnings)).To(Equal(1))
+			Expect(warnings[0]).To(ContainSubstring("PeerRequestTimeout (6s) is less than ApiServerTimeout + MinimumBuffer"))
+		})
+
+		It("should not produce warning when PeerRequestTimeout is safe", func() {
+			// Set ApiServerTimeout to 5s and PeerRequestTimeout to 8s (greater than 5s + 2s buffer)
+			snrc.Spec.ApiServerTimeout = &metav1.Duration{Duration: 5 * time.Second}
+			snrc.Spec.PeerRequestTimeout = &metav1.Duration{Duration: 8 * time.Second}
+
+			warnings, err := snrc.ValidateCreate()
+			Expect(err).To(BeNil())
+			Expect(len(warnings)).To(Equal(0))
+		})
+
+		It("should not produce warning when using default values", func() {
+			// Use default values: ApiServerTimeout=5s, PeerRequestTimeout=7s (which is safe)
+
+			warnings, err := snrc.ValidateCreate()
+			Expect(err).To(BeNil())
+			Expect(len(warnings)).To(Equal(0))
+		})
+	})
 })
 
 func testSingleInvalidField(validationType validationType) {

bundle/manifests/self-node-remediation.medik8s.io_selfnoderemediationconfigs.yaml

@@ -153,7 +153,7 @@ spec:
                 pattern: ^([0-9]+(\.[0-9]+)?(ns|us|µs|ms|s|m|h))+$
                 type: string
               peerRequestTimeout:
-                default: 5s
+                default: 7s
                 description: |-
                   Timeout for each peer request.
                   Valid time units are "ms", "s", "m", "h".

config/crd/bases/self-node-remediation.medik8s.io_selfnoderemediationconfigs.yaml

@@ -151,7 +151,7 @@ spec:
                 pattern: ^([0-9]+(\.[0-9]+)?(ns|us|µs|ms|s|m|h))+$
                 type: string
               peerRequestTimeout:
-                default: 5s
+                default: 7s
                 description: |-
                   Timeout for each peer request.
                   Valid time units are "ms", "s", "m", "h".

controllers/tests/config/selfnoderemediationconfig_controller_test.go

@@ -233,7 +233,7 @@ var _ = Describe("SNR Config Test", func() {
 			Expect(createdConfig.Spec.MaxApiErrorThreshold).To(Equal(3))
 
 			Expect(createdConfig.Spec.PeerApiServerTimeout.Seconds()).To(BeEquivalentTo(5))
-			Expect(createdConfig.Spec.PeerRequestTimeout.Seconds()).To(BeEquivalentTo(5))
+			Expect(createdConfig.Spec.PeerRequestTimeout.Seconds()).To(BeEquivalentTo(7))
 			Expect(createdConfig.Spec.PeerDialTimeout.Seconds()).To(BeEquivalentTo(5))
 			Expect(createdConfig.Spec.ApiServerTimeout.Seconds()).To(BeEquivalentTo(5))
 			Expect(createdConfig.Spec.ApiCheckInterval.Seconds()).To(BeEquivalentTo(15))

main.go

@@ -344,6 +344,7 @@ func initSelfNodeRemediationAgent(mgr manager.Manager) {
 		PeerRequestTimeout:        peerRequestTimeout,
 		PeerHealthPort:            peerHealthDefaultPort,
 		MaxTimeForNoPeersResponse: reboot.MaxTimeForNoPeersResponse,
+		Recorder:                  mgr.GetEventRecorderFor("ApiConnectivityCheck"),
 	}
 
 	controlPlaneManager := controlplane.NewManager(myNodeName, mgr.GetClient())
@@ -377,7 +378,7 @@ func initSelfNodeRemediationAgent(mgr manager.Manager) {
 
 	setupLog.Info("init grpc server")
 	// TODO make port configurable?
-	server, err := peerhealth.NewServer(mgr.GetClient(), mgr.GetAPIReader(), ctrl.Log.WithName("peerhealth").WithName("server"), peerHealthDefaultPort, certReader)
+	server, err := peerhealth.NewServer(mgr.GetClient(), mgr.GetAPIReader(), ctrl.Log.WithName("peerhealth").WithName("server"), peerHealthDefaultPort, certReader, apiServerTimeout)
 	if err != nil {
 		setupLog.Error(err, "failed to init grpc server")
 		os.Exit(1)

pkg/apicheck/check.go

@@ -8,15 +8,19 @@ import (
 	"time"
 
 	"github.com/go-logr/logr"
+	"github.com/medik8s/common/pkg/events"
 	"google.golang.org/grpc/credentials"
 
 	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/record"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	selfNodeRemediation "github.com/medik8s/self-node-remediation/api"
+	"github.com/medik8s/self-node-remediation/api/v1alpha1"
 	"github.com/medik8s/self-node-remediation/pkg/certificates"
 	"github.com/medik8s/self-node-remediation/pkg/controlplane"
 	"github.com/medik8s/self-node-remediation/pkg/peerhealth"
@@ -25,6 +29,10 @@ import (
 	"github.com/medik8s/self-node-remediation/pkg/utils"
 )
 
+const (
+	eventReasonPeerTimeoutAdjusted = "PeerTimeoutAdjusted"
+)
+
 type ApiConnectivityCheck struct {
 	client.Reader
 	config                 *ApiConnectivityCheckConfig
@@ -51,6 +59,7 @@ type ApiConnectivityCheckConfig struct {
 	PeerHealthPort            int
 	MaxTimeForNoPeersResponse time.Duration
 	MinPeersForRemediation    int
+	Recorder                  record.EventRecorder
 }
 
 func New(config *ApiConnectivityCheckConfig, controlPlaneManager *controlplane.Manager) *ApiConnectivityCheck {
@@ -275,6 +284,25 @@ func (c *ApiConnectivityCheck) getHealthStatusFromPeers(addresses []corev1.PodIP
 	return c.sumPeersResponses(nrAddresses, responsesChan)
 }
 
+// getEffectivePeerRequestTimeout calculates the effective peer request timeout
+// ensuring it's safe relative to the API server timeout by enforcing a minimum buffer
+func (c *ApiConnectivityCheck) getEffectivePeerRequestTimeout() time.Duration {
+	minimumSafeTimeout := c.config.ApiServerTimeout + v1alpha1.MinimumBuffer
+
+	if c.config.PeerRequestTimeout < minimumSafeTimeout {
+		// Log warning about timeout adjustment
+		c.config.Log.Info("PeerRequestTimeout is too low, using adjusted value for safety",
+			"configuredTimeout", c.config.PeerRequestTimeout,
+			"apiServerTimeout", c.config.ApiServerTimeout,
+			"minimumBuffer", v1alpha1.MinimumBuffer,
+			"effectiveTimeout", minimumSafeTimeout)
+		events.WarningEventf(c.config.Recorder, &v1alpha1.SelfNodeRemediationConfig{ObjectMeta: metav1.ObjectMeta{Name: v1alpha1.ConfigCRName}}, eventReasonPeerTimeoutAdjusted, "PeerRequestTimeout (%s) was too low compared to ApiServerTimeout (%s), using safe value (%s) instead", c.config.PeerRequestTimeout, c.config.ApiServerTimeout, minimumSafeTimeout)
+		return minimumSafeTimeout
+	}
+
+	return c.config.PeerRequestTimeout
+}
+
 // getHealthStatusFromPeer issues a GET request to the specified IP and returns the result from the peer into the given channel
 func (c *ApiConnectivityCheck) getHealthStatusFromPeer(endpointIp corev1.PodIP, results chan<- selfNodeRemediation.HealthCheckResponseCode) {
 
@@ -296,7 +324,8 @@ func (c *ApiConnectivityCheck) getHealthStatusFromPeer(endpointIp corev1.PodIP,
 	}
 	defer phClient.Close()
 
-	ctx, cancel := context.WithTimeout(context.Background(), c.config.PeerRequestTimeout)
+	effectiveTimeout := c.getEffectivePeerRequestTimeout()
+	ctx, cancel := context.WithTimeout(context.Background(), effectiveTimeout)
 	defer cancel()
 
 	resp, err := phClient.IsHealthy(ctx, &peerhealth.HealthRequest{

pkg/peerhealth/client_server_test.go

@@ -49,7 +49,7 @@ var _ = Describe("Checking health using grpc client and server", func() {
 		}
 
 		By("Creating server")
-		phServer, err = NewServer(k8sClient, reader, ctrl.Log.WithName("peerhealth test").WithName("phServer"), 9000, certReader)
+		phServer, err = NewServer(k8sClient, reader, ctrl.Log.WithName("peerhealth test").WithName("phServer"), 9000, certReader, 7*time.Second)
 		Expect(err).ToNot(HaveOccurred())
 
 		By("Starting server")

pkg/peerhealth/server.go

@@ -21,10 +21,6 @@ import (
 
 const (
 	connectionTimeout = 5 * time.Second
-	//IMPORTANT! this MUST be less than PeerRequestTimeout in apicheck
-	//The difference between them should allow some time for sending the request over the network
-	//todo enforce this
-	apiServerTimeout = 3 * time.Second
 )
 
 var (
@@ -42,21 +38,23 @@ var (
 
 type Server struct {
 	UnimplementedPeerHealthServer
-	c          client.Client
-	reader     client.Reader
-	log        logr.Logger
-	certReader certificates.CertStorageReader
-	port       int
+	c                client.Client
+	reader           client.Reader
+	log              logr.Logger
+	certReader       certificates.CertStorageReader
+	port             int
+	apiServerTimeout time.Duration
 }
 
 // NewServer returns a new Server
-func NewServer(c client.Client, reader client.Reader, log logr.Logger, port int, certReader certificates.CertStorageReader) (*Server, error) {
+func NewServer(c client.Client, reader client.Reader, log logr.Logger, port int, certReader certificates.CertStorageReader, apiServerTimeout time.Duration) (*Server, error) {
 	return &Server{
-		c:          c,
-		reader:     reader,
-		log:        log,
-		certReader: certReader,
-		port:       port,
+		c:                c,
+		reader:           reader,
+		log:              log,
+		certReader:       certReader,
+		port:             port,
+		apiServerTimeout: apiServerTimeout,
 	}, nil
 }
 
@@ -109,7 +107,7 @@ func (s *Server) IsHealthy(ctx context.Context, request *HealthRequest) (*Health
 		return nil, fmt.Errorf("empty node name in HealthRequest")
 	}
 
-	apiCtx, cancelFunc := context.WithTimeout(ctx, apiServerTimeout)
+	apiCtx, cancelFunc := context.WithTimeout(ctx, s.apiServerTimeout)
 	defer cancelFunc()
 
 	snrs := &v1alpha1.SelfNodeRemediationList{}
@@ -157,7 +155,7 @@ func (s *Server) listWithTimeoutHandling(apiCtx context.Context, snrs *v1alpha1.
 }
 
 func (s *Server) getNode(ctx context.Context, nodeName string) (*corev1.Node, error) {
-	apiCtx, cancelFunc := context.WithTimeout(ctx, apiServerTimeout)
+	apiCtx, cancelFunc := context.WithTimeout(ctx, s.apiServerTimeout)
 	defer cancelFunc()
 
 	node := &corev1.Node{}