Repository: cloudnative-pg/cloudnative-pg

Issue:
[Feature]: Experimental - Isolation Checker for the liveness probe of a primary
### Is there an existing issue already for this feature request/idea?

- [x] I have searched for an existing issue, and could not find anything. I believe this is a new feature request to be evaluated.

### What problem is this feature going to solve? Why should it be added?

The problem is explained in discussion #7462 and relates to a potential and rare split-brain scenario caused by the non-transient isolation of a Postgres instance following a severe network partition.

### Describe the solution you'd like

The solution proposed in this specific patch is to explore the possibility of relying on the Kubernetes liveness probe infrastructure to enable an isolated primary to self-fence itself in case of severe isolation.

Right now, the liveness probe returns false only if the REST web server of the instance manager is down. While this is still good for a replica and a single instance cluster, we could extend its behaviour and fail if all the following conditions are met:

1. The instance is a primary in a multi-instance Postgres cluster.
2. The API server cannot be reached.
3. At least one replica’s WAL receiver is not connected to the primary.
4. At least one instance is unreachable via the instance manager's REST API.

We should consider this a prototype feature for the moment. Should it fail, we will take a different approach. If it succeeds, we can continue exploring:

- the user interface (the current implementation of the `spec.probes.liveness` stanza follows the Kubernetes standard, and we might think of introducing an option that enables/disables self-fencing)
- the default values (currently, the liveness probe is set to 30 seconds via the `.spec.livenessProbeTimeout` option)

As usual, during prototyping and exploration, feel free to change any of the above if better alternatives emerge.

### Describe alternatives you've considered

If the liveness probe approach fails, we need to find a different way to achieve this that is equivalent in terms of steps and outcomes, but will require a different API.

### Additional context

We should simulate the three main failures, cumulatively, in this order:

- API server is unreachable by the primary
- a WAL receiver is down
- an instance manager of a replica is non reachable

Network policies are the best way to test these scenarios, but I leave it up to the implementation.

### Backport?

No

### Are you willing to actively contribute to this feature?

Yes

### Code of Conduct

- [x] I agree to follow this project's Code of Conduct

Discussion:
@gbartolini: Related to #7407@gbartolini: @leonardoce built a proof of concept and I was able to test it. See https://github.com/cloudnative-pg/cloudnative-pg/pull/7466#issuecomment-2846748446

I believe we should proceed in this direction.@gbartolini: This feature has now been introduced as opt-in experimental in 1.26. The ticket for the definitive API to be in 1.27 is #7551 

Solution patch:
diff --git a/docs/src/instance_manager.md b/docs/src/instance_manager.md
index 0e1641d3..a914637b 100644
--- a/docs/src/instance_manager.md
+++ b/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.
diff --git a/internal/webhook/v1/cluster_webhook.go b/internal/webhook/v1/cluster_webhook.go
index 0aecfbc1..c52d5cb5 100644
--- a/internal/webhook/v1/cluster_webhook.go
+++ b/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
+}
diff --git a/pkg/certs/tls.go b/pkg/certs/tls.go
index 30c5b257..68b41ebc 100644
--- a/pkg/certs/tls.go
+++ b/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
diff --git a/pkg/management/postgres/webserver/probes/liveness.go b/pkg/management/postgres/webserver/probes/liveness.go
new file mode 100644
index 00000000..0449b767
--- /dev/null
+++ b/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
+}
diff --git a/pkg/management/postgres/webserver/probes/pinger.go b/pkg/management/postgres/webserver/probes/pinger.go
new file mode 100644
index 00000000..a6647032
--- /dev/null
+++ b/pkg/management/postgres/webserver/probes/pinger.go
@@ -0,0 +1,204 @@
+/*
+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"
+	"crypto/x509"
+	"encoding/json"
+	"fmt"
+	"net"
+	"net/http"
+	"net/url"
+	"os"
+	"time"
+
+	"github.com/cloudnative-pg/machinery/pkg/log"
+	"k8s.io/utils/ptr"
+
+	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/certs"
+	cnpgUrl "github.com/cloudnative-pg/cloudnative-pg/pkg/management/url"
+	postgresSpec "github.com/cloudnative-pg/cloudnative-pg/pkg/postgres"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
+)
+
+// LivenessPingerCfg if the configuration of the instance
+// reachability checker
+type LivenessPingerCfg struct {
+	Enabled           *bool `json:"enabled"`
+	RequestTimeout    int   `json:"requestTimeout,omitempty"`
+	ConnectionTimeout int   `json:"connectionTimeout,omitempty"`
+}
+
+func (probe *LivenessPingerCfg) isEnabled() bool {
+	if probe == nil || probe.Enabled == nil {
+		return false
+	}
+
+	return *probe.Enabled
+}
+
+// NewLivenessPingerConfigFromAnnotations creates a new pinger configuration from the annotations
+// in the cluster definition
+func NewLivenessPingerConfigFromAnnotations(
+	ctx context.Context,
+	annotations map[string]string,
+) (*LivenessPingerCfg, error) {
+	const (
+		// defaultRequestTimeout is the default value of the request timeout
+		defaultRequestTimeout = 500
+
+		// defaultConnectionTimeout is the default value of the connection timeout
+		defaultConnectionTimeout = 1000
+	)
+
+	contextLogger := log.FromContext(ctx)
+
+	v, ok := annotations[utils.LivenessPingerAnnotationName]
+	if !ok {
+		contextLogger.Debug("pinger config not found in the cluster annotations")
+		return &LivenessPingerCfg{
+			Enabled: ptr.To(false),
+		}, nil
+	}
+
+	var cfg LivenessPingerCfg
+	if err := json.Unmarshal([]byte(v), &cfg); err != nil {
+		contextLogger.Error(err, "failed to unmarshal pinger config")
+		return nil, fmt.Errorf("while unmarshalling pinger config: %w", err)
+	}
+
+	if cfg.Enabled == nil {
+		return nil, fmt.Errorf("pinger config is missing the enabled field")
+	}
+
+	if cfg.RequestTimeout == 0 {
+		cfg.RequestTimeout = defaultRequestTimeout
+	}
+	if cfg.ConnectionTimeout == 0 {
+		cfg.ConnectionTimeout = defaultConnectionTimeout
+	}
+
+	return &cfg, nil
+}
+
+// pinger can check if a certain instance is reachable by using
+// the failsafe REST endpoint
+type pinger struct {
+	dialer *net.Dialer
+	client *http.Client
+
+	config LivenessPingerCfg
+}
+
+// buildInstanceReachabilityChecker creates a new instance reachability checker by loading
+// the server CA certificate from the same location that will be used by PostgreSQL.
+// In this case, we avoid using the API Server as it may be unreliable.
+func buildInstanceReachabilityChecker(cfg LivenessPingerCfg) (*pinger, error) {
+	certificateLocation := postgresSpec.ServerCACertificateLocation
+	caCertificate, err := os.ReadFile(certificateLocation) //nolint:gosec
+	if err != nil {
+		return nil, fmt.Errorf("while reading server CA certificate [%s]: %w", certificateLocation, err)
+	}
+
+	caCertPool := x509.NewCertPool()
+	caCertPool.AppendCertsFromPEM(caCertificate)
+
+	tlsConfig := certs.NewTLSConfigFromCertPool(caCertPool)
+
+	dialer := &net.Dialer{Timeout: time.Duration(cfg.ConnectionTimeout) * time.Millisecond}
+
+	client := http.Client{
+		Transport: &http.Transport{
+			DialContext:     dialer.DialContext,
+			TLSClientConfig: tlsConfig,
+		},
+		Timeout: time.Duration(cfg.RequestTimeout) * time.Millisecond,
+	}
+
+	return &pinger{
+		dialer: dialer,
+		client: &client,
+		config: cfg,
+	}, nil
+}
+
+// ping checks if the instance with the passed coordinates is reachable
+// by calling the failsafe endpoint.
+func (e *pinger) ping(host, ip string) error {
+	failsafeURL := url.URL{
+		Scheme: "https",
+		Host:   fmt.Sprintf("%s:%d", ip, cnpgUrl.StatusPort),
+		Path:   cnpgUrl.PathFailSafe,
+	}
+
+	var res *http.Response
+	var err error
+	if res, err = e.client.Get(failsafeURL.String()); err != nil {
+		return &pingError{
+			host:   host,
+			err:    err,
+			config: e.config,
+		}
+	}
+
+	_ = res.Body.Close()
+
+	return nil
+}
+
+func (e pinger) ensureInstancesAreReachable(cluster *apiv1.Cluster) error {
+	for name, state := range cluster.Status.InstancesReportedState {
+		host := string(name)
+		ip := state.IP
+		if err := e.ping(host, ip); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// pingError is raised when the instance connectivity test failed.
+type pingError struct {
+	host string
+	ip   string
+
+	config LivenessPingerCfg
+
+	err error
+}
+
+// Error implements the error interface
+func (e *pingError) Error() string {
+	return fmt.Sprintf(
+		"instance connectivity error for instance [%s] with ip [%s] (requestTimeout:%v connectionTimeout:%v): %s",
+		e.host,
+		e.ip,
+		e.config.RequestTimeout,
+		e.config.ConnectionTimeout,
+		e.err.Error())
+}
+
+// Unwrap implements the error interface
+func (e *pingError) Unwrap() error {
+	return e.err
+}
diff --git a/pkg/management/postgres/webserver/remote.go b/pkg/management/postgres/webserver/remote.go
index cf6656e9..ef76017b 100644
--- a/pkg/management/postgres/webserver/remote.go
+++ b/pkg/management/postgres/webserver/remote.go
@@ -66,6 +66,8 @@ type remoteWebserverEndpoints struct {
 	instance             *postgres.Instance
 	currentBackup        *backupConnection
 	ongoingBackupRequest sync.Mutex
+	// livenessChecker is a  stateful probe
+	livenessChecker probes.Checker
 }
 
 // StartBackupRequest the required data to execute the pg_start_backup
@@ -97,11 +99,13 @@ func NewRemoteWebServer(
 	}
 
 	endpoints := remoteWebserverEndpoints{
-		typedClient: typedClient,
-		instance:    instance,
+		typedClient:     typedClient,
+		instance:        instance,
+		livenessChecker: probes.NewLivenessChecker(typedClient, instance),
 	}
 
 	serveMux := http.NewServeMux()
+	serveMux.HandleFunc(url.PathFailSafe, endpoints.failSafe)
 	serveMux.HandleFunc(url.PathPgModeBackup, endpoints.backup)
 	serveMux.HandleFunc(url.PathHealth, endpoints.isServerHealthy)
 	serveMux.HandleFunc(url.PathReady, endpoints.isServerReady)
@@ -222,10 +226,16 @@ func (ws *remoteWebserverEndpoints) isServerStartedUp(w http.ResponseWriter, req
 	checker.IsHealthy(req.Context(), w)
 }
 
-func (ws *remoteWebserverEndpoints) isServerHealthy(w http.ResponseWriter, _ *http.Request) {
+// This is the failsafe entrypoint
+func (ws *remoteWebserverEndpoints) failSafe(w http.ResponseWriter, _ *http.Request) {
 	_, _ = fmt.Fprint(w, "OK")
 }
 
+// This is the failsafe probe
+func (ws *remoteWebserverEndpoints) isServerHealthy(w http.ResponseWriter, req *http.Request) {
+	ws.livenessChecker.IsHealthy(req.Context(), w)
+}
+
 // This is the readiness probe
 func (ws *remoteWebserverEndpoints) isServerReady(w http.ResponseWriter, req *http.Request) {
 	if !ws.instance.CanCheckReadiness() {
diff --git a/pkg/management/url/url.go b/pkg/management/url/url.go
index c54cde86..b3157163 100644
--- a/pkg/management/url/url.go
+++ b/pkg/management/url/url.go
@@ -34,6 +34,9 @@ const (
 	// PgBouncerMetricsPort is the port for the exporter of PgBouncer related metrics (HTTP)
 	PgBouncerMetricsPort int32 = 9127
 
+	// PathFailSafe is the path for the failsafe entrypoint
+	PathFailSafe string = "/failsafe"
+
 	// PathHealth is the URL path for Health State
 	PathHealth string = "/healthz"
 
diff --git a/pkg/utils/labels_annotations.go b/pkg/utils/labels_annotations.go
index 37143c54..1ce7eddf 100644
--- a/pkg/utils/labels_annotations.go
+++ b/pkg/utils/labels_annotations.go
@@ -30,6 +30,10 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 )
 
+// AlphaMetadataNamespace is the annotation and label namespace used by the alpha features of
+// the operator
+const AlphaMetadataNamespace = "alpha.cnpg.io"
+
 // MetadataNamespace is the annotation and label namespace used by the operator
 const MetadataNamespace = "cnpg.io"
 
@@ -105,6 +109,9 @@ const (
 	// PluginNameLabelName is the name of the label to be applied to services
 	// to have them detected as CNPG-i plugins
 	PluginNameLabelName = MetadataNamespace + "/pluginName"
+
+	// LivenessPingerAnnotationName is the name of the pinger configuration
+	LivenessPingerAnnotationName = AlphaMetadataNamespace + "/livenessPinger"
 )
 
 const (



Test patch:
diff --git a/internal/webhook/v1/cluster_webhook_test.go b/internal/webhook/v1/cluster_webhook_test.go
index a652aced..d92b6a90 100644
--- a/internal/webhook/v1/cluster_webhook_test.go
+++ b/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"))
+	})
+})
diff --git a/tests/e2e/fixtures/self-fencing/cluster-self-fencing.yaml.template b/tests/e2e/fixtures/self-fencing/cluster-self-fencing.yaml.template
new file mode 100644
index 00000000..5235464f
--- /dev/null
+++ b/tests/e2e/fixtures/self-fencing/cluster-self-fencing.yaml.template
@@ -0,0 +1,27 @@
+apiVersion: postgresql.cnpg.io/v1
+kind: Cluster
+metadata:
+  name: postgresql-self-fencing
+  annotations:
+    alpha.cnpg.io/livenessPinger: '{"enabled": true}'
+spec:
+  instances: 3
+
+  postgresql:
+    parameters:
+      max_connections: "110"
+      log_checkpoints: "on"
+      log_lock_waits: "on"
+      log_min_duration_statement: '1000'
+      log_statement: 'ddl'
+      log_temp_files: '1024'
+      log_autovacuum_min_duration: '1s'
+      log_replication_commands: 'on'
+
+  # Persistent storage configuration
+  storage:
+    storageClass: ${E2E_DEFAULT_STORAGE_CLASS}
+    size: 1Gi
+  walStorage:
+    storageClass: ${E2E_DEFAULT_STORAGE_CLASS}
+    size: 1Gi
diff --git a/tests/e2e/self_fencing_test.go b/tests/e2e/self_fencing_test.go
new file mode 100644
index 00000000..072ed327
--- /dev/null
+++ b/tests/e2e/self_fencing_test.go
@@ -0,0 +1,142 @@
+/*
+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 e2e
+
+import (
+	"fmt"
+	"strings"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/types"
+
+	apiv1 "github.com/cloudnative-pg/cloudnative-pg/api/v1"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/specs"
+	"github.com/cloudnative-pg/cloudnative-pg/pkg/utils"
+	"github.com/cloudnative-pg/cloudnative-pg/tests"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/clusterutils"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/operator"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/run"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/timeouts"
+	"github.com/cloudnative-pg/cloudnative-pg/tests/utils/yaml"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+var _ = Describe("Self-fencing with liveness probe", Serial, Label(tests.LabelDisruptive), func() {
+	const (
+		level           = tests.Lowest
+		sampleFile      = fixturesDir + "/self-fencing/cluster-self-fencing.yaml.template"
+		namespacePrefix = "self-fencing"
+	)
+
+	BeforeEach(func() {
+		if testLevelEnv.Depth < int(level) {
+			Skip("Test depth is lower than the amount requested for this test")
+		}
+		if !IsLocal() {
+			Skip("This test is only run on local cluster")
+		}
+	})
+
+	It("will terminate an isolated primary", func() {
+		var namespace, clusterName, isolatedNode string
+		var err error
+		var oldPrimaryPod *corev1.Pod
+
+		DeferCleanup(func() {
+			// Ensure the isolatedNode networking is re-established
+			if CurrentSpecReport().Failed() {
+				_, _, _ = run.Unchecked(fmt.Sprintf("docker network connect kind %v", isolatedNode))
+			}
+		})
+
+		By("creating a Cluster", func() {
+			clusterName, err = yaml.GetResourceNameFromYAML(env.Scheme, sampleFile)
+			Expect(err).ToNot(HaveOccurred())
+			namespace, err = env.CreateUniqueTestNamespace(env.Ctx, env.Client, namespacePrefix)
+			Expect(err).ToNot(HaveOccurred())
+			AssertCreateCluster(namespace, clusterName, sampleFile, env)
+		})
+
+		By("setting up the environment", func() {
+			// Ensure the operator is not running on the same node as the primary.
+			// If it is, we switch to a new primary
+			primaryPod, err := clusterutils.GetPrimary(env.Ctx, env.Client, namespace, clusterName)
+			Expect(err).ToNot(HaveOccurred())
+			operatorPod, err := operator.GetPod(env.Ctx, env.Client)
+			Expect(err).NotTo(HaveOccurred())
+			if primaryPod.Spec.NodeName == operatorPod.Spec.NodeName {
+				AssertSwitchover(namespace, clusterName, env)
+			}
+		})
+
+		By("disconnecting the node containing the primary", func() {
+			oldPrimaryPod, err = clusterutils.GetPrimary(env.Ctx, env.Client, namespace, clusterName)
+			Expect(err).ToNot(HaveOccurred())
+			isolatedNode = oldPrimaryPod.Spec.NodeName
+			_, _, err = run.Unchecked(fmt.Sprintf("docker network disconnect kind %v", isolatedNode))
+			Expect(err).ToNot(HaveOccurred())
+		})
+
+		By("verifying that a new primary has been promoted", func() {
+			AssertClusterEventuallyReachesPhase(namespace, clusterName,
+				[]string{apiv1.PhaseFailOver}, 120)
+			Eventually(func(g Gomega) {
+				cluster, err := clusterutils.Get(env.Ctx, env.Client, namespace, clusterName)
+				g.Expect(err).ToNot(HaveOccurred())
+				g.Expect(cluster.Status.CurrentPrimary).ToNot(BeEquivalentTo(oldPrimaryPod.Name))
+			}, testTimeouts[timeouts.NewPrimaryAfterFailover]).Should(Succeed())
+		})
+
+		By("verifying that oldPrimary will self isolate", func() {
+			// Assert that the oldPrimary is eventually terminated
+			Eventually(func(g Gomega) {
+				out, _, err := run.Unchecked(fmt.Sprintf(
+					"docker exec %v crictl ps -a "+
+						"--label io.kubernetes.pod.namespace=%s,io.kubernetes.pod.name=%s "+
+						"--name postgres -s Exited -q", isolatedNode, namespace, oldPrimaryPod.Name))
+				g.Expect(err).ToNot(HaveOccurred())
+				g.Expect(out).ToNot(BeEmpty())
+				if out != "" {
+					GinkgoWriter.Printf("Container %s (%s) has been terminated\n",
+						oldPrimaryPod.Name, strings.TrimSpace(out))
+				}
+			}, 120).Should(Succeed())
+		})
+
+		By("reconnecting the isolated Node", func() {
+			_, _, err = run.Unchecked(fmt.Sprintf("docker network connect kind %v", isolatedNode))
+			Expect(err).ToNot(HaveOccurred())
+
+			// Assert that the oldPrimary comes back as a replica
+			namespacedName := types.NamespacedName{
+				Namespace: oldPrimaryPod.Namespace,
+				Name:      oldPrimaryPod.Name,
+			}
+			timeout := 180
+			Eventually(func() (bool, error) {
+				pod := corev1.Pod{}
+				err := env.Client.Get(env.Ctx, namespacedName, &pod)
+				return utils.IsPodActive(pod) && utils.IsPodReady(pod) && specs.IsPodStandby(pod), err
+			}, timeout).Should(BeTrue())
+		})
+	})
+})