Merge pull request #264 from mshitrit/fix-peertimeout-logic

Update peer timeout logic

Author: openshift-merge-bot[bot]

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. "+
+				"Overriding PeerRequestTimeout to %s for safer operation.",
+			peerRequestTimeoutDuration,
+			apiServerTimeoutDuration,
+			MinimumBuffer,
+			minimumSafePeerTimeout,
+			minimumSafePeerTimeout,
+		)
+		warnings = append(warnings, warningMsg)
+		selfNodeRemediationConfigLog.Info("PeerRequestTimeout safety warning, overriding PeerRequestTimeout to minimumSafeTimeout",
+			"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/apicheck/check_test.go

@@ -0,0 +1,83 @@
+package apicheck
+
+import (
+	"testing"
+	"time"
+
+	"github.com/go-logr/logr"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+
+	"k8s.io/client-go/tools/record"
+	ctrl "sigs.k8s.io/controller-runtime"
+
+	"github.com/medik8s/self-node-remediation/api/v1alpha1"
+)
+
+func TestApiCheck(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "ApiCheck Suite")
+}
+
+var _ = Describe("ApiConnectivityCheck", func() {
+	var (
+		apiCheck     *ApiConnectivityCheck
+		config       *ApiConnectivityCheckConfig
+		fakeRecorder *record.FakeRecorder
+		log          logr.Logger
+	)
+
+	BeforeEach(func() {
+		log = ctrl.Log.WithName("test")
+		fakeRecorder = record.NewFakeRecorder(10)
+
+		config = &ApiConnectivityCheckConfig{
+			Log:                log,
+			MyNodeName:         "test-node",
+			ApiServerTimeout:   5 * time.Second,
+			PeerRequestTimeout: 7 * time.Second,
+			Recorder:           fakeRecorder,
+		}
+
+		apiCheck = &ApiConnectivityCheck{
+			config: config,
+		}
+	})
+
+	Describe("getEffectivePeerRequestTimeout", func() {
+		Context("when PeerRequestTimeout is safe", func() {
+			It("should return the configured PeerRequestTimeout", func() {
+				// ApiServerTimeout=5s, PeerRequestTimeout=7s, MinimumBuffer=2s
+				// 7s >= (5s + 2s), so it's safe
+				effectiveTimeout := apiCheck.getEffectivePeerRequestTimeout()
+
+				Expect(effectiveTimeout).To(Equal(7 * time.Second))
+
+				// Should not emit any events
+				Expect(len(fakeRecorder.Events)).To(Equal(0))
+			})
+		})
+
+		Context("when PeerRequestTimeout is unsafe", func() {
+			It("should return adjusted timeout and emit warning event", func() {
+				config.PeerRequestTimeout = 6 * time.Second // Less than 5s + 2s = 7s
+
+				effectiveTimeout := apiCheck.getEffectivePeerRequestTimeout()
+
+				expectedMinimumTimeout := config.ApiServerTimeout + v1alpha1.MinimumBuffer // 7s
+				Expect(effectiveTimeout).To(Equal(expectedMinimumTimeout))
+
+				// Should emit a warning event
+				Expect(len(fakeRecorder.Events)).To(Equal(1))
+				event := <-fakeRecorder.Events
+				Expect(event).To(ContainSubstring("Warning"))
+				Expect(event).To(ContainSubstring("PeerTimeoutAdjusted"))
+				Expect(event).To(ContainSubstring("6s")) // configured timeout
+				Expect(event).To(ContainSubstring("5s")) // API server timeout
+				Expect(event).To(ContainSubstring("7s")) // safe timeout
+			})
+
+		})
+
+	})
+})

pkg/peerhealth/client_server_test.go

@@ -23,8 +23,9 @@ var _ = Describe("Checking health using grpc client and server", func() {
 	var phServer *Server
 	var cancel context.CancelFunc
 	var phClient *Client
+	var apiServerTimeout = 5 * time.Second
 
-	BeforeEach(func() {
+	JustBeforeEach(func() {
 
 		By("creating test node")
 		node := &v1.Node{
@@ -49,7 +50,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, apiServerTimeout)
 		Expect(err).ToNot(HaveOccurred())
 
 		By("Starting server")
@@ -132,15 +133,18 @@ var _ = Describe("Checking health using grpc client and server", func() {
 
 		BeforeEach(func() {
 			reader.delay = &apiCallDelay
+			apiServerTimeout = 3 * time.Second
+
 		})
 
 		AfterEach(func() {
 			reader.delay = nil
+			apiServerTimeout = 5 * time.Second
 		})
 
 		It("should return API error", func() {
 			By("calling isHealthy")
-			// The health server code has a hardcoded timeout of 3s for the API call!
+			// The health server code has a timeout of 3s for the API call!
 			// When we add a delay of > 3s to the API call, that 3s timeout needs to be respected,
 			// to not exceed the peerRequestTimeout (5s) of the client.
 			ctx, cancel := context.WithTimeout(context.Background(), peerRequestTimeout)