Merge branch 'master' into default-privileges-for-users

This commit is contained in:
Felix Kunde
2021-06-22 16:29:58 +02:00
26 changed files with 818 additions and 564 deletions
+79 -11
View File
@@ -5,6 +5,7 @@ package cluster
import (
"context"
"database/sql"
"encoding/json"
"fmt"
"reflect"
"regexp"
@@ -519,7 +520,7 @@ func (c *Cluster) compareContainers(description string, setA, setB []v1.Containe
newCheck("new statefulset %s's %s (index %d) resources do not match the current ones",
func(a, b v1.Container) bool { return !compareResources(&a.Resources, &b.Resources) }),
newCheck("new statefulset %s's %s (index %d) environment does not match the current one",
func(a, b v1.Container) bool { return !reflect.DeepEqual(a.Env, b.Env) }),
func(a, b v1.Container) bool { return !compareEnv(a.Env, b.Env) }),
newCheck("new statefulset %s's %s (index %d) environment sources do not match the current one",
func(a, b v1.Container) bool { return !reflect.DeepEqual(a.EnvFrom, b.EnvFrom) }),
newCheck("new statefulset %s's %s (index %d) security context does not match the current one",
@@ -576,6 +577,56 @@ func compareResourcesAssumeFirstNotNil(a *v1.ResourceRequirements, b *v1.Resourc
}
func compareEnv(a, b []v1.EnvVar) bool {
if len(a) != len(b) {
return false
}
equal := true
for _, enva := range a {
hasmatch := false
for _, envb := range b {
if enva.Name == envb.Name {
hasmatch = true
if enva.Name == "SPILO_CONFIGURATION" {
equal = compareSpiloConfiguration(enva.Value, envb.Value)
} else {
if enva.Value == "" && envb.Value == "" {
equal = reflect.DeepEqual(enva.ValueFrom, envb.ValueFrom)
} else {
equal = (enva.Value == envb.Value)
}
}
if !equal {
return false
}
}
}
if !hasmatch {
return false
}
}
return true
}
func compareSpiloConfiguration(configa, configb string) bool {
var (
oa, ob spiloConfiguration
)
var err error
err = json.Unmarshal([]byte(configa), &oa)
if err != nil {
return false
}
oa.Bootstrap.DCS = patroniDCS{}
err = json.Unmarshal([]byte(configb), &ob)
if err != nil {
return false
}
ob.Bootstrap.DCS = patroniDCS{}
return reflect.DeepEqual(oa, ob)
}
func (c *Cluster) enforceMinResourceLimits(spec *acidv1.PostgresSpec) error {
var (
@@ -940,14 +991,16 @@ func (c *Cluster) initSystemUsers() {
// secrets, therefore, setting flags like SUPERUSER or REPLICATION
// is not necessary here
c.systemUsers[constants.SuperuserKeyName] = spec.PgUser{
Origin: spec.RoleOriginSystem,
Name: c.OpConfig.SuperUsername,
Password: util.RandomPassword(constants.PasswordLength),
Origin: spec.RoleOriginSystem,
Name: c.OpConfig.SuperUsername,
Namespace: c.Namespace,
Password: util.RandomPassword(constants.PasswordLength),
}
c.systemUsers[constants.ReplicationUserKeyName] = spec.PgUser{
Origin: spec.RoleOriginSystem,
Name: c.OpConfig.ReplicationUsername,
Password: util.RandomPassword(constants.PasswordLength),
Origin: spec.RoleOriginSystem,
Name: c.OpConfig.ReplicationUsername,
Namespace: c.Namespace,
Password: util.RandomPassword(constants.PasswordLength),
}
// Connection pooler user is an exception, if requested it's going to be
@@ -975,10 +1028,11 @@ func (c *Cluster) initSystemUsers() {
// connection pooler application should be able to login with this role
connectionPoolerUser := spec.PgUser{
Origin: spec.RoleConnectionPooler,
Name: username,
Flags: []string{constants.RoleFlagLogin},
Password: util.RandomPassword(constants.PasswordLength),
Origin: spec.RoleConnectionPooler,
Name: username,
Namespace: c.Namespace,
Flags: []string{constants.RoleFlagLogin},
Password: util.RandomPassword(constants.PasswordLength),
}
if _, exists := c.pgUsers[username]; !exists {
@@ -1081,6 +1135,7 @@ func (c *Cluster) initDefaultRoles(defaultRoles map[string]string, admin, prefix
newRole := spec.PgUser{
Origin: spec.RoleOriginBootstrap,
Name: roleName,
Namespace: c.Namespace,
Password: util.RandomPassword(constants.PasswordLength),
Flags: flags,
MemberOf: memberOf,
@@ -1105,6 +1160,17 @@ func (c *Cluster) initRobotUsers() error {
if c.shouldAvoidProtectedOrSystemRole(username, "manifest robot role") {
continue
}
namespace := c.Namespace
//if namespaced secrets are allowed
if c.Postgresql.Spec.EnableNamespacedSecret != nil &&
*c.Postgresql.Spec.EnableNamespacedSecret {
if strings.Contains(username, ".") {
splits := strings.Split(username, ".")
namespace = splits[0]
}
}
flags, err := normalizeUserFlags(userFlags)
if err != nil {
return fmt.Errorf("invalid flags for user %q: %v", username, err)
@@ -1116,6 +1182,7 @@ func (c *Cluster) initRobotUsers() error {
newRole := spec.PgUser{
Origin: spec.RoleOriginManifest,
Name: username,
Namespace: namespace,
Password: util.RandomPassword(constants.PasswordLength),
Flags: flags,
AdminRole: adminRole,
@@ -1233,6 +1300,7 @@ func (c *Cluster) initInfrastructureRoles() error {
return fmt.Errorf("invalid flags for user '%v': %v", username, err)
}
newRole.Flags = flags
newRole.Namespace = c.Namespace
if currentRole, present := c.pgUsers[username]; present {
c.pgUsers[username] = c.resolveNameConflict(&currentRole, &newRole)
+245 -2
View File
@@ -7,12 +7,15 @@ import (
"github.com/sirupsen/logrus"
acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
fakeacidv1 "github.com/zalando/postgres-operator/pkg/generated/clientset/versioned/fake"
"github.com/zalando/postgres-operator/pkg/spec"
"github.com/zalando/postgres-operator/pkg/util/config"
"github.com/zalando/postgres-operator/pkg/util/constants"
"github.com/zalando/postgres-operator/pkg/util/k8sutil"
"github.com/zalando/postgres-operator/pkg/util/teams"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/kubernetes/fake"
"k8s.io/client-go/tools/record"
)
@@ -79,8 +82,8 @@ func TestInitRobotUsers(t *testing.T) {
}{
{
manifestUsers: map[string]acidv1.UserFlags{"foo": {"superuser", "createdb"}},
infraRoles: map[string]spec.PgUser{"foo": {Origin: spec.RoleOriginInfrastructure, Name: "foo", Password: "bar"}},
result: map[string]spec.PgUser{"foo": {Origin: spec.RoleOriginInfrastructure, Name: "foo", Password: "bar"}},
infraRoles: map[string]spec.PgUser{"foo": {Origin: spec.RoleOriginInfrastructure, Name: "foo", Namespace: cl.Namespace, Password: "bar"}},
result: map[string]spec.PgUser{"foo": {Origin: spec.RoleOriginInfrastructure, Name: "foo", Namespace: cl.Namespace, Password: "bar"}},
err: nil,
},
{
@@ -845,3 +848,243 @@ func TestPreparedDatabases(t *testing.T) {
}
}
}
func TestCompareSpiloConfiguration(t *testing.T) {
testCases := []struct {
Config string
ExpectedResult bool
}{
{
`{"postgresql":{"bin_dir":"/usr/lib/postgresql/12/bin","parameters":{"autovacuum_analyze_scale_factor":"0.1"},"pg_hba":["hostssl all all 0.0.0.0/0 md5","host all all 0.0.0.0/0 md5"]},"bootstrap":{"initdb":[{"auth-host":"md5"},{"auth-local":"trust"},"data-checksums",{"encoding":"UTF8"},{"locale":"en_US.UTF-8"}],"users":{"test":{"password":"","options":["CREATEDB","NOLOGIN"]}},"dcs":{"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432,"postgresql":{"parameters":{"max_connections":"100","max_locks_per_transaction":"64","max_worker_processes":"4"}}}}}`,
true,
},
{
`{"postgresql":{"bin_dir":"/usr/lib/postgresql/12/bin","parameters":{"autovacuum_analyze_scale_factor":"0.1"},"pg_hba":["hostssl all all 0.0.0.0/0 md5","host all all 0.0.0.0/0 md5"]},"bootstrap":{"initdb":[{"auth-host":"md5"},{"auth-local":"trust"},"data-checksums",{"encoding":"UTF8"},{"locale":"en_US.UTF-8"}],"users":{"test":{"password":"","options":["CREATEDB","NOLOGIN"]}},"dcs":{"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432,"postgresql":{"parameters":{"max_connections":"200","max_locks_per_transaction":"64","max_worker_processes":"4"}}}}}`,
true,
},
{
`{"postgresql":{"bin_dir":"/usr/lib/postgresql/12/bin","parameters":{"autovacuum_analyze_scale_factor":"0.1"},"pg_hba":["hostssl all all 0.0.0.0/0 md5","host all all 0.0.0.0/0 md5"]},"bootstrap":{"initdb":[{"auth-host":"md5"},{"auth-local":"trust"},"data-checksums",{"encoding":"UTF8"},{"locale":"en_US.UTF-8"}],"users":{"test":{"password":"","options":["CREATEDB"]}},"dcs":{"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432,"postgresql":{"parameters":{"max_connections":"200","max_locks_per_transaction":"64","max_worker_processes":"4"}}}}}`,
false,
},
{
`{}`,
false,
},
{
`invalidjson`,
false,
},
}
refCase := testCases[0]
for _, testCase := range testCases {
if result := compareSpiloConfiguration(refCase.Config, testCase.Config); result != testCase.ExpectedResult {
t.Errorf("expected %v got %v", testCase.ExpectedResult, result)
}
}
}
func TestCompareEnv(t *testing.T) {
testCases := []struct {
Envs []v1.EnvVar
ExpectedResult bool
}{
{
Envs: []v1.EnvVar{
{
Name: "VARIABLE1",
Value: "value1",
},
{
Name: "VARIABLE2",
Value: "value2",
},
{
Name: "VARIABLE3",
Value: "value3",
},
{
Name: "SPILO_CONFIGURATION",
Value: `{"postgresql":{"bin_dir":"/usr/lib/postgresql/12/bin","parameters":{"autovacuum_analyze_scale_factor":"0.1"},"pg_hba":["hostssl all all 0.0.0.0/0 md5","host all all 0.0.0.0/0 md5"]},"bootstrap":{"initdb":[{"auth-host":"md5"},{"auth-local":"trust"},"data-checksums",{"encoding":"UTF8"},{"locale":"en_US.UTF-8"}],"users":{"test":{"password":"","options":["CREATEDB","NOLOGIN"]}},"dcs":{"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432,"postgresql":{"parameters":{"max_connections":"100","max_locks_per_transaction":"64","max_worker_processes":"4"}}}}}`,
},
},
ExpectedResult: true,
},
{
Envs: []v1.EnvVar{
{
Name: "VARIABLE1",
Value: "value1",
},
{
Name: "VARIABLE2",
Value: "value2",
},
{
Name: "VARIABLE3",
Value: "value3",
},
{
Name: "SPILO_CONFIGURATION",
Value: `{"postgresql":{"bin_dir":"/usr/lib/postgresql/12/bin","parameters":{"autovacuum_analyze_scale_factor":"0.1"},"pg_hba":["hostssl all all 0.0.0.0/0 md5","host all all 0.0.0.0/0 md5"]},"bootstrap":{"initdb":[{"auth-host":"md5"},{"auth-local":"trust"},"data-checksums",{"encoding":"UTF8"},{"locale":"en_US.UTF-8"}],"users":{"test":{"password":"","options":["CREATEDB","NOLOGIN"]}},"dcs":{"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432,"postgresql":{"parameters":{"max_locks_per_transaction":"64","max_worker_processes":"4"}}}}}`,
},
},
ExpectedResult: true,
},
{
Envs: []v1.EnvVar{
{
Name: "VARIABLE4",
Value: "value4",
},
{
Name: "VARIABLE2",
Value: "value2",
},
{
Name: "VARIABLE3",
Value: "value3",
},
{
Name: "SPILO_CONFIGURATION",
Value: `{"postgresql":{"bin_dir":"/usr/lib/postgresql/12/bin","parameters":{"autovacuum_analyze_scale_factor":"0.1"},"pg_hba":["hostssl all all 0.0.0.0/0 md5","host all all 0.0.0.0/0 md5"]},"bootstrap":{"initdb":[{"auth-host":"md5"},{"auth-local":"trust"},"data-checksums",{"encoding":"UTF8"},{"locale":"en_US.UTF-8"}],"users":{"test":{"password":"","options":["CREATEDB","NOLOGIN"]}},"dcs":{"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432,"postgresql":{"parameters":{"max_locks_per_transaction":"64","max_worker_processes":"4"}}}}}`,
},
},
ExpectedResult: false,
},
{
Envs: []v1.EnvVar{
{
Name: "VARIABLE1",
Value: "value1",
},
{
Name: "VARIABLE2",
Value: "value2",
},
{
Name: "VARIABLE3",
Value: "value3",
},
{
Name: "VARIABLE4",
Value: "value4",
},
{
Name: "SPILO_CONFIGURATION",
Value: `{"postgresql":{"bin_dir":"/usr/lib/postgresql/12/bin","parameters":{"autovacuum_analyze_scale_factor":"0.1"},"pg_hba":["hostssl all all 0.0.0.0/0 md5","host all all 0.0.0.0/0 md5"]},"bootstrap":{"initdb":[{"auth-host":"md5"},{"auth-local":"trust"},"data-checksums",{"encoding":"UTF8"},{"locale":"en_US.UTF-8"}],"users":{"test":{"password":"","options":["CREATEDB","NOLOGIN"]}},"dcs":{"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432,"postgresql":{"parameters":{"max_connections":"100","max_locks_per_transaction":"64","max_worker_processes":"4"}}}}}`,
},
},
ExpectedResult: false,
},
{
Envs: []v1.EnvVar{
{
Name: "VARIABLE1",
Value: "value1",
},
{
Name: "VARIABLE2",
Value: "value2",
},
{
Name: "SPILO_CONFIGURATION",
Value: `{"postgresql":{"bin_dir":"/usr/lib/postgresql/12/bin","parameters":{"autovacuum_analyze_scale_factor":"0.1"},"pg_hba":["hostssl all all 0.0.0.0/0 md5","host all all 0.0.0.0/0 md5"]},"bootstrap":{"initdb":[{"auth-host":"md5"},{"auth-local":"trust"},"data-checksums",{"encoding":"UTF8"},{"locale":"en_US.UTF-8"}],"users":{"test":{"password":"","options":["CREATEDB","NOLOGIN"]}},"dcs":{"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432,"postgresql":{"parameters":{"max_connections":"100","max_locks_per_transaction":"64","max_worker_processes":"4"}}}}}`,
},
},
ExpectedResult: false,
},
}
refCase := testCases[0]
for _, testCase := range testCases {
if result := compareEnv(refCase.Envs, testCase.Envs); result != testCase.ExpectedResult {
t.Errorf("expected %v got %v", testCase.ExpectedResult, result)
}
}
}
func TestCrossNamespacedSecrets(t *testing.T) {
testName := "test secrets in different namespace"
clientSet := fake.NewSimpleClientset()
acidClientSet := fakeacidv1.NewSimpleClientset()
namespace := "default"
client := k8sutil.KubernetesClient{
StatefulSetsGetter: clientSet.AppsV1(),
ServicesGetter: clientSet.CoreV1(),
DeploymentsGetter: clientSet.AppsV1(),
PostgresqlsGetter: acidClientSet.AcidV1(),
SecretsGetter: clientSet.CoreV1(),
}
pg := acidv1.Postgresql{
ObjectMeta: metav1.ObjectMeta{
Name: "acid-fake-cluster",
Namespace: namespace,
},
Spec: acidv1.PostgresSpec{
Volume: acidv1.Volume{
Size: "1Gi",
},
EnableNamespacedSecret: boolToPointer(true),
Users: map[string]acidv1.UserFlags{
"appspace.db_user": {},
"db_user": {},
},
},
}
var cluster = New(
Config{
OpConfig: config.Config{
ConnectionPooler: config.ConnectionPooler{
ConnectionPoolerDefaultCPURequest: "100m",
ConnectionPoolerDefaultCPULimit: "100m",
ConnectionPoolerDefaultMemoryRequest: "100Mi",
ConnectionPoolerDefaultMemoryLimit: "100Mi",
NumberOfInstances: int32ToPointer(1),
},
PodManagementPolicy: "ordered_ready",
Resources: config.Resources{
ClusterLabels: map[string]string{"application": "spilo"},
ClusterNameLabel: "cluster-name",
DefaultCPURequest: "300m",
DefaultCPULimit: "300m",
DefaultMemoryRequest: "300Mi",
DefaultMemoryLimit: "300Mi",
PodRoleLabel: "spilo-role",
},
},
}, client, pg, logger, eventRecorder)
userNamespaceMap := map[string]string{
cluster.Namespace: "db_user",
"appspace": "appspace.db_user",
}
err := cluster.initRobotUsers()
if err != nil {
t.Errorf("Could not create secret for namespaced users with error: %s", err)
}
for _, u := range cluster.pgUsers {
if u.Name != userNamespaceMap[u.Namespace] {
t.Errorf("%s: Could not create namespaced user in its correct namespaces for user %s in namespace %s", testName, u.Name, u.Namespace)
}
}
}
func TestValidUsernames(t *testing.T) {
testName := "test username validity"
invalidUsernames := []string{"_", ".", ".user", "appspace.", "user_", "_user", "-user", "user-", ",", "-", ",user", "user,", "namespace,user"}
validUsernames := []string{"user", "appspace.user", "appspace.dot.user", "user_name", "app_space.user_name"}
for _, username := range invalidUsernames {
if isValidUsername(username) {
t.Errorf("%s Invalid username is allowed: %s", testName, username)
}
}
for _, username := range validUsernames {
if !isValidUsername(username) {
t.Errorf("%s Valid username is not allowed: %s", testName, username)
}
}
}
+30 -9
View File
@@ -412,13 +412,33 @@ func tolerations(tolerationsSpec *[]v1.Toleration, podToleration map[string]stri
// Those parameters must go to the bootstrap/dcs/postgresql/parameters section.
// See http://patroni.readthedocs.io/en/latest/dynamic_configuration.html.
func isBootstrapOnlyParameter(param string) bool {
return param == "max_connections" ||
param == "max_locks_per_transaction" ||
param == "max_worker_processes" ||
param == "max_prepared_transactions" ||
param == "wal_level" ||
param == "wal_log_hints" ||
param == "track_commit_timestamp"
params := map[string]bool{
"archive_command": false,
"shared_buffers": false,
"logging_collector": false,
"log_destination": false,
"log_directory": false,
"log_filename": false,
"log_file_mode": false,
"log_rotation_age": false,
"log_truncate_on_rotation": false,
"ssl": false,
"ssl_ca_file": false,
"ssl_crl_file": false,
"ssl_cert_file": false,
"ssl_key_file": false,
"shared_preload_libraries": false,
"bg_mon.listen_address": false,
"bg_mon.history_buckets": false,
"pg_stat_statements.track_utility": false,
"extwlist.extensions": false,
"extwlist.custom_path": false,
}
result, ok := params[param]
if !ok {
result = true
}
return result
}
func generateVolumeMounts(volume acidv1.Volume) []v1.VolumeMount {
@@ -1547,10 +1567,11 @@ func (c *Cluster) generateUserSecrets() map[string]*v1.Secret {
namespace := c.Namespace
for username, pgUser := range c.pgUsers {
//Skip users with no password i.e. human users (they'll be authenticated using pam)
secret := c.generateSingleUserSecret(namespace, pgUser)
secret := c.generateSingleUserSecret(pgUser.Namespace, pgUser)
if secret != nil {
secrets[username] = secret
}
namespace = pgUser.Namespace
}
/* special case for the system user */
for _, systemUser := range c.systemUsers {
@@ -1590,7 +1611,7 @@ func (c *Cluster) generateSingleUserSecret(namespace string, pgUser spec.PgUser)
secret := v1.Secret{
ObjectMeta: metav1.ObjectMeta{
Name: c.credentialSecretName(username),
Namespace: namespace,
Namespace: pgUser.Namespace,
Labels: lbls,
Annotations: c.annotationsSet(nil),
},
+6
View File
@@ -1207,6 +1207,12 @@ func TestSidecars(t *testing.T) {
}
spec = acidv1.PostgresSpec{
PostgresqlParam: acidv1.PostgresqlParam{
PgVersion: "12.1",
Parameters: map[string]string{
"max_connections": "100",
},
},
TeamID: "myapp", NumberOfInstances: 1,
Resources: acidv1.Resources{
ResourceRequests: acidv1.ResourceDescription{CPU: "1", Memory: "10"},
+13 -2
View File
@@ -304,8 +304,19 @@ func (c *Cluster) MigrateMasterPod(podName spec.NamespacedName) error {
}
masterCandidateName := util.NameFromMeta(masterCandidatePod.ObjectMeta)
if err := c.Switchover(oldMaster, masterCandidateName); err != nil {
return fmt.Errorf("could not failover to pod %q: %v", masterCandidateName, err)
err = retryutil.Retry(1*time.Minute, 5*time.Minute,
func() (bool, error) {
err := c.Switchover(oldMaster, masterCandidateName)
if err != nil {
c.logger.Errorf("could not failover to pod %q: %v", masterCandidateName, err)
return false, nil
}
return true, nil
},
)
if err != nil {
return fmt.Errorf("could not migrate master pod: %v", err)
}
return nil
+1 -1
View File
@@ -32,7 +32,7 @@ func (c *Cluster) listResources() error {
}
for _, obj := range c.Secrets {
c.logger.Infof("found secret: %q (uid: %q)", util.NameFromMeta(obj.ObjectMeta), obj.UID)
c.logger.Infof("found secret: %q (uid: %q) namesapce: %s", util.NameFromMeta(obj.ObjectMeta), obj.UID, obj.ObjectMeta.Namespace)
}
for role, endpoint := range c.Endpoints {
+86 -14
View File
@@ -5,6 +5,7 @@ import (
"fmt"
"regexp"
"strings"
"time"
acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
"github.com/zalando/postgres-operator/pkg/spec"
@@ -260,6 +261,7 @@ func (c *Cluster) syncPodDisruptionBudget(isUpdate bool) error {
}
func (c *Cluster) syncStatefulSet() error {
var instancesRestartRequired bool
podsToRecreate := make([]v1.Pod, 0)
switchoverCandidates := make([]spec.NamespacedName, 0)
@@ -379,10 +381,21 @@ func (c *Cluster) syncStatefulSet() error {
// Apply special PostgreSQL parameters that can only be set via the Patroni API.
// it is important to do it after the statefulset pods are there, but before the rolling update
// since those parameters require PostgreSQL restart.
if err := c.checkAndSetGlobalPostgreSQLConfiguration(); err != nil {
instancesRestartRequired, err = c.checkAndSetGlobalPostgreSQLConfiguration()
if err != nil {
return fmt.Errorf("could not set cluster-wide PostgreSQL configuration options: %v", err)
}
if instancesRestartRequired {
c.logger.Debugln("restarting Postgres server within pods")
c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", "restarting Postgres server within pods")
if err := c.restartInstances(); err != nil {
c.logger.Warningf("could not restart Postgres server within pods: %v", err)
}
c.logger.Infof("Postgres server successfuly restarted on all pods")
c.eventRecorder.Event(c.GetReference(), v1.EventTypeNormal, "Update", "Postgres server restart done - all instances have been restarted")
}
// if we get here we also need to re-create the pods (either leftovers from the old
// statefulset or those that got their configuration from the outdated statefulset)
if len(podsToRecreate) > 0 {
@@ -396,6 +409,57 @@ func (c *Cluster) syncStatefulSet() error {
return nil
}
func (c *Cluster) restartInstances() error {
c.setProcessName("starting to restart Postgres servers")
ls := c.labelsSet(false)
namespace := c.Namespace
listOptions := metav1.ListOptions{
LabelSelector: ls.String(),
}
pods, err := c.KubeClient.Pods(namespace).List(context.TODO(), listOptions)
if err != nil {
return fmt.Errorf("could not get the list of pods: %v", err)
}
c.logger.Infof("there are %d pods in the cluster which resquire Postgres server restart", len(pods.Items))
var (
masterPod *v1.Pod
)
for i, pod := range pods.Items {
role := PostgresRole(pod.Labels[c.OpConfig.PodRoleLabel])
if role == Master {
masterPod = &pods.Items[i]
continue
}
podName := util.NameFromMeta(pods.Items[i].ObjectMeta)
config, err := c.patroni.GetConfig(&pod)
if err != nil {
return fmt.Errorf("could not get config for pod %s: %v", podName, err)
}
ttl, ok := config["ttl"].(int32)
if !ok {
ttl = 30
}
if err = c.patroni.Restart(&pod); err != nil {
return fmt.Errorf("could not restart Postgres server on pod %s: %v", podName, err)
}
time.Sleep(time.Duration(ttl) * time.Second)
}
if masterPod != nil {
podName := util.NameFromMeta(masterPod.ObjectMeta)
if err = c.patroni.Restart(masterPod); err != nil {
return fmt.Errorf("could not restart postgres server on masterPod %s: %v", podName, err)
}
}
return nil
}
// AnnotationsToPropagate get the annotations to update if required
// based on the annotations in postgres CRD
func (c *Cluster) AnnotationsToPropagate(annotations map[string]string) map[string]string {
@@ -430,10 +494,11 @@ func (c *Cluster) AnnotationsToPropagate(annotations map[string]string) map[stri
// checkAndSetGlobalPostgreSQLConfiguration checks whether cluster-wide API parameters
// (like max_connections) has changed and if necessary sets it via the Patroni API
func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration() error {
func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration() (bool, error) {
var (
err error
pods []v1.Pod
err error
pods []v1.Pod
restartRequired bool
)
// we need to extract those options from the cluster manifest.
@@ -447,14 +512,14 @@ func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration() error {
}
if len(optionsToSet) == 0 {
return nil
return restartRequired, nil
}
if pods, err = c.listPods(); err != nil {
return err
return restartRequired, err
}
if len(pods) == 0 {
return fmt.Errorf("could not call Patroni API: cluster has no pods")
return restartRequired, fmt.Errorf("could not call Patroni API: cluster has no pods")
}
// try all pods until the first one that is successful, as it doesn't matter which pod
// carries the request to change configuration through
@@ -463,11 +528,12 @@ func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration() error {
c.logger.Debugf("calling Patroni API on a pod %s to set the following Postgres options: %v",
podName, optionsToSet)
if err = c.patroni.SetPostgresParameters(&pod, optionsToSet); err == nil {
return nil
restartRequired = true
return restartRequired, nil
}
c.logger.Warningf("could not patch postgres parameters with a pod %s: %v", podName, err)
}
return fmt.Errorf("could not reach Patroni API to set Postgres options: failed on every pod (%d total)",
return restartRequired, fmt.Errorf("could not reach Patroni API to set Postgres options: failed on every pod (%d total)",
len(pods))
}
@@ -483,7 +549,7 @@ func (c *Cluster) syncSecrets() error {
for secretUsername, secretSpec := range secrets {
if secret, err = c.KubeClient.Secrets(secretSpec.Namespace).Create(context.TODO(), secretSpec, metav1.CreateOptions{}); err == nil {
c.Secrets[secret.UID] = secret
c.logger.Debugf("created new secret %s, uid: %s", util.NameFromMeta(secret.ObjectMeta), secret.UID)
c.logger.Debugf("created new secret %s, namespace: %s, uid: %s", util.NameFromMeta(secret.ObjectMeta), secretSpec.Namespace, secret.UID)
continue
}
if k8sutil.ResourceAlreadyExists(err) {
@@ -521,7 +587,7 @@ func (c *Cluster) syncSecrets() error {
userMap[secretUsername] = pwdUser
}
} else {
return fmt.Errorf("could not create secret for user %s: %v", secretUsername, err)
return fmt.Errorf("could not create secret for user %s: in namespace %s: %v", secretUsername, secretSpec.Namespace, err)
}
}
@@ -556,11 +622,17 @@ func (c *Cluster) syncRoles() (err error) {
// create list of database roles to query
for _, u := range c.pgUsers {
userNames = append(userNames, u.Name)
pgRole := u.Name
if u.Namespace != c.Namespace && u.Namespace != "" {
// to avoid the conflict of having multiple users of same name
// but each in different namespace.
pgRole = fmt.Sprintf("%s.%s", u.Name, u.Namespace)
}
userNames = append(userNames, pgRole)
// add team member role name with rename suffix in case we need to rename it back
if u.Origin == spec.RoleOriginTeamsAPI && c.OpConfig.EnableTeamMemberDeprecation {
deletedUsers[u.Name+c.OpConfig.RoleDeletionSuffix] = u.Name
userNames = append(userNames, u.Name+c.OpConfig.RoleDeletionSuffix)
deletedUsers[pgRole+c.OpConfig.RoleDeletionSuffix] = pgRole
userNames = append(userNames, pgRole+c.OpConfig.RoleDeletionSuffix)
}
}