Merge branch 'master' into feature/connection-pooler

This commit is contained in:
Dmitrii Dolgov
2020-03-16 12:07:56 +01:00
20 changed files with 501 additions and 38 deletions
+18
View File
@@ -490,6 +490,24 @@ var PostgresCRDResourceValidation = apiextv1beta1.CustomResourceValidation{
"teamId": {
Type: "string",
},
"tls": {
Type: "object",
Required: []string{"secretName"},
Properties: map[string]apiextv1beta1.JSONSchemaProps{
"secretName": {
Type: "string",
},
"certificateFile": {
Type: "string",
},
"privateKeyFile": {
Type: "string",
},
"caFile": {
Type: "string",
},
},
},
"tolerations": {
Type: "array",
Items: &apiextv1beta1.JSONSchemaPropsOrArray{
@@ -55,7 +55,7 @@ type KubernetesMetaConfiguration struct {
EnableInitContainers *bool `json:"enable_init_containers,omitempty"`
EnableSidecars *bool `json:"enable_sidecars,omitempty"`
SecretNameTemplate config.StringTemplate `json:"secret_name_template,omitempty"`
ClusterDomain string `json:"cluster_domain"`
ClusterDomain string `json:"cluster_domain,omitempty"`
OAuthTokenSecretName spec.NamespacedName `json:"oauth_token_secret_name,omitempty"`
InfrastructureRolesSecretName spec.NamespacedName `json:"infrastructure_roles_secret_name,omitempty"`
PodRoleLabel string `json:"pod_role_label,omitempty"`
@@ -66,6 +66,7 @@ type PostgresSpec struct {
StandbyCluster *StandbyDescription `json:"standby"`
PodAnnotations map[string]string `json:"podAnnotations"`
ServiceAnnotations map[string]string `json:"serviceAnnotations"`
TLS *TLSDescription `json:"tls"`
// deprecated json tags
InitContainersOld []v1.Container `json:"init_containers,omitempty"`
@@ -131,6 +132,13 @@ type StandbyDescription struct {
S3WalPath string `json:"s3_wal_path,omitempty"`
}
type TLSDescription struct {
SecretName string `json:"secretName,omitempty"`
CertificateFile string `json:"certificateFile,omitempty"`
PrivateKeyFile string `json:"privateKeyFile,omitempty"`
CAFile string `json:"caFile,omitempty"`
}
// CloneDescription describes which cluster the new should clone and up to which point in time
type CloneDescription struct {
ClusterName string `json:"cluster,omitempty"`
@@ -585,6 +585,11 @@ func (in *PostgresSpec) DeepCopyInto(out *PostgresSpec) {
(*out)[key] = val
}
}
if in.TLS != nil {
in, out := &in.TLS, &out.TLS
*out = new(TLSDescription)
**out = **in
}
if in.InitContainersOld != nil {
in, out := &in.InitContainersOld, &out.InitContainersOld
*out = make([]corev1.Container, len(*in))
@@ -816,6 +821,22 @@ func (in *StandbyDescription) DeepCopy() *StandbyDescription {
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *TLSDescription) DeepCopyInto(out *TLSDescription) {
*out = *in
return
}
// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new TLSDescription.
func (in *TLSDescription) DeepCopy() *TLSDescription {
if in == nil {
return nil
}
out := new(TLSDescription)
in.DeepCopyInto(out)
return out
}
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
func (in *TeamsAPIConfiguration) DeepCopyInto(out *TeamsAPIConfiguration) {
*out = *in
+4 -3
View File
@@ -593,10 +593,11 @@ func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
}
}()
if oldSpec.Spec.PgVersion != newSpec.Spec.PgVersion { // PG versions comparison
c.logger.Warningf("postgresql version change(%q -> %q) has no effect", oldSpec.Spec.PgVersion, newSpec.Spec.PgVersion)
if oldSpec.Spec.PostgresqlParam.PgVersion != newSpec.Spec.PostgresqlParam.PgVersion { // PG versions comparison
c.logger.Warningf("postgresql version change(%q -> %q) has no effect",
oldSpec.Spec.PostgresqlParam.PgVersion, newSpec.Spec.PostgresqlParam.PgVersion)
//we need that hack to generate statefulset with the old version
newSpec.Spec.PgVersion = oldSpec.Spec.PgVersion
newSpec.Spec.PostgresqlParam.PgVersion = oldSpec.Spec.PostgresqlParam.PgVersion
}
// Service
+128 -18
View File
@@ -3,6 +3,7 @@ package cluster
import (
"encoding/json"
"fmt"
"path"
"sort"
"github.com/sirupsen/logrus"
@@ -27,13 +28,16 @@ import (
)
const (
pgBinariesLocationTemplate = "/usr/lib/postgresql/%s/bin"
pgBinariesLocationTemplate = "/usr/lib/postgresql/%v/bin"
patroniPGBinariesParameterName = "bin_dir"
patroniPGParametersParameterName = "parameters"
patroniPGHBAConfParameterName = "pg_hba"
localHost = "127.0.0.1/32"
connectionPoolContainer = "connection-pool"
pgPort = 5432
// the gid of the postgres user in the default spilo image
spiloPostgresGID = 103
)
type pgUser struct {
@@ -508,6 +512,7 @@ func generatePodTemplate(
podAntiAffinityTopologyKey string,
additionalSecretMount string,
additionalSecretMountPath string,
volumes []v1.Volume,
) (*v1.PodTemplateSpec, error) {
terminateGracePeriodSeconds := terminateGracePeriod
@@ -526,6 +531,7 @@ func generatePodTemplate(
InitContainers: initContainers,
Tolerations: *tolerationsSpec,
SecurityContext: &securityContext,
Volumes: volumes,
}
if shmVolume != nil && *shmVolume {
@@ -778,6 +784,50 @@ func makeResources(cpuRequest, memoryRequest, cpuLimit, memoryLimit string) acid
}
}
func extractPgVersionFromBinPath(binPath string, template string) (string, error) {
var pgVersion float32
_, err := fmt.Sscanf(binPath, template, &pgVersion)
if err != nil {
return "", err
}
return fmt.Sprintf("%v", pgVersion), nil
}
func (c *Cluster) getNewPgVersion(container v1.Container, newPgVersion string) (string, error) {
var (
spiloConfiguration spiloConfiguration
runningPgVersion string
err error
)
for _, env := range container.Env {
if env.Name != "SPILO_CONFIGURATION" {
continue
}
err = json.Unmarshal([]byte(env.Value), &spiloConfiguration)
if err != nil {
return newPgVersion, err
}
}
if len(spiloConfiguration.PgLocalConfiguration) > 0 {
currentBinPath := fmt.Sprintf("%v", spiloConfiguration.PgLocalConfiguration[patroniPGBinariesParameterName])
runningPgVersion, err = extractPgVersionFromBinPath(currentBinPath, pgBinariesLocationTemplate)
if err != nil {
return "", fmt.Errorf("could not extract Postgres version from %v in SPILO_CONFIGURATION", currentBinPath)
}
} else {
return "", fmt.Errorf("could not find %q setting in SPILO_CONFIGURATION", patroniPGBinariesParameterName)
}
if runningPgVersion != newPgVersion {
c.logger.Warningf("postgresql version change(%q -> %q) has no effect", runningPgVersion, newPgVersion)
newPgVersion = runningPgVersion
}
return newPgVersion, nil
}
func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.StatefulSet, error) {
var (
@@ -786,6 +836,7 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
sidecarContainers []v1.Container
podTemplate *v1.PodTemplateSpec
volumeClaimTemplate *v1.PersistentVolumeClaim
volumes []v1.Volume
)
// Improve me. Please.
@@ -902,21 +953,76 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
}
// generate environment variables for the spilo container
spiloEnvVars := deduplicateEnvVars(
c.generateSpiloPodEnvVars(c.Postgresql.GetUID(), spiloConfiguration, &spec.Clone,
spec.StandbyCluster, customPodEnvVarsList), c.containerName(), c.logger)
spiloEnvVars := c.generateSpiloPodEnvVars(
c.Postgresql.GetUID(),
spiloConfiguration,
&spec.Clone,
spec.StandbyCluster,
customPodEnvVarsList,
)
// pickup the docker image for the spilo container
effectiveDockerImage := util.Coalesce(spec.DockerImage, c.OpConfig.DockerImage)
// determine the FSGroup for the spilo pod
effectiveFSGroup := c.OpConfig.Resources.SpiloFSGroup
if spec.SpiloFSGroup != nil {
effectiveFSGroup = spec.SpiloFSGroup
}
volumeMounts := generateVolumeMounts(spec.Volume)
// configure TLS with a custom secret volume
if spec.TLS != nil && spec.TLS.SecretName != "" {
if effectiveFSGroup == nil {
c.logger.Warnf("Setting the default FSGroup to satisfy the TLS configuration")
fsGroup := int64(spiloPostgresGID)
effectiveFSGroup = &fsGroup
}
// this is combined with the FSGroup above to give read access to the
// postgres user
defaultMode := int32(0640)
volumes = append(volumes, v1.Volume{
Name: "tls-secret",
VolumeSource: v1.VolumeSource{
Secret: &v1.SecretVolumeSource{
SecretName: spec.TLS.SecretName,
DefaultMode: &defaultMode,
},
},
})
mountPath := "/tls"
volumeMounts = append(volumeMounts, v1.VolumeMount{
MountPath: mountPath,
Name: "tls-secret",
ReadOnly: true,
})
// use the same filenames as Secret resources by default
certFile := ensurePath(spec.TLS.CertificateFile, mountPath, "tls.crt")
privateKeyFile := ensurePath(spec.TLS.PrivateKeyFile, mountPath, "tls.key")
spiloEnvVars = append(
spiloEnvVars,
v1.EnvVar{Name: "SSL_CERTIFICATE_FILE", Value: certFile},
v1.EnvVar{Name: "SSL_PRIVATE_KEY_FILE", Value: privateKeyFile},
)
if spec.TLS.CAFile != "" {
caFile := ensurePath(spec.TLS.CAFile, mountPath, "")
spiloEnvVars = append(
spiloEnvVars,
v1.EnvVar{Name: "SSL_CA_FILE", Value: caFile},
)
}
}
// generate the spilo container
c.logger.Debugf("Generating Spilo container, environment variables: %v", spiloEnvVars)
spiloContainer := generateContainer(c.containerName(),
&effectiveDockerImage,
resourceRequirements,
spiloEnvVars,
deduplicateEnvVars(spiloEnvVars, c.containerName(), c.logger),
volumeMounts,
c.OpConfig.Resources.SpiloPrivileged,
)
@@ -955,16 +1061,10 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
tolerationSpec := tolerations(&spec.Tolerations, c.OpConfig.PodToleration)
effectivePodPriorityClassName := util.Coalesce(spec.PodPriorityClassName, c.OpConfig.PodPriorityClassName)
// determine the FSGroup for the spilo pod
effectiveFSGroup := c.OpConfig.Resources.SpiloFSGroup
if spec.SpiloFSGroup != nil {
effectiveFSGroup = spec.SpiloFSGroup
}
annotations := c.generatePodAnnotations(spec)
// generate pod template for the statefulset, based on the spilo container and sidecars
if podTemplate, err = generatePodTemplate(
podTemplate, err = generatePodTemplate(
c.Namespace,
c.labelsSet(true),
annotations,
@@ -982,10 +1082,9 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.Statef
c.OpConfig.EnablePodAntiAffinity,
c.OpConfig.PodAntiAffinityTopologyKey,
c.OpConfig.AdditionalSecretMount,
c.OpConfig.AdditionalSecretMountPath); err != nil {
return nil, fmt.Errorf("could not generate pod template: %v", err)
}
c.OpConfig.AdditionalSecretMountPath,
volumes,
)
if err != nil {
return nil, fmt.Errorf("could not generate pod template: %v", err)
}
@@ -1601,7 +1700,8 @@ func (c *Cluster) generateLogicalBackupJob() (*batchv1beta1.CronJob, error) {
false,
"",
c.OpConfig.AdditionalSecretMount,
c.OpConfig.AdditionalSecretMountPath); err != nil {
c.OpConfig.AdditionalSecretMountPath,
nil); err != nil {
return nil, fmt.Errorf("could not generate pod template for logical backup pod: %v", err)
}
@@ -1686,7 +1786,7 @@ func (c *Cluster) generateLogicalBackupPodEnvVars() []v1.EnvVar {
// Postgres env vars
{
Name: "PG_VERSION",
Value: c.Spec.PgVersion,
Value: c.Spec.PostgresqlParam.PgVersion,
},
{
Name: "PGPORT",
@@ -2026,3 +2126,13 @@ func (c *Cluster) generateConnPoolService(spec *acidv1.PostgresSpec) *v1.Service
return service
}
func ensurePath(file string, defaultDir string, defaultFile string) string {
if file == "" {
return path.Join(defaultDir, defaultFile)
}
if !path.IsAbs(file) {
return path.Join(defaultDir, file)
}
return file
}
+194 -2
View File
@@ -5,10 +5,10 @@ import (
"fmt"
"reflect"
v1 "k8s.io/api/core/v1"
"testing"
"github.com/stretchr/testify/assert"
acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
"github.com/zalando/postgres-operator/pkg/util"
"github.com/zalando/postgres-operator/pkg/util/config"
@@ -16,6 +16,7 @@ import (
"github.com/zalando/postgres-operator/pkg/util/k8sutil"
appsv1 "k8s.io/api/apps/v1"
v1 "k8s.io/api/core/v1"
policyv1beta1 "k8s.io/api/policy/v1beta1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/intstr"
@@ -384,6 +385,135 @@ func TestCloneEnv(t *testing.T) {
}
}
func TestExtractPgVersionFromBinPath(t *testing.T) {
testName := "TestExtractPgVersionFromBinPath"
tests := []struct {
subTest string
binPath string
template string
expected string
}{
{
subTest: "test current bin path with decimal against hard coded template",
binPath: "/usr/lib/postgresql/9.6/bin",
template: pgBinariesLocationTemplate,
expected: "9.6",
},
{
subTest: "test current bin path against hard coded template",
binPath: "/usr/lib/postgresql/12/bin",
template: pgBinariesLocationTemplate,
expected: "12",
},
{
subTest: "test alternative bin path against a matching template",
binPath: "/usr/pgsql-12/bin",
template: "/usr/pgsql-%v/bin",
expected: "12",
},
}
for _, tt := range tests {
pgVersion, err := extractPgVersionFromBinPath(tt.binPath, tt.template)
if err != nil {
t.Errorf("Unexpected error: %v", err)
}
if pgVersion != tt.expected {
t.Errorf("%s %s: Expected version %s, have %s instead",
testName, tt.subTest, tt.expected, pgVersion)
}
}
}
func TestGetPgVersion(t *testing.T) {
testName := "TestGetPgVersion"
tests := []struct {
subTest string
pgContainer v1.Container
currentPgVersion string
newPgVersion string
}{
{
subTest: "new version with decimal point differs from current SPILO_CONFIGURATION",
pgContainer: v1.Container{
Name: "postgres",
Env: []v1.EnvVar{
{
Name: "SPILO_CONFIGURATION",
Value: "{\"postgresql\": {\"bin_dir\": \"/usr/lib/postgresql/9.6/bin\"}}",
},
},
},
currentPgVersion: "9.6",
newPgVersion: "12",
},
{
subTest: "new version differs from current SPILO_CONFIGURATION",
pgContainer: v1.Container{
Name: "postgres",
Env: []v1.EnvVar{
{
Name: "SPILO_CONFIGURATION",
Value: "{\"postgresql\": {\"bin_dir\": \"/usr/lib/postgresql/11/bin\"}}",
},
},
},
currentPgVersion: "11",
newPgVersion: "12",
},
{
subTest: "new version is lower than the one found in current SPILO_CONFIGURATION",
pgContainer: v1.Container{
Name: "postgres",
Env: []v1.EnvVar{
{
Name: "SPILO_CONFIGURATION",
Value: "{\"postgresql\": {\"bin_dir\": \"/usr/lib/postgresql/12/bin\"}}",
},
},
},
currentPgVersion: "12",
newPgVersion: "11",
},
{
subTest: "new version is the same like in the current SPILO_CONFIGURATION",
pgContainer: v1.Container{
Name: "postgres",
Env: []v1.EnvVar{
{
Name: "SPILO_CONFIGURATION",
Value: "{\"postgresql\": {\"bin_dir\": \"/usr/lib/postgresql/12/bin\"}}",
},
},
},
currentPgVersion: "12",
newPgVersion: "12",
},
}
var cluster = New(
Config{
OpConfig: config.Config{
ProtectedRoles: []string{"admin"},
Auth: config.Auth{
SuperUsername: superUserName,
ReplicationUsername: replicationUserName,
},
},
}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
for _, tt := range tests {
pgVersion, err := cluster.getNewPgVersion(tt.pgContainer, tt.newPgVersion)
if err != nil {
t.Errorf("Unexpected error: %v", err)
}
if pgVersion != tt.currentPgVersion {
t.Errorf("%s %s: Expected version %s, have %s instead",
testName, tt.subTest, tt.currentPgVersion, pgVersion)
}
}
}
func TestSecretVolume(t *testing.T) {
testName := "TestSecretVolume"
tests := []struct {
@@ -823,3 +953,65 @@ func TestConnPoolServiceSpec(t *testing.T) {
}
}
}
func TestTLS(t *testing.T) {
var err error
var spec acidv1.PostgresSpec
var cluster *Cluster
makeSpec := func(tls acidv1.TLSDescription) acidv1.PostgresSpec {
return acidv1.PostgresSpec{
TeamID: "myapp", NumberOfInstances: 1,
Resources: acidv1.Resources{
ResourceRequests: acidv1.ResourceDescription{CPU: "1", Memory: "10"},
ResourceLimits: acidv1.ResourceDescription{CPU: "1", Memory: "10"},
},
Volume: acidv1.Volume{
Size: "1G",
},
TLS: &tls,
}
}
cluster = New(
Config{
OpConfig: config.Config{
PodManagementPolicy: "ordered_ready",
ProtectedRoles: []string{"admin"},
Auth: config.Auth{
SuperUsername: superUserName,
ReplicationUsername: replicationUserName,
},
},
}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
spec = makeSpec(acidv1.TLSDescription{SecretName: "my-secret", CAFile: "ca.crt"})
s, err := cluster.generateStatefulSet(&spec)
if err != nil {
assert.NoError(t, err)
}
fsGroup := int64(103)
assert.Equal(t, &fsGroup, s.Spec.Template.Spec.SecurityContext.FSGroup, "has a default FSGroup assigned")
defaultMode := int32(0640)
volume := v1.Volume{
Name: "tls-secret",
VolumeSource: v1.VolumeSource{
Secret: &v1.SecretVolumeSource{
SecretName: "my-secret",
DefaultMode: &defaultMode,
},
},
}
assert.Contains(t, s.Spec.Template.Spec.Volumes, volume, "the pod gets a secret volume")
assert.Contains(t, s.Spec.Template.Spec.Containers[0].VolumeMounts, v1.VolumeMount{
MountPath: "/tls",
Name: "tls-secret",
ReadOnly: true,
}, "the volume gets mounted in /tls")
assert.Contains(t, s.Spec.Template.Spec.Containers[0].Env, v1.EnvVar{Name: "SSL_CERTIFICATE_FILE", Value: "/tls/tls.crt"})
assert.Contains(t, s.Spec.Template.Spec.Containers[0].Env, v1.EnvVar{Name: "SSL_PRIVATE_KEY_FILE", Value: "/tls/tls.key"})
assert.Contains(t, s.Spec.Template.Spec.Containers[0].Env, v1.EnvVar{Name: "SSL_CA_FILE", Value: "/tls/ca.crt"})
}
+12
View File
@@ -291,6 +291,18 @@ func (c *Cluster) syncStatefulSet() error {
// statefulset is already there, make sure we use its definition in order to compare with the spec.
c.Statefulset = sset
// check if there is no Postgres version mismatch
for _, container := range c.Statefulset.Spec.Template.Spec.Containers {
if container.Name != "postgres" {
continue
}
pgVersion, err := c.getNewPgVersion(container, c.Spec.PostgresqlParam.PgVersion)
if err != nil {
return fmt.Errorf("could not parse current Postgres version: %v", err)
}
c.Spec.PostgresqlParam.PgVersion = pgVersion
}
desiredSS, err := c.generateStatefulSet(&c.Spec)
if err != nil {
return fmt.Errorf("could not generate statefulset: %v", err)
+1 -1
View File
@@ -56,7 +56,7 @@ func (c *Controller) importConfigurationFromCRD(fromCRD *acidv1.OperatorConfigur
result.PodTerminateGracePeriod = time.Duration(fromCRD.Kubernetes.PodTerminateGracePeriod)
result.SpiloPrivileged = fromCRD.Kubernetes.SpiloPrivileged
result.SpiloFSGroup = fromCRD.Kubernetes.SpiloFSGroup
result.ClusterDomain = fromCRD.Kubernetes.ClusterDomain
result.ClusterDomain = util.Coalesce(fromCRD.Kubernetes.ClusterDomain, "cluster.local")
result.WatchedNamespace = fromCRD.Kubernetes.WatchedNamespace
result.PDBNameFormat = fromCRD.Kubernetes.PDBNameFormat
result.EnablePodDisruptionBudget = fromCRD.Kubernetes.EnablePodDisruptionBudget