Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion docs/reference/operator_parameters.md
Original file line number Diff line number Diff line change
Expand Up @@ -556,7 +556,10 @@ configuration they are grouped under the `kubernetes` key.
* **master_pod_move_timeout**
The period of time to wait for the success of migration of master pods from
an unschedulable node. The migration includes Patroni switchovers to
respective replicas on healthy nodes. The situation where master pods still
respective replicas on healthy nodes. For a single-pod cluster, the operator
instead recreates the master on another node and waits for its role label,
without attempting a switchover. This causes downtime until the pod returns.
The situation where master pods still
exist on the old node after this timeout expires has to be fixed manually.
The default is 20 minutes.

Expand Down
2 changes: 1 addition & 1 deletion pkg/cluster/pod.go
Original file line number Diff line number Diff line change
Expand Up @@ -250,7 +250,7 @@ func (c *Cluster) MigrateMasterPod(podName spec.NamespacedName) error {
}
// we may not have a cached statefulset if the initial cluster sync has aborted, revert to the spec in that case
masterCandidateName := podName
masterCandidatePod := oldMaster
var masterCandidatePod *v1.Pod
if *c.Statefulset.Spec.Replicas > 1 {
if masterCandidateName, err = c.getSwitchoverCandidate(oldMaster); err != nil {
return fmt.Errorf("could not find suitable replica pod as candidate for failover: %v", err)
Expand Down
128 changes: 128 additions & 0 deletions pkg/cluster/pod_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -5,6 +5,7 @@ import (
"fmt"
"io"
"net/http"
"strings"
"testing"
"time"

Expand All @@ -15,10 +16,137 @@ import (
"github.com/zalando/postgres-operator/v2/pkg/util/config"
"github.com/zalando/postgres-operator/v2/pkg/util/k8sutil"
"github.com/zalando/postgres-operator/v2/pkg/util/patroni"
appsv1 "k8s.io/api/apps/v1"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
k8sfake "k8s.io/client-go/kubernetes/fake"
k8stesting "k8s.io/client-go/testing"
"k8s.io/client-go/tools/record"
)

func TestMigrateSingleMasterPod(t *testing.T) {
for _, tt := range []struct {
name string
newNode string
deleteError error
expectedError string
}{
{name: "relocated without switchover", newNode: "new-node"},
{name: "deletion fails", deleteError: fmt.Errorf("delete failed"), expectedError: "delete failed"},
{name: "pod remains on old node", newNode: "old-node", expectedError: "remained on the same node"},
} {
t.Run(tt.name, func(t *testing.T) {
podName := spec.NamespacedName{Namespace: "default", Name: "acid-test-cluster-0"}
oldPod := &v1.Pod{
ObjectMeta: metav1.ObjectMeta{Name: podName.Name, Namespace: podName.Namespace, Labels: map[string]string{"spilo-role": "master"}},
Spec: v1.PodSpec{NodeName: "old-node"},
Status: v1.PodStatus{PodIP: "192.0.2.1"},
}
newPod := oldPod.DeepCopy()
newPod.Spec.NodeName = tt.newNode
newPod.Status.PodIP = "192.0.2.2"
client := k8sfake.NewSimpleClientset(oldPod,
&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "old-node"}, Spec: v1.NodeSpec{Unschedulable: true}},
&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "new-node"}},
)
opConfig := config.Config{}
opConfig.PodRoleLabel = "spilo-role"
opConfig.PodDeletionWaitTimeout = &metav1.Duration{Duration: time.Second}
opConfig.PodLabelWaitTimeout = &metav1.Duration{Duration: time.Second}
c := New(Config{OpConfig: opConfig}, k8sutil.KubernetesClient{PodsGetter: client.CoreV1(), NodesGetter: client.CoreV1()},
acidv1.Postgresql{ObjectMeta: metav1.ObjectMeta{Name: "acid-test-cluster", Namespace: podName.Namespace}}, logger, record.NewFakeRecorder(2))
replicas := int32(1)
c.Statefulset = &appsv1.StatefulSet{Spec: appsv1.StatefulSetSpec{Replicas: &replicas}}
// A single-member cluster must not make any Patroni switchover request,
// especially to the IP of the deleted pod.
c.patroni = patroni.New(patroniLogger, mocks.NewMockHTTPClient(gomock.NewController(t)))
deletions := 0
client.PrependReactor("delete", "pods", func(action k8stesting.Action) (bool, runtime.Object, error) {
deletions++
if tt.deleteError != nil {
return true, nil, tt.deleteError
}
ch := c.podSubscribers[podName]
go func() {
ch <- PodEvent{EventType: PodEventDelete, PrevPod: oldPod}
ch <- PodEvent{EventType: PodEventAdd, CurPod: newPod}
}()
return true, nil, nil
})
err := c.MigrateMasterPod(podName)
if tt.expectedError == "" {
if err != nil {
t.Fatalf("migration failed: %v", err)
}
} else if err == nil || !strings.Contains(err.Error(), tt.expectedError) {
t.Fatalf("expected error containing %q, got %v", tt.expectedError, err)
}
if deletions != 1 {
t.Fatalf("expected one pod recreation, got %d deletions", deletions)
}
if len(c.podSubscribers) != 0 {
t.Fatal("pod event subscription was not removed")
}
})
}
}

func TestMigrateMasterPodWithReplica(t *testing.T) {
podName := spec.NamespacedName{Namespace: "default", Name: "acid-test-cluster-0"}
master := &v1.Pod{
ObjectMeta: metav1.ObjectMeta{Name: podName.Name, Namespace: podName.Namespace, Labels: map[string]string{"spilo-role": "master"}},
Spec: v1.PodSpec{NodeName: "old-node"},
Status: v1.PodStatus{PodIP: "192.0.2.1"},
}
replica := master.DeepCopy()
replica.Name = "acid-test-cluster-1"
replica.Labels["spilo-role"] = "replica"
replica.Spec.NodeName = "new-node"
replica.Status.PodIP = "192.0.2.2"
client := k8sfake.NewSimpleClientset(master, replica,
&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "old-node"}, Spec: v1.NodeSpec{Unschedulable: true}},
&v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "new-node"}},
)
opConfig := config.Config{}
opConfig.PodRoleLabel = "spilo-role"
opConfig.PodLabelWaitTimeout = &metav1.Duration{Duration: time.Second}
opConfig.PatroniAPICheckInterval = &metav1.Duration{Duration: time.Millisecond}
opConfig.PatroniAPICheckTimeout = &metav1.Duration{Duration: time.Second}
c := New(Config{OpConfig: opConfig}, k8sutil.KubernetesClient{PodsGetter: client.CoreV1(), NodesGetter: client.CoreV1()},
acidv1.Postgresql{ObjectMeta: metav1.ObjectMeta{Name: "acid-test-cluster", Namespace: podName.Namespace}}, logger, record.NewFakeRecorder(2))
replicas := int32(2)
c.Statefulset = &appsv1.StatefulSet{Spec: appsv1.StatefulSetSpec{Replicas: &replicas}}
mockClient := mocks.NewMockHTTPClient(gomock.NewController(t))
c.patroni = patroni.New(patroniLogger, mockClient)
mockClient.EXPECT().Get("http://192.0.2.1:8008/cluster").Return(&http.Response{
StatusCode: http.StatusOK,
Body: io.NopCloser(strings.NewReader(`{"members":[{"name":"acid-test-cluster-1","role":"replica","state":"streaming","lag":0}]}`)),
}, nil)
mockClient.EXPECT().Do(gomock.Any()).DoAndReturn(func(req *http.Request) (*http.Response, error) {
body, err := io.ReadAll(req.Body)
if err != nil {
t.Fatal(err)
}
if req.Method != http.MethodPost || req.URL.String() != "http://192.0.2.1:8008/switchover" || !strings.Contains(string(body), `"member":"acid-test-cluster-1"`) {
t.Fatalf("unexpected switchover: %s %s %s", req.Method, req.URL, body)
}
ch := c.podSubscribers[spec.NamespacedName{Namespace: replica.Namespace, Name: replica.Name}]
promoted := replica.DeepCopy()
promoted.Labels["spilo-role"] = "master"
go func() { ch <- PodEvent{EventType: PodEventUpdate, CurPod: promoted} }()
return &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(""))}, nil
})
if err := c.MigrateMasterPod(podName); err != nil {
t.Fatalf("migration failed: %v", err)
}
for _, action := range client.Actions() {
if action.GetVerb() == "delete" {
t.Fatal("a healthy replica must not be recreated")
}
}
}

func TestGetSwitchoverCandidate(t *testing.T) {
testName := "test getting right switchover candidate"
namespace := "default"
Expand Down