public_git/postgres-operator
1
0
Fork 0
mirror of https://github.com/zalando/postgres-operator.git synced 2026-10-06 03:40:30 +02:00
Code Issues Projects Releases Wiki Activity

Merge branch 'master' into fes-support

Browse Source
This commit is contained in:
Felix Kunde
2021-09-16 13:27:57 +02:00
parent d2addd3b17 c8dc7ca871
commit 79bd69d3ab
44 changed files with 745 additions and 215 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/apis/acid.zalan.do/v1/crds.go
+51
View File
@@ -573,6 +573,9 @@ var PostgresCRDResourceValidation = apiextv1.CustomResourceValidation{
},
},
},
"secretNamespace": {
Type: "string",
},
},
},
},
@@ -894,6 +897,54 @@ var PostgresCRDResourceValidation = apiextv1.CustomResourceValidation{
"iops": {
Type: "integer",
},
"selector": {
Type: "object",
Properties: map[string]apiextv1.JSONSchemaProps{
"matchExpressions": {
Type: "array",
Items: &apiextv1.JSONSchemaPropsOrArray{
Schema: &apiextv1.JSONSchemaProps{
Type: "object",
Required: []string{"key", "operator", "values"},
Properties: map[string]apiextv1.JSONSchemaProps{
"key": {
Type: "string",
},
"operator": {
Type: "string",
Enum: []apiextv1.JSON{
{
Raw: []byte(`"In"`),
},
{
Raw: []byte(`"NotIn"`),
},
{
Raw: []byte(`"Exists"`),
},
{
Raw: []byte(`"DoesNotExist"`),
},
},
},
"values": {
Type: "array",
Items: &apiextv1.JSONSchemaPropsOrArray{
Schema: &apiextv1.JSONSchemaProps{
Type: "string",
},
},
},
},
},
},
},
"matchLabels": {
Type: "object",
XPreserveUnknownFields: util.True(),
},
},
},
"size": {
Type: "string",
Description: "Value must not be zero",
pkg/apis/acid.zalan.do/v1/operator_configuration_type.go
+1
View File
@@ -132,6 +132,7 @@ type AWSGCPConfiguration struct {
AWSRegion string `json:"aws_region,omitempty"`
WALGSBucket string `json:"wal_gs_bucket,omitempty"`
GCPCredentials string `json:"gcp_credentials,omitempty"`
WALAZStorageAccount string `json:"wal_az_storage_account,omitempty"`
LogS3Bucket string `json:"log_s3_bucket,omitempty"`
KubeIAMRole string `json:"kube_iam_role,omitempty"`
AdditionalSecretMount string `json:"additional_secret_mount,omitempty"`
pkg/apis/acid.zalan.do/v1/postgresql_type.go
+8 -6
View File
@@ -96,6 +96,7 @@ type PreparedDatabase struct {
PreparedSchemas map[string]PreparedSchema `json:"schemas,omitempty"`
DefaultUsers bool `json:"defaultUsers,omitempty" defaults:"false"`
Extensions map[string]string `json:"extensions,omitempty"`
SecretNamespace string `json:"secretNamespace,omitempty"`
}
// PreparedSchema describes elements to be bootstrapped per schema
@@ -114,12 +115,13 @@ type MaintenanceWindow struct {
// Volume describes a single volume in the manifest.
type Volume struct {
Size string `json:"size"`
StorageClass string `json:"storageClass,omitempty"`
SubPath string `json:"subPath,omitempty"`
Iops *int64 `json:"iops,omitempty"`
Throughput *int64 `json:"throughput,omitempty"`
VolumeType string `json:"type,omitempty"`
Selector *metav1.LabelSelector `json:"selector,omitempty"`
Size string `json:"size"`
StorageClass string `json:"storageClass,omitempty"`
SubPath string `json:"subPath,omitempty"`
Iops *int64 `json:"iops,omitempty"`
Throughput *int64 `json:"throughput,omitempty"`
VolumeType string `json:"type,omitempty"`
}
// AdditionalVolume specs additional optional volumes for statefulset
pkg/apis/acid.zalan.do/v1/zz_generated.deepcopy.go
+6 -17
View File
@@ -29,6 +29,7 @@ package v1
import (
config "github.com/zalando/postgres-operator/pkg/util/config"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
runtime "k8s.io/apimachinery/pkg/runtime"
)
@@ -314,22 +315,6 @@ func (in *MaintenanceWindow) DeepCopy() *MaintenanceWindow {
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *MajorVersionUpgradeConfiguration) DeepCopyInto(out *MajorVersionUpgradeConfiguration) {
*out = *in
return
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new MajorVersionUpgradeConfiguration.
func (in *MajorVersionUpgradeConfiguration) DeepCopy() *MajorVersionUpgradeConfiguration {
if in == nil {
return nil
}
out := new(MajorVersionUpgradeConfiguration)
in.DeepCopyInto(out)
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *OperatorConfiguration) DeepCopyInto(out *OperatorConfiguration) {
*out = *in
@@ -385,7 +370,6 @@ func (in *OperatorConfigurationData) DeepCopyInto(out *OperatorConfigurationData
}
}
out.PostgresUsersConfiguration = in.PostgresUsersConfiguration
out.MajorVersionUpgrade = in.MajorVersionUpgrade
in.Kubernetes.DeepCopyInto(&out.Kubernetes)
out.PostgresPodResources = in.PostgresPodResources
out.Timeouts = in.Timeouts
@@ -1234,6 +1218,11 @@ func (in UserFlags) DeepCopy() UserFlags {
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *Volume) DeepCopyInto(out *Volume) {
*out = *in
if in.Selector != nil {
in, out := &in.Selector, &out.Selector
*out = new(metav1.LabelSelector)
(*in).DeepCopyInto(*out)
}
if in.Iops != nil {
in, out := &in.Iops, &out.Iops
*out = new(int64)
pkg/cluster/cluster.go
+21 -12
View File
@@ -1019,9 +1019,9 @@ func (c *Cluster) initSystemUsers() {
// Connection pooler user is an exception, if requested it's going to be
// created by operator as a normal pgUser
if needConnectionPooler(&c.Spec) {
// initialize empty connection pooler if not done yet
if c.Spec.ConnectionPooler == nil {
c.Spec.ConnectionPooler = &acidv1.ConnectionPooler{}
connectionPoolerSpec := c.Spec.ConnectionPooler
if connectionPoolerSpec == nil {
connectionPoolerSpec = &acidv1.ConnectionPooler{}
}
// Using superuser as pooler user is not a good idea. First of all it's
@@ -1029,13 +1029,13 @@ func (c *Cluster) initSystemUsers() {
// and second it's a bad practice.
username := c.OpConfig.ConnectionPooler.User
isSuperUser := c.Spec.ConnectionPooler.User == c.OpConfig.SuperUsername
isSuperUser := connectionPoolerSpec.User == c.OpConfig.SuperUsername
isProtectedUser := c.shouldAvoidProtectedOrSystemRole(
c.Spec.ConnectionPooler.User, "connection pool role")
connectionPoolerSpec.User, "connection pool role")
if !isSuperUser && !isProtectedUser {
username = util.Coalesce(
c.Spec.ConnectionPooler.User,
connectionPoolerSpec.User,
c.OpConfig.ConnectionPooler.User)
}
@@ -1107,11 +1107,11 @@ func (c *Cluster) initPreparedDatabaseRoles() error {
}
// default roles per database
if err := c.initDefaultRoles(defaultRoles, "admin", preparedDbName, searchPath.String()); err != nil {
if err := c.initDefaultRoles(defaultRoles, "admin", preparedDbName, searchPath.String(), preparedDB.SecretNamespace); err != nil {
return fmt.Errorf("could not initialize default roles for database %s: %v", preparedDbName, err)
}
if preparedDB.DefaultUsers {
if err := c.initDefaultRoles(defaultUsers, "admin", preparedDbName, searchPath.String()); err != nil {
if err := c.initDefaultRoles(defaultUsers, "admin", preparedDbName, searchPath.String(), preparedDB.SecretNamespace); err != nil {
return fmt.Errorf("could not initialize default roles for database %s: %v", preparedDbName, err)
}
}
@@ -1122,14 +1122,14 @@ func (c *Cluster) initPreparedDatabaseRoles() error {
if err := c.initDefaultRoles(defaultRoles,
preparedDbName+constants.OwnerRoleNameSuffix,
preparedDbName+"_"+preparedSchemaName,
constants.DefaultSearchPath+", "+preparedSchemaName); err != nil {
constants.DefaultSearchPath+", "+preparedSchemaName, preparedDB.SecretNamespace); err != nil {
return fmt.Errorf("could not initialize default roles for database schema %s: %v", preparedSchemaName, err)
}
if preparedSchema.DefaultUsers {
if err := c.initDefaultRoles(defaultUsers,
preparedDbName+constants.OwnerRoleNameSuffix,
preparedDbName+"_"+preparedSchemaName,
constants.DefaultSearchPath+", "+preparedSchemaName); err != nil {
constants.DefaultSearchPath+", "+preparedSchemaName, preparedDB.SecretNamespace); err != nil {
return fmt.Errorf("could not initialize default users for database schema %s: %v", preparedSchemaName, err)
}
}
@@ -1139,10 +1139,19 @@ func (c *Cluster) initPreparedDatabaseRoles() error {
return nil
}
func (c *Cluster) initDefaultRoles(defaultRoles map[string]string, admin, prefix string, searchPath string) error {
func (c *Cluster) initDefaultRoles(defaultRoles map[string]string, admin, prefix, searchPath, secretNamespace string) error {
for defaultRole, inherits := range defaultRoles {
namespace := c.Namespace
//if namespaced secrets are allowed
if secretNamespace != "" {
if c.Config.OpConfig.EnableCrossNamespaceSecret {
namespace = secretNamespace
} else {
c.logger.Warn("secretNamespace ignored because enable_cross_namespace_secret set to false. Creating secrets in cluster namespace.")
}
}
roleName := prefix + defaultRole
flags := []string{constants.RoleFlagNoLogin}
@@ -1165,7 +1174,7 @@ func (c *Cluster) initDefaultRoles(defaultRoles map[string]string, admin, prefix
newRole := spec.PgUser{
Origin: spec.RoleOriginBootstrap,
Name: roleName,
Namespace: c.Namespace,
Namespace: namespace,
Password: util.RandomPassword(constants.PasswordLength),
Flags: flags,
MemberOf: memberOf,
pkg/cluster/connection_pooler.go
+59 -68
View File
@@ -3,6 +3,7 @@ package cluster
import (
"context"
"fmt"
"reflect"
"strings"
"github.com/r3labs/diff"
@@ -60,7 +61,7 @@ func needMasterConnectionPooler(spec *acidv1.PostgresSpec) bool {
}
func needMasterConnectionPoolerWorker(spec *acidv1.PostgresSpec) bool {
return (nil != spec.EnableConnectionPooler && *spec.EnableConnectionPooler) ||
return (spec.EnableConnectionPooler != nil && *spec.EnableConnectionPooler) ||
(spec.ConnectionPooler != nil && spec.EnableConnectionPooler == nil)
}
@@ -114,7 +115,7 @@ func (c *Cluster) createConnectionPooler(LookupFunction InstallFunction) (SyncRe
c.setProcessName("creating connection pooler")
//this is essentially sync with nil as oldSpec
if reason, err := c.syncConnectionPooler(nil, &c.Postgresql, LookupFunction); err != nil {
if reason, err := c.syncConnectionPooler(&acidv1.Postgresql{}, &c.Postgresql, LookupFunction); err != nil {
return reason, err
}
return reason, nil
@@ -140,11 +141,15 @@ func (c *Cluster) createConnectionPooler(LookupFunction InstallFunction) (SyncRe
// RESERVE_SIZE is how many additional connections to allow for a pooler.
func (c *Cluster) getConnectionPoolerEnvVars() []v1.EnvVar {
spec := &c.Spec
connectionPoolerSpec := spec.ConnectionPooler
if connectionPoolerSpec == nil {
connectionPoolerSpec = &acidv1.ConnectionPooler{}
}
effectiveMode := util.Coalesce(
spec.ConnectionPooler.Mode,
connectionPoolerSpec.Mode,
c.OpConfig.ConnectionPooler.Mode)
numberOfInstances := spec.ConnectionPooler.NumberOfInstances
numberOfInstances := connectionPoolerSpec.NumberOfInstances
if numberOfInstances == nil {
numberOfInstances = util.CoalesceInt32(
c.OpConfig.ConnectionPooler.NumberOfInstances,
@@ -152,7 +157,7 @@ func (c *Cluster) getConnectionPoolerEnvVars() []v1.EnvVar {
}
effectiveMaxDBConn := util.CoalesceInt32(
spec.ConnectionPooler.MaxDBConnections,
connectionPoolerSpec.MaxDBConnections,
c.OpConfig.ConnectionPooler.MaxDBConnections)
if effectiveMaxDBConn == nil {
@@ -201,17 +206,21 @@ func (c *Cluster) getConnectionPoolerEnvVars() []v1.EnvVar {
func (c *Cluster) generateConnectionPoolerPodTemplate(role PostgresRole) (
*v1.PodTemplateSpec, error) {
spec := &c.Spec
connectionPoolerSpec := spec.ConnectionPooler
if connectionPoolerSpec == nil {
connectionPoolerSpec = &acidv1.ConnectionPooler{}
}
gracePeriod := int64(c.OpConfig.PodTerminateGracePeriod.Seconds())
resources, err := generateResourceRequirements(
spec.ConnectionPooler.Resources,
connectionPoolerSpec.Resources,
makeDefaultConnectionPoolerResources(&c.OpConfig))
effectiveDockerImage := util.Coalesce(
spec.ConnectionPooler.DockerImage,
connectionPoolerSpec.DockerImage,
c.OpConfig.ConnectionPooler.Image)
effectiveSchema := util.Coalesce(
spec.ConnectionPooler.Schema,
connectionPoolerSpec.Schema,
c.OpConfig.ConnectionPooler.Schema)
if err != nil {
@@ -220,7 +229,7 @@ func (c *Cluster) generateConnectionPoolerPodTemplate(role PostgresRole) (
secretSelector := func(key string) *v1.SecretKeySelector {
effectiveUser := util.Coalesce(
spec.ConnectionPooler.User,
connectionPoolerSpec.User,
c.OpConfig.ConnectionPooler.User)
return &v1.SecretKeySelector{
@@ -285,6 +294,8 @@ func (c *Cluster) generateConnectionPoolerPodTemplate(role PostgresRole) (
},
}
tolerationsSpec := tolerations(&spec.Tolerations, c.OpConfig.PodToleration)
podTemplate := &v1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: c.connectionPoolerLabels(role, true).MatchLabels,
@@ -294,12 +305,18 @@ func (c *Cluster) generateConnectionPoolerPodTemplate(role PostgresRole) (
Spec: v1.PodSpec{
TerminationGracePeriodSeconds: &gracePeriod,
Containers: []v1.Container{poolerContainer},
// TODO: add tolerations to scheduler pooler on the same node
// as database
//Tolerations: *tolerationsSpec,
Tolerations: tolerationsSpec,
},
}
nodeAffinity := nodeAffinity(c.OpConfig.NodeReadinessLabel, spec.NodeAffinity)
if c.OpConfig.EnablePodAntiAffinity {
labelsSet := labels.Set(c.connectionPoolerLabels(role, false).MatchLabels)
podTemplate.Spec.Affinity = generatePodAffinity(labelsSet, c.OpConfig.PodAntiAffinityTopologyKey, nodeAffinity)
} else if nodeAffinity != nil {
podTemplate.Spec.Affinity = nodeAffinity
}
return podTemplate, nil
}
@@ -313,12 +330,13 @@ func (c *Cluster) generateConnectionPoolerDeployment(connectionPooler *Connectio
// default values, initialize it to an empty structure. It could be done
// anywhere, but here is the earliest common entry point between sync and
// create code, so init here.
if spec.ConnectionPooler == nil {
spec.ConnectionPooler = &acidv1.ConnectionPooler{}
connectionPoolerSpec := spec.ConnectionPooler
if connectionPoolerSpec == nil {
connectionPoolerSpec = &acidv1.ConnectionPooler{}
}
podTemplate, err := c.generateConnectionPoolerPodTemplate(connectionPooler.Role)
numberOfInstances := spec.ConnectionPooler.NumberOfInstances
numberOfInstances := connectionPoolerSpec.NumberOfInstances
if numberOfInstances == nil {
numberOfInstances = util.CoalesceInt32(
c.OpConfig.ConnectionPooler.NumberOfInstances,
@@ -363,16 +381,6 @@ func (c *Cluster) generateConnectionPoolerDeployment(connectionPooler *Connectio
func (c *Cluster) generateConnectionPoolerService(connectionPooler *ConnectionPoolerObjects) *v1.Service {
spec := &c.Spec
// there are two ways to enable connection pooler, either to specify a
// connectionPooler section or enableConnectionPooler. In the second case
// spec.connectionPooler will be nil, so to make it easier to calculate
// default values, initialize it to an empty structure. It could be done
// anywhere, but here is the earliest common entry point between sync and
// create code, so init here.
if spec.ConnectionPooler == nil {
spec.ConnectionPooler = &acidv1.ConnectionPooler{}
}
serviceSpec := v1.ServiceSpec{
Ports: []v1.ServicePort{
{
@@ -660,12 +668,14 @@ func makeDefaultConnectionPoolerResources(config *config.Config) acidv1.Resource
func logPoolerEssentials(log *logrus.Entry, oldSpec, newSpec *acidv1.Postgresql) {
var v []string
var input []*bool
newMasterConnectionPoolerEnabled := needMasterConnectionPoolerWorker(&newSpec.Spec)
if oldSpec == nil {
input = []*bool{nil, nil, newSpec.Spec.EnableConnectionPooler, newSpec.Spec.EnableReplicaConnectionPooler}
input = []*bool{nil, nil, &newMasterConnectionPoolerEnabled, newSpec.Spec.EnableReplicaConnectionPooler}
} else {
input = []*bool{oldSpec.Spec.EnableConnectionPooler, oldSpec.Spec.EnableReplicaConnectionPooler, newSpec.Spec.EnableConnectionPooler, newSpec.Spec.EnableReplicaConnectionPooler}
oldMasterConnectionPoolerEnabled := needMasterConnectionPoolerWorker(&oldSpec.Spec)
input = []*bool{&oldMasterConnectionPoolerEnabled, oldSpec.Spec.EnableReplicaConnectionPooler, &newMasterConnectionPoolerEnabled, newSpec.Spec.EnableReplicaConnectionPooler}
}
for _, b := range input {
@@ -676,25 +686,16 @@ func logPoolerEssentials(log *logrus.Entry, oldSpec, newSpec *acidv1.Postgresql)
}
}
log.Debugf("syncing connection pooler from (%v, %v) to (%v, %v)", v[0], v[1], v[2], v[3])
log.Debugf("syncing connection pooler (master, replica) from (%v, %v) to (%v, %v)", v[0], v[1], v[2], v[3])
}
func (c *Cluster) syncConnectionPooler(oldSpec, newSpec *acidv1.Postgresql, LookupFunction InstallFunction) (SyncReason, error) {
var reason SyncReason
var err error
var newNeedConnectionPooler, oldNeedConnectionPooler bool
oldNeedConnectionPooler = false
var connectionPoolerNeeded bool
if oldSpec == nil {
oldSpec = &acidv1.Postgresql{
Spec: acidv1.PostgresSpec{
ConnectionPooler: &acidv1.ConnectionPooler{},
},
}
}
needSync, _ := needSyncConnectionPoolerSpecs(oldSpec.Spec.ConnectionPooler, newSpec.Spec.ConnectionPooler, c.logger)
needSync := !reflect.DeepEqual(oldSpec.Spec.ConnectionPooler, newSpec.Spec.ConnectionPooler)
masterChanges, err := diff.Diff(oldSpec.Spec.EnableConnectionPooler, newSpec.Spec.EnableConnectionPooler)
if err != nil {
c.logger.Error("Error in getting diff of master connection pooler changes")
@@ -704,15 +705,14 @@ func (c *Cluster) syncConnectionPooler(oldSpec, newSpec *acidv1.Postgresql, Look
c.logger.Error("Error in getting diff of replica connection pooler changes")
}
// skip pooler sync only
// 1. if there is no diff in spec, AND
// 2. if connection pooler is already there and is also required as per newSpec
//
// Handling the case when connectionPooler is not there but it is required
// skip pooler sync when theres no diff or it's deactivated
// but, handling the case when connectionPooler is not there but it is required
// as per spec, hence do not skip syncing in that case, even though there
// is no diff in specs
if (!needSync && len(masterChanges) <= 0 && len(replicaChanges) <= 0) &&
(c.ConnectionPooler != nil && (needConnectionPooler(&newSpec.Spec))) {
((!needConnectionPooler(&newSpec.Spec) && (c.ConnectionPooler == nil || !needConnectionPooler(&oldSpec.Spec))) ||
(c.ConnectionPooler != nil && needConnectionPooler(&newSpec.Spec) &&
(c.ConnectionPooler[Master].LookupFunction || c.ConnectionPooler[Replica].LookupFunction))) {
c.logger.Debugln("syncing pooler is not required")
return nil, nil
}
@@ -723,15 +723,9 @@ func (c *Cluster) syncConnectionPooler(oldSpec, newSpec *acidv1.Postgresql, Look
for _, role := range [2]PostgresRole{Master, Replica} {
if role == Master {
newNeedConnectionPooler = needMasterConnectionPoolerWorker(&newSpec.Spec)
if oldSpec != nil {
oldNeedConnectionPooler = needMasterConnectionPoolerWorker(&oldSpec.Spec)
}
connectionPoolerNeeded = needMasterConnectionPoolerWorker(&newSpec.Spec)
} else {
newNeedConnectionPooler = needReplicaConnectionPoolerWorker(&newSpec.Spec)
if oldSpec != nil {
oldNeedConnectionPooler = needReplicaConnectionPoolerWorker(&oldSpec.Spec)
}
connectionPoolerNeeded = needReplicaConnectionPoolerWorker(&newSpec.Spec)
}
// if the call is via createConnectionPooler, then it is required to initialize
@@ -751,24 +745,22 @@ func (c *Cluster) syncConnectionPooler(oldSpec, newSpec *acidv1.Postgresql, Look
}
}
if newNeedConnectionPooler {
if connectionPoolerNeeded {
// Try to sync in any case. If we didn't needed connection pooler before,
// it means we want to create it. If it was already present, still sync
// since it could happen that there is no difference in specs, and all
// the resources are remembered, but the deployment was manually deleted
// in between
// in this case also do not forget to install lookup function as for
// creating cluster
if !oldNeedConnectionPooler || !c.ConnectionPooler[role].LookupFunction {
newConnectionPooler := newSpec.Spec.ConnectionPooler
// in this case also do not forget to install lookup function
if !c.ConnectionPooler[role].LookupFunction {
connectionPooler := c.Spec.ConnectionPooler
specSchema := ""
specUser := ""
if newConnectionPooler != nil {
specSchema = newConnectionPooler.Schema
specUser = newConnectionPooler.User
if connectionPooler != nil {
specSchema = connectionPooler.Schema
specUser = connectionPooler.User
}
schema := util.Coalesce(
@@ -779,9 +771,10 @@ func (c *Cluster) syncConnectionPooler(oldSpec, newSpec *acidv1.Postgresql, Look
specUser,
c.OpConfig.ConnectionPooler.User)
if err = LookupFunction(schema, user, role); err != nil {
if err = LookupFunction(schema, user); err != nil {
return NoSync, err
}
c.ConnectionPooler[role].LookupFunction = true
}
if reason, err = c.syncConnectionPoolerWorker(oldSpec, newSpec, role); err != nil {
@@ -800,8 +793,8 @@ func (c *Cluster) syncConnectionPooler(oldSpec, newSpec *acidv1.Postgresql, Look
}
}
}
if !needMasterConnectionPoolerWorker(&newSpec.Spec) &&
!needReplicaConnectionPoolerWorker(&newSpec.Spec) {
if (needMasterConnectionPoolerWorker(&oldSpec.Spec) || needReplicaConnectionPoolerWorker(&oldSpec.Spec)) &&
!needMasterConnectionPoolerWorker(&newSpec.Spec) && !needReplicaConnectionPoolerWorker(&newSpec.Spec) {
if err = c.deleteConnectionPoolerSecret(); err != nil {
c.logger.Warningf("could not remove connection pooler secret: %v", err)
}
@@ -866,8 +859,6 @@ func (c *Cluster) syncConnectionPoolerWorker(oldSpec, newSpec *acidv1.Postgresql
newConnectionPooler = &acidv1.ConnectionPooler{}
}
c.logger.Infof("old: %+v, new %+v", oldConnectionPooler, newConnectionPooler)
var specSync bool
var specReason []string
pkg/cluster/connection_pooler_test.go
+1 -1
View File
@@ -19,7 +19,7 @@ import (
"k8s.io/client-go/kubernetes/fake"
)
func mockInstallLookupFunction(schema string, user string, role PostgresRole) error {
func mockInstallLookupFunction(schema string, user string) error {
return nil
}
pkg/cluster/database.go
+3 -3
View File
@@ -508,7 +508,7 @@ func (c *Cluster) execCreateOrAlterExtension(extName, schemaName, statement, doi
// Creates a connection pool credentials lookup function in every database to
// perform remote authentication.
func (c *Cluster) installLookupFunction(poolerSchema, poolerUser string, role PostgresRole) error {
func (c *Cluster) installLookupFunction(poolerSchema, poolerUser string) error {
var stmtBytes bytes.Buffer
c.logger.Info("Installing lookup function")
@@ -604,8 +604,8 @@ func (c *Cluster) installLookupFunction(poolerSchema, poolerUser string, role Po
c.logger.Infof("pooler lookup function installed into %s", dbname)
}
if len(failedDatabases) == 0 {
c.ConnectionPooler[role].LookupFunction = true
if len(failedDatabases) > 0 {
return fmt.Errorf("could not install pooler lookup function in every specified databases")
}
return nil
pkg/cluster/k8sres.go
+18 -5
View File
@@ -798,6 +798,12 @@ func (c *Cluster) generateSpiloPodEnvVars(uid types.UID, spiloConfiguration stri
envVars = append(envVars, v1.EnvVar{Name: "WAL_BUCKET_SCOPE_PREFIX", Value: ""})
}
if c.OpConfig.WALAZStorageAccount != "" {
envVars = append(envVars, v1.EnvVar{Name: "AZURE_STORAGE_ACCOUNT", Value: c.OpConfig.WALAZStorageAccount})
envVars = append(envVars, v1.EnvVar{Name: "WAL_BUCKET_SCOPE_SUFFIX", Value: getBucketScopeSuffix(string(uid))})
envVars = append(envVars, v1.EnvVar{Name: "WAL_BUCKET_SCOPE_PREFIX", Value: ""})
}
if c.OpConfig.GCPCredentials != "" {
envVars = append(envVars, v1.EnvVar{Name: "GOOGLE_APPLICATION_CREDENTIALS", Value: c.OpConfig.GCPCredentials})
}
@@ -1170,9 +1176,6 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
}
// generate the spilo container
c.logger.Debugf("Generating Spilo container, environment variables")
c.logger.Debugf("%v", spiloEnvVars)
spiloContainer := generateContainer(constants.PostgresContainerName,
&effectiveDockerImage,
resourceRequirements,
@@ -1275,7 +1278,7 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
}
if volumeClaimTemplate, err = generatePersistentVolumeClaimTemplate(spec.Volume.Size,
spec.Volume.StorageClass); err != nil {
spec.Volume.StorageClass, spec.Volume.Selector); err != nil {
return nil, fmt.Errorf("could not generate volume claim template: %v", err)
}
@@ -1523,7 +1526,8 @@ func (c *Cluster) addAdditionalVolumes(podSpec *v1.PodSpec,
podSpec.Volumes = volumes
}
func generatePersistentVolumeClaimTemplate(volumeSize, volumeStorageClass string) (*v1.PersistentVolumeClaim, error) {
func generatePersistentVolumeClaimTemplate(volumeSize, volumeStorageClass string,
volumeSelector *metav1.LabelSelector) (*v1.PersistentVolumeClaim, error) {
var storageClassName *string
@@ -1556,6 +1560,7 @@ func generatePersistentVolumeClaimTemplate(volumeSize, volumeStorageClass string
},
StorageClassName: storageClassName,
VolumeMode: &volumeMode,
Selector: volumeSelector,
},
}
@@ -1806,6 +1811,14 @@ func (c *Cluster) generateCloneEnvironment(description *acidv1.CloneDescription)
},
}
result = append(result, envs...)
} else if c.OpConfig.WALAZStorageAccount != "" {
envs := []v1.EnvVar{
{
Name: "CLONE_AZURE_STORAGE_ACCOUNT",
Value: c.OpConfig.WALAZStorageAccount,
},
}
result = append(result, envs...)
} else {
c.logger.Error("Cannot figure out S3 or GS bucket. Both are empty.")
}
pkg/cluster/k8sres_test.go
+103
View File
@@ -1509,3 +1509,106 @@ func TestGenerateCapabilities(t *testing.T) {
}
}
}
func TestVolumeSelector(t *testing.T) {
testName := "TestVolumeSelector"
makeSpec := func(volume acidv1.Volume) acidv1.PostgresSpec {
return acidv1.PostgresSpec{
TeamID: "myapp",
NumberOfInstances: 0,
Resources: acidv1.Resources{
ResourceRequests: acidv1.ResourceDescription{CPU: "1", Memory: "10"},
ResourceLimits: acidv1.ResourceDescription{CPU: "1", Memory: "10"},
},
Volume: volume,
}
}
tests := []struct {
subTest string
volume acidv1.Volume
wantSelector *metav1.LabelSelector
}{
{
subTest: "PVC template has no selector",
volume: acidv1.Volume{
Size: "1G",
},
wantSelector: nil,
},
{
subTest: "PVC template has simple label selector",
volume: acidv1.Volume{
Size: "1G",
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{"environment": "unittest"},
},
},
wantSelector: &metav1.LabelSelector{
MatchLabels: map[string]string{"environment": "unittest"},
},
},
{
subTest: "PVC template has full selector",
volume: acidv1.Volume{
Size: "1G",
Selector: &metav1.LabelSelector{
MatchLabels: map[string]string{"environment": "unittest"},
MatchExpressions: []metav1.LabelSelectorRequirement{
{
Key: "flavour",
Operator: metav1.LabelSelectorOpIn,
Values: []string{"banana", "chocolate"},
},
},
},
},
wantSelector: &metav1.LabelSelector{
MatchLabels: map[string]string{"environment": "unittest"},
MatchExpressions: []metav1.LabelSelectorRequirement{
{
Key: "flavour",
Operator: metav1.LabelSelectorOpIn,
Values: []string{"banana", "chocolate"},
},
},
},
},
}
cluster := New(
Config{
OpConfig: config.Config{
PodManagementPolicy: "ordered_ready",
ProtectedRoles: []string{"admin"},
Auth: config.Auth{
SuperUsername: superUserName,
ReplicationUsername: replicationUserName,
},
},
}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger, eventRecorder)
for _, tt := range tests {
pgSpec := makeSpec(tt.volume)
sts, err := cluster.generateStatefulSet(&pgSpec)
if err != nil {
t.Fatalf("%s %s: no statefulset created %v", testName, tt.subTest, err)
}
volIdx := len(sts.Spec.VolumeClaimTemplates)
for i, ct := range sts.Spec.VolumeClaimTemplates {
if ct.ObjectMeta.Name == constants.DataVolumeName {
volIdx = i
break
}
}
if volIdx == len(sts.Spec.VolumeClaimTemplates) {
t.Errorf("%s %s: no datavolume found in sts", testName, tt.subTest)
}
selector := sts.Spec.VolumeClaimTemplates[volIdx].Spec.Selector
if !reflect.DeepEqual(selector, tt.wantSelector) {
t.Errorf("%s %s: expected: %#v but got: %#v", testName, tt.subTest, tt.wantSelector, selector)
}
}
}
pkg/cluster/sync.go
+9
View File
@@ -746,6 +746,15 @@ func (c *Cluster) syncDatabases() error {
}
}
if len(createDatabases) > 0 {
// trigger creation of pooler objects in new database in syncConnectionPooler
if c.ConnectionPooler != nil {
for _, role := range [2]PostgresRole{Master, Replica} {
c.ConnectionPooler[role].LookupFunction = false
}
}
}
// set default privileges for prepared database
for _, preparedDatabase := range preparedDatabases {
if err := c.initDbConnWithName(preparedDatabase); err != nil {
pkg/cluster/types.go
+1 -1
View File
@@ -72,7 +72,7 @@ type ClusterStatus struct {
type TemplateParams map[string]interface{}
type InstallFunction func(schema string, user string, role PostgresRole) error
type InstallFunction func(schema string, user string) error
type SyncReason []string
pkg/controller/operator_config.go
+3 -2
View File
@@ -39,7 +39,7 @@ func (c *Controller) importConfigurationFromCRD(fromCRD *acidv1.OperatorConfigur
result.EnableSpiloWalPathCompat = fromCRD.EnableSpiloWalPathCompat
result.EtcdHost = fromCRD.EtcdHost
result.KubernetesUseConfigMaps = fromCRD.KubernetesUseConfigMaps
result.DockerImage = util.Coalesce(fromCRD.DockerImage, "registry.opensource.zalan.do/acid/spilo-13:2.0-p7")
result.DockerImage = util.Coalesce(fromCRD.DockerImage, "registry.opensource.zalan.do/acid/spilo-13:2.1-p1")
result.Workers = util.CoalesceUInt32(fromCRD.Workers, 8)
result.MinInstances = fromCRD.MinInstances
result.MaxInstances = fromCRD.MaxInstances
@@ -146,6 +146,7 @@ func (c *Controller) importConfigurationFromCRD(fromCRD *acidv1.OperatorConfigur
result.KubeIAMRole = fromCRD.AWSGCP.KubeIAMRole
result.WALGSBucket = fromCRD.AWSGCP.WALGSBucket
result.GCPCredentials = fromCRD.AWSGCP.GCPCredentials
result.WALAZStorageAccount = fromCRD.AWSGCP.WALAZStorageAccount
result.AdditionalSecretMount = fromCRD.AWSGCP.AdditionalSecretMount
result.AdditionalSecretMountPath = util.Coalesce(fromCRD.AWSGCP.AdditionalSecretMountPath, "/meta/credentials")
result.EnableEBSGp3Migration = fromCRD.AWSGCP.EnableEBSGp3Migration
@@ -153,7 +154,7 @@ func (c *Controller) importConfigurationFromCRD(fromCRD *acidv1.OperatorConfigur
// logical backup config
result.LogicalBackupSchedule = util.Coalesce(fromCRD.LogicalBackup.Schedule, "30 00 * * *")
result.LogicalBackupDockerImage = util.Coalesce(fromCRD.LogicalBackup.DockerImage, "registry.opensource.zalan.do/acid/logical-backup:v1.6.3")
result.LogicalBackupDockerImage = util.Coalesce(fromCRD.LogicalBackup.DockerImage, "registry.opensource.zalan.do/acid/logical-backup:v1.7.0")
result.LogicalBackupProvider = util.Coalesce(fromCRD.LogicalBackup.BackupProvider, "s3")
result.LogicalBackupS3Bucket = fromCRD.LogicalBackup.S3Bucket
result.LogicalBackupS3Region = fromCRD.LogicalBackup.S3Region
pkg/util/config/config.go
+3 -2
View File
@@ -114,7 +114,7 @@ type Scalyr struct {
// LogicalBackup defines configuration for logical backup
type LogicalBackup struct {
LogicalBackupSchedule string `name:"logical_backup_schedule" default:"30 00 * * *"`
LogicalBackupDockerImage string `name:"logical_backup_docker_image" default:"registry.opensource.zalan.do/acid/logical-backup:v1.6.3"`
LogicalBackupDockerImage string `name:"logical_backup_docker_image" default:"registry.opensource.zalan.do/acid/logical-backup:v1.7.0"`
LogicalBackupProvider string `name:"logical_backup_provider" default:"s3"`
LogicalBackupS3Bucket string `name:"logical_backup_s3_bucket" default:""`
LogicalBackupS3Region string `name:"logical_backup_s3_region" default:""`
@@ -152,7 +152,7 @@ type Config struct {
WatchedNamespace string `name:"watched_namespace"` // special values: "*" means 'watch all namespaces', the empty string "" means 'watch a namespace where operator is deployed to'
KubernetesUseConfigMaps bool `name:"kubernetes_use_configmaps" default:"false"`
EtcdHost string `name:"etcd_host" default:""` // special values: the empty string "" means Patroni will use K8s as a DCS
DockerImage string `name:"docker_image" default:"registry.opensource.zalan.do/acid/spilo-13:2.0-p7"`
DockerImage string `name:"docker_image" default:"registry.opensource.zalan.do/acid/spilo-13:2.1-p1"`
SidecarImages map[string]string `name:"sidecar_docker_images"` // deprecated in favour of SidecarContainers
SidecarContainers []v1.Container `name:"sidecars"`
PodServiceAccountName string `name:"pod_service_account_name" default:"postgres-pod"`
@@ -167,6 +167,7 @@ type Config struct {
KubeIAMRole string `name:"kube_iam_role"`
WALGSBucket string `name:"wal_gs_bucket"`
GCPCredentials string `name:"gcp_credentials"`
WALAZStorageAccount string `name:"wal_az_storage_account"`
AdditionalSecretMount string `name:"additional_secret_mount"`
AdditionalSecretMountPath string `name:"additional_secret_mount_path" default:"/meta/credentials"`
EnableEBSGp3Migration bool `name:"enable_ebs_gp3_migration" default:"false"`
pkg/util/httpclient/httpclient.go
+1 -1
View File
@@ -1,6 +1,6 @@
package httpclient
//go:generate mockgen -package mocks -destination=$PWD/mocks/$GOFILE -source=$GOFILE -build_flags=-mod=vendor
//go:generate mockgen -package mocks -destination=../../../mocks/$GOFILE -source=$GOFILE -build_flags=-mod=vendor
import "net/http"
pkg/util/volumes/volumes.go
+1 -1
View File
@@ -1,6 +1,6 @@
package volumes
//go:generate mockgen -package mocks -destination=$PWD/mocks/$GOFILE -source=$GOFILE -build_flags=-mod=vendor
//go:generate mockgen -package mocks -destination=../../../mocks/$GOFILE -source=$GOFILE -build_flags=-mod=vendor
import v1 "k8s.io/api/core/v1"