feat(instance): liveness probe isolation checker (cloudnative-pg#7466)

Enhances the liveness probe logic for primary instances in HA clusters
to detect network isolation scenarios.

The probe first checks if the Pod can reach the API server. If the API
server is unreachable, it then attempts to contact peer PostgreSQL
instances in the same cluster via a REST entrypoint.

If neither the API server nor any replicas are reachable, the liveness
probe fails, prompting Kubernetes to restart the Pod.

Upon restart, the instance manager will refuse to start PostgreSQL, as
it cannot download the Cluster definition—preventing unsafe behavior in
isolated environments.

This behavior applies only to primary instances in clusters with HA
replicas and is disabled by default.

Closes: cloudnative-pg#7465

Signed-off-by: Leonardo Cecchi <leonardo.cecchi@enterprisedb.com>
Signed-off-by: Armando Ruocco <armando.ruocco@enterprisedb.com>
Signed-off-by: Niccolò Fei <niccolo.fei@enterprisedb.com>
Signed-off-by: Gabriele Bartolini <gabriele.bartolini@enterprisedb.com>
Signed-off-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>
Co-authored-by: Armando Ruocco <armando.ruocco@enterprisedb.com>
Co-authored-by: Niccolò Fei <niccolo.fei@enterprisedb.com>
Co-authored-by: Gabriele Bartolini <gabriele.bartolini@enterprisedb.com>
Co-authored-by: Marco Nenciarini <marco.nenciarini@enterprisedb.com>

Author: 5 people

docs/src/instance_manager.md

@@ -188,6 +188,48 @@ spec:
       failureThreshold: 10
 ```
 
+### Primary Isolation (alpha)
+
+CloudNativePG 1.26 introduces an opt-in experimental behavior for the liveness
+probe of a PostgreSQL primary, which will report a failure if **both** of the
+following conditions are met:
+
+1. The instance manager cannot reach the Kubernetes API server
+2. The instance manager cannot reach **any** other instance via the instance manager’s REST API
+
+The effect of this behavior is to consider an isolated primary to be not alive and subsequently **shut it down** when the liveness probe fails.
+
+It is **disabled by default** and can be enabled by adding the following
+annotation to the `Cluster` resource:
+
+```yaml
+metadata:
+  annotations:
+    alpha.cnpg.io/livenessPinger: '{"enabled": true}'
+```
+
+!!! Warning
+    This feature is experimental and will be introduced in a future CloudNativePG
+    release with a new API. If you decide to use it now, note that the API **will
+    change**.
+
+!!! Important
+    If you plan to enable this experimental feature, be aware that the default
+    liveness probe settings—automatically derived from `livenessProbeTimeout`—might
+    be aggressive (30 seconds). As such, we recommend explicitly setting the
+    liveness probe configuration to suit your environment.
+
+The annotation also accepts two optional network settings: `requestTimeout`
+and `connectionTimeout`, both defaulting to `500` (in milliseconds).
+In cloud environments, you may need to increase these values.
+For example:
+
+```yaml
+metadata:
+  annotations:
+    alpha.cnpg.io/livenessPinger: '{"enabled": true,"requestTimeout":1000,"connectionTimeout":1000}'
+```
+
 ## Readiness Probe
 
 The readiness probe starts once the startup probe has successfully completed.

internal/webhook/v1/cluster_webhook.go

@@ -49,6 +49,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
 
 	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres/webserver/probes"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/postgres"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/specs"
 	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
@@ -218,6 +219,7 @@ func (v *ClusterCustomValidator) validate(r *apiv1.Cluster) (allErrs field.Error
 		v.validatePodPatchAnnotation,
 		v.validatePromotionToken,
 		v.validatePluginConfiguration,
+		v.validateLivenessPingerProbe,
 	}
 
 	for _, validate := range validations {
@@ -2453,3 +2455,23 @@ func (v *ClusterCustomValidator) validatePluginConfiguration(r *apiv1.Cluster) f
 
 	return errorList
 }
+
+func (v *ClusterCustomValidator) validateLivenessPingerProbe(r *apiv1.Cluster) field.ErrorList {
+	value, ok := r.Annotations[utils.LivenessPingerAnnotationName]
+	if !ok {
+		return nil
+	}
+
+	_, err := probes.NewLivenessPingerConfigFromAnnotations(context.Background(), r.Annotations)
+	if err != nil {
+		return field.ErrorList{
+			field.Invalid(
+				field.NewPath("metadata", "annotations", utils.LivenessPingerAnnotationName),
+				value,
+				fmt.Sprintf("error decoding liveness pinger config: %s", err.Error()),
+			),
+		}
+	}
+
+	return nil
+}

internal/webhook/v1/cluster_webhook_test.go

@@ -5089,3 +5089,43 @@ var _ = Describe("validatePluginConfiguration", func() {
 		Expect(v.validatePluginConfiguration(cluster)).To(BeNil())
 	})
 })
+
+var _ = Describe("", func() {
+	var v *ClusterCustomValidator
+	BeforeEach(func() {
+		v = &ClusterCustomValidator{}
+	})
+
+	It("returns no errors if the liveness pinger annotation is not present", func() {
+		cluster := &apiv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{},
+			},
+		}
+		Expect(v.validateLivenessPingerProbe(cluster)).To(BeNil())
+	})
+
+	It("returns no errors if the liveness pinger annotation is valid", func() {
+		cluster := &apiv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{
+					utils.LivenessPingerAnnotationName: `{"connectionTimeout": 1000, "requestTimeout": 5000, "enabled": true}`,
+				},
+			},
+		}
+		Expect(v.validateLivenessPingerProbe(cluster)).To(BeNil())
+	})
+
+	It("returns an error if the liveness pinger annotation is invalid", func() {
+		cluster := &apiv1.Cluster{
+			ObjectMeta: metav1.ObjectMeta{
+				Annotations: map[string]string{
+					utils.LivenessPingerAnnotationName: `{"requestTimeout": 5000}`,
+				},
+			},
+		}
+		errs := v.validateLivenessPingerProbe(cluster)
+		Expect(errs).To(HaveLen(1))
+		Expect(errs[0].Error()).To(ContainSubstring("error decoding liveness pinger config"))
+	})
+})

pkg/certs/tls.go

@@ -57,9 +57,18 @@ func newTLSConfigFromSecret(
 	// for the <cluster>-rw service, which would cause a name verification error.
 	caCertPool := x509.NewCertPool()
 	caCertPool.AppendCertsFromPEM(caCertificate)
+
+	return NewTLSConfigFromCertPool(caCertPool), nil
+}
+
+// NewTLSConfigFromCertPool creates a tls.Config object from X509 cert pool
+// containing the expected server CA
+func NewTLSConfigFromCertPool(
+	certPool *x509.CertPool,
+) *tls.Config {
 	tlsConfig := tls.Config{
 		MinVersion:         tls.VersionTLS13,
-		RootCAs:            caCertPool,
+		RootCAs:            certPool,
 		InsecureSkipVerify: true, //#nosec G402 -- we are verifying the certificate ourselves
 		VerifyPeerCertificate: func(rawCerts [][]byte, _ [][]*x509.Certificate) error {
 			// Code adapted from https://go.dev/src/crypto/tls/handshake_client.go#L986
@@ -77,7 +86,7 @@ func newTLSConfigFromSecret(
 			}
 
 			opts := x509.VerifyOptions{
-				Roots:         caCertPool,
+				Roots:         certPool,
 				Intermediates: x509.NewCertPool(),
 			}
 
@@ -93,7 +102,7 @@ func newTLSConfigFromSecret(
 		},
 	}
 
-	return &tlsConfig, nil
+	return &tlsConfig
 }
 
 // NewTLSConfigForContext creates a tls.config with the provided data and returns an expanded context that contains

pkg/management/postgres/webserver/probes/liveness.go

@@ -0,0 +1,171 @@
+/*
+Copyright © contributors to CloudNativePG, established as
+CloudNativePG a Series of LF Projects, LLC.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package probes
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+
+	"github.com/cloudnative-pg/machinery/pkg/log"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/management/postgres"
+)
+
+type livenessExecutor struct {
+	cli      client.Client
+	instance *postgres.Instance
+
+	lastestKnownCluster *apiv1.Cluster
+}
+
+// NewLivenessChecker creates a new instance of the liveness probe checker
+func NewLivenessChecker(
+	cli client.Client,
+	instance *postgres.Instance,
+) Checker {
+	return &livenessExecutor{
+		cli:      cli,
+		instance: instance,
+	}
+}
+
+// tryRefreshLatestCluster refreshes the latest cluster definition, returns a bool indicating if the operation was
+// successful
+func (e *livenessExecutor) tryRefreshLatestCluster(ctx context.Context) bool {
+	var cluster apiv1.Cluster
+	err := e.cli.Get(
+		ctx,
+		client.ObjectKey{Namespace: e.instance.GetNamespaceName(), Name: e.instance.GetClusterName()},
+		&cluster,
+	)
+	if err != nil {
+		return false
+	}
+
+	e.lastestKnownCluster = cluster.DeepCopy()
+	return true
+}
+
+func (e *livenessExecutor) IsHealthy(
+	ctx context.Context,
+	w http.ResponseWriter,
+) {
+	contextLogger := log.FromContext(ctx)
+
+	isPrimary, isPrimaryErr := e.instance.IsPrimary()
+	if isPrimaryErr != nil {
+		contextLogger.Error(
+			isPrimaryErr,
+			"Error while checking the instance role, skipping automatic shutdown.")
+		_, _ = fmt.Fprint(w, "OK")
+		return
+	}
+
+	if !isPrimary {
+		// There's no need to restart a replica if isolated
+		_, _ = fmt.Fprint(w, "OK")
+		return
+	}
+
+	if clusterRefreshed := e.tryRefreshLatestCluster(ctx); clusterRefreshed {
+		// We correctly reached the API server but, as a failsafe measure, we
+		// exercise the reachability checker and leave a log message if something
+		// is not right.
+		// In this way a network configuration problem can be discovered as
+		// quickly as possible.
+		if err := evaluateLivenessPinger(ctx, e.lastestKnownCluster.DeepCopy()); err != nil {
+			contextLogger.Warning(
+				"Instance connectivity error - liveness probe failing but API server is reachable",
+				"err",
+				err.Error(),
+			)
+		}
+		_, _ = fmt.Fprint(w, "OK")
+		return
+	}
+
+	contextLogger = contextLogger.WithValues("apiServerReachable", false)
+
+	if e.lastestKnownCluster == nil {
+		// We were never able to download a cluster definition. This should not
+		// happen because we check the API server connectivity as soon as the
+		// instance manager starts, before starting the probe web server.
+		//
+		// To be safe, we classify this instance manager to be not isolated and
+		// postpone any decision to a later liveness probe call.
+		contextLogger.Warning(
+			"No cluster definition has been received, skipping automatic shutdown.")
+
+		_, _ = fmt.Fprint(w, "OK")
+		return
+	}
+
+	err := evaluateLivenessPinger(ctx, e.lastestKnownCluster.DeepCopy())
+	if err != nil {
+		contextLogger.Error(err, "Instance connectivity error - liveness probe failing")
+		http.Error(
+			w,
+			fmt.Sprintf("liveness check failed: %s", err.Error()),
+			http.StatusInternalServerError,
+		)
+		return
+	}
+
+	contextLogger.Debug(
+		"Instance connectivity test succeeded - liveness probe succeeding",
+		"latestKnownInstancesReportedState", e.lastestKnownCluster.Status.InstancesReportedState,
+	)
+	_, _ = fmt.Fprint(w, "OK")
+}
+
+func evaluateLivenessPinger(
+	ctx context.Context,
+	cluster *apiv1.Cluster,
+) error {
+	contextLogger := log.FromContext(ctx)
+
+	cfg, err := NewLivenessPingerConfigFromAnnotations(ctx, cluster.Annotations)
+	if err != nil {
+		return err
+	}
+	if !cfg.isEnabled() {
+		contextLogger.Debug("pinger config not enabled, skipping")
+		return nil
+	}
+
+	if cluster.Spec.Instances == 1 {
+		contextLogger.Debug("Only one instance present in the latest known cluster definition. Skipping automatic shutdown.")
+		return nil
+	}
+
+	checker, err := buildInstanceReachabilityChecker(*cfg)
+	if err != nil {
+		return fmt.Errorf("failed to build instance reachability checker: %w", err)
+	}
+
+	if err := checker.ensureInstancesAreReachable(cluster); err != nil {
+		return fmt.Errorf("liveness check failed: %w", err)
+	}
+
+	return nil
+}