mirror of
https://github.com/zalando/postgres-operator.git
synced 2026-10-06 21:31:47 +02:00
Merge branch 'master' into feature/actions-for-secret-syncing
This commit is contained in:
@@ -1,10 +1,7 @@
|
||||
package v1
|
||||
|
||||
// ClusterStatusUnknown etc : status of a Postgres cluster known to the operator
|
||||
const (
|
||||
serviceNameMaxLength = 63
|
||||
clusterNameMaxLength = serviceNameMaxLength - len("-repl")
|
||||
serviceNameRegexString = `^[a-z]([-a-z0-9]*[a-z0-9])?$`
|
||||
|
||||
ClusterStatusUnknown PostgresStatus = ""
|
||||
ClusterStatusCreating PostgresStatus = "Creating"
|
||||
ClusterStatusUpdating PostgresStatus = "Updating"
|
||||
@@ -14,3 +11,9 @@ const (
|
||||
ClusterStatusRunning PostgresStatus = "Running"
|
||||
ClusterStatusInvalid PostgresStatus = "Invalid"
|
||||
)
|
||||
|
||||
const (
|
||||
serviceNameMaxLength = 63
|
||||
clusterNameMaxLength = serviceNameMaxLength - len("-repl")
|
||||
serviceNameRegexString = `^[a-z]([-a-z0-9]*[a-z0-9])?$`
|
||||
)
|
||||
|
||||
@@ -6,6 +6,7 @@ import (
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
)
|
||||
|
||||
// CRDResource* define names necesssary for the k8s CRD API
|
||||
const (
|
||||
PostgresCRDResourceKind = "postgresql"
|
||||
PostgresCRDResourcePlural = "postgresqls"
|
||||
@@ -39,6 +40,7 @@ func buildCRD(name, kind, plural, short string) *apiextv1beta1.CustomResourceDef
|
||||
}
|
||||
}
|
||||
|
||||
// PostgresCRD returns CustomResourceDefinition built from PostgresCRDResource
|
||||
func PostgresCRD() *apiextv1beta1.CustomResourceDefinition {
|
||||
return buildCRD(PostgresCRDResouceName,
|
||||
PostgresCRDResourceKind,
|
||||
@@ -46,6 +48,7 @@ func PostgresCRD() *apiextv1beta1.CustomResourceDefinition {
|
||||
PostgresCRDResourceShort)
|
||||
}
|
||||
|
||||
// ConfigurationCRD returns CustomResourceDefinition built from OperatorConfigCRDResource
|
||||
func ConfigurationCRD() *apiextv1beta1.CustomResourceDefinition {
|
||||
return buildCRD(OperatorConfigCRDResourceName,
|
||||
OperatorConfigCRDResouceKind,
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
// Package v1 is the v1 version of the API.
|
||||
// +k8s:deepcopy-gen=package,register
|
||||
|
||||
// Package v1 is the v1 version of the API.
|
||||
// +groupName=acid.zalan.do
|
||||
|
||||
package v1
|
||||
|
||||
@@ -104,6 +104,7 @@ func (p *Postgresql) UnmarshalJSON(data []byte) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// UnmarshalJSON convert to Duration from byte slice of json
|
||||
func (d *Duration) UnmarshalJSON(b []byte) error {
|
||||
var (
|
||||
v interface{}
|
||||
|
||||
@@ -13,6 +13,8 @@ import (
|
||||
// +genclient:onlyVerbs=get
|
||||
// +genclient:noStatus
|
||||
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
|
||||
|
||||
// OperatorConfiguration defines the specification for the OperatorConfiguration.
|
||||
type OperatorConfiguration struct {
|
||||
metav1.TypeMeta `json:",inline"`
|
||||
metav1.ObjectMeta `json:"metadata"`
|
||||
@@ -21,6 +23,8 @@ type OperatorConfiguration struct {
|
||||
}
|
||||
|
||||
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
|
||||
|
||||
// OperatorConfigurationList is used in the k8s API calls
|
||||
type OperatorConfigurationList struct {
|
||||
metav1.TypeMeta `json:",inline"`
|
||||
metav1.ListMeta `json:"metadata"`
|
||||
@@ -28,11 +32,13 @@ type OperatorConfigurationList struct {
|
||||
Items []OperatorConfiguration `json:"items"`
|
||||
}
|
||||
|
||||
// PostgresUsersConfiguration defines the system users of Postgres.
|
||||
type PostgresUsersConfiguration struct {
|
||||
SuperUsername string `json:"super_username,omitempty"`
|
||||
ReplicationUsername string `json:"replication_username,omitempty"`
|
||||
}
|
||||
|
||||
// KubernetesMetaConfiguration defines k8s conf required for all Postgres clusters and the operator itself
|
||||
type KubernetesMetaConfiguration struct {
|
||||
PodServiceAccountName string `json:"pod_service_account_name,omitempty"`
|
||||
// TODO: change it to the proper json
|
||||
@@ -55,6 +61,7 @@ type KubernetesMetaConfiguration struct {
|
||||
PodPriorityClassName string `json:"pod_priority_class_name,omitempty"`
|
||||
}
|
||||
|
||||
// PostgresPodResourcesDefaults defines the spec of default resources
|
||||
type PostgresPodResourcesDefaults struct {
|
||||
DefaultCPURequest string `json:"default_cpu_request,omitempty"`
|
||||
DefaultMemoryRequest string `json:"default_memory_request,omitempty"`
|
||||
@@ -62,6 +69,7 @@ type PostgresPodResourcesDefaults struct {
|
||||
DefaultMemoryLimit string `json:"default_memory_limit,omitempty"`
|
||||
}
|
||||
|
||||
// OperatorTimeouts defines the timeout of ResourceCheck, PodWait, ReadyWait
|
||||
type OperatorTimeouts struct {
|
||||
ResourceCheckInterval Duration `json:"resource_check_interval,omitempty"`
|
||||
ResourceCheckTimeout Duration `json:"resource_check_timeout,omitempty"`
|
||||
@@ -71,6 +79,7 @@ type OperatorTimeouts struct {
|
||||
ReadyWaitTimeout Duration `json:"ready_wait_timeout,omitempty"`
|
||||
}
|
||||
|
||||
// LoadBalancerConfiguration defines the LB configuration
|
||||
type LoadBalancerConfiguration struct {
|
||||
DbHostedZone string `json:"db_hosted_zone,omitempty"`
|
||||
EnableMasterLoadBalancer bool `json:"enable_master_load_balancer,omitempty"`
|
||||
@@ -79,6 +88,8 @@ type LoadBalancerConfiguration struct {
|
||||
ReplicaDNSNameFormat config.StringTemplate `json:"replica_dns_name_format,omitempty"`
|
||||
}
|
||||
|
||||
// AWSGCPConfiguration defines the configuration for AWS
|
||||
// TODO complete Google Cloud Platform (GCP) configuration
|
||||
type AWSGCPConfiguration struct {
|
||||
WALES3Bucket string `json:"wal_s3_bucket,omitempty"`
|
||||
AWSRegion string `json:"aws_region,omitempty"`
|
||||
@@ -86,11 +97,13 @@ type AWSGCPConfiguration struct {
|
||||
KubeIAMRole string `json:"kube_iam_role,omitempty"`
|
||||
}
|
||||
|
||||
// OperatorDebugConfiguration defines options for the debug mode
|
||||
type OperatorDebugConfiguration struct {
|
||||
DebugLogging bool `json:"debug_logging,omitempty"`
|
||||
EnableDBAccess bool `json:"enable_database_access,omitempty"`
|
||||
}
|
||||
|
||||
// TeamsAPIConfiguration defines the configration of TeamsAPI
|
||||
type TeamsAPIConfiguration struct {
|
||||
EnableTeamsAPI bool `json:"enable_teams_api,omitempty"`
|
||||
TeamsAPIUrl string `json:"teams_api_url,omitempty"`
|
||||
@@ -103,12 +116,14 @@ type TeamsAPIConfiguration struct {
|
||||
PostgresSuperuserTeams []string `json:"postgres_superuser_teams,omitempty"`
|
||||
}
|
||||
|
||||
// LoggingRESTAPIConfiguration defines Logging API conf
|
||||
type LoggingRESTAPIConfiguration struct {
|
||||
APIPort int `json:"api_port,omitempty"`
|
||||
RingLogLines int `json:"ring_log_lines,omitempty"`
|
||||
ClusterHistoryEntries int `json:"cluster_history_entries,omitempty"`
|
||||
}
|
||||
|
||||
// ScalyrConfiguration defines the configuration for ScalyrAPI
|
||||
type ScalyrConfiguration struct {
|
||||
ScalyrAPIKey string `json:"scalyr_api_key,omitempty"`
|
||||
ScalyrImage string `json:"scalyr_image,omitempty"`
|
||||
@@ -119,6 +134,7 @@ type ScalyrConfiguration struct {
|
||||
ScalyrMemoryLimit string `json:"scalyr_memory_limit,omitempty"`
|
||||
}
|
||||
|
||||
// OperatorConfigurationData defines the operation config
|
||||
type OperatorConfigurationData struct {
|
||||
EtcdHost string `json:"etcd_host,omitempty"`
|
||||
DockerImage string `json:"docker_image,omitempty"`
|
||||
@@ -131,6 +147,7 @@ type OperatorConfigurationData struct {
|
||||
PostgresUsersConfiguration PostgresUsersConfiguration `json:"users"`
|
||||
Kubernetes KubernetesMetaConfiguration `json:"kubernetes"`
|
||||
PostgresPodResources PostgresPodResourcesDefaults `json:"postgres_pod_resources"`
|
||||
SetMemoryRequestToLimit bool `json:"set_memory_request_to_limit,omitempty"`
|
||||
Timeouts OperatorTimeouts `json:"timeouts"`
|
||||
LoadBalancer LoadBalancerConfiguration `json:"load_balancer"`
|
||||
AWSGCP AWSGCPConfiguration `json:"aws_or_gcp"`
|
||||
@@ -140,6 +157,7 @@ type OperatorConfigurationData struct {
|
||||
Scalyr ScalyrConfiguration `json:"scalyr"`
|
||||
}
|
||||
|
||||
// OperatorConfigurationUsers defines configration for super user
|
||||
type OperatorConfigurationUsers struct {
|
||||
SuperUserName string `json:"superuser_name,omitempty"`
|
||||
Replication string `json:"replication_user_name,omitempty"`
|
||||
@@ -147,4 +165,5 @@ type OperatorConfigurationUsers struct {
|
||||
TeamAPIRoleConfiguration map[string]string `json:"team_api_role_configuration,omitempty"`
|
||||
}
|
||||
|
||||
//Duration shortens this frequently used name
|
||||
type Duration time.Duration
|
||||
|
||||
@@ -9,7 +9,8 @@ import (
|
||||
|
||||
// +genclient
|
||||
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
|
||||
//Postgresql defines PostgreSQL Custom Resource Definition Object.
|
||||
|
||||
// Postgresql defines PostgreSQL Custom Resource Definition Object.
|
||||
type Postgresql struct {
|
||||
metav1.TypeMeta `json:",inline"`
|
||||
metav1.ObjectMeta `json:"metadata,omitempty"`
|
||||
@@ -51,9 +52,11 @@ type PostgresSpec struct {
|
||||
Tolerations []v1.Toleration `json:"tolerations,omitempty"`
|
||||
Sidecars []Sidecar `json:"sidecars,omitempty"`
|
||||
PodPriorityClassName string `json:"pod_priority_class_name,omitempty"`
|
||||
ShmVolume *bool `json:"enableShmVolume,omitempty"`
|
||||
}
|
||||
|
||||
// +k8s:deepcopy-gen:interfaces=k8s.io/apimachinery/pkg/runtime.Object
|
||||
|
||||
// PostgresqlList defines a list of PostgreSQL clusters.
|
||||
type PostgresqlList struct {
|
||||
metav1.TypeMeta `json:",inline"`
|
||||
@@ -90,18 +93,19 @@ type ResourceDescription struct {
|
||||
|
||||
// Resources describes requests and limits for the cluster resouces.
|
||||
type Resources struct {
|
||||
ResourceRequest ResourceDescription `json:"requests,omitempty"`
|
||||
ResourceLimits ResourceDescription `json:"limits,omitempty"`
|
||||
ResourceRequests ResourceDescription `json:"requests,omitempty"`
|
||||
ResourceLimits ResourceDescription `json:"limits,omitempty"`
|
||||
}
|
||||
|
||||
// Patroni contains Patroni-specific configuration
|
||||
type Patroni struct {
|
||||
InitDB map[string]string `json:"initdb"`
|
||||
PgHba []string `json:"pg_hba"`
|
||||
TTL uint32 `json:"ttl"`
|
||||
LoopWait uint32 `json:"loop_wait"`
|
||||
RetryTimeout uint32 `json:"retry_timeout"`
|
||||
MaximumLagOnFailover float32 `json:"maximum_lag_on_failover"` // float32 because https://github.com/kubernetes/kubernetes/issues/30213
|
||||
InitDB map[string]string `json:"initdb"`
|
||||
PgHba []string `json:"pg_hba"`
|
||||
TTL uint32 `json:"ttl"`
|
||||
LoopWait uint32 `json:"loop_wait"`
|
||||
RetryTimeout uint32 `json:"retry_timeout"`
|
||||
MaximumLagOnFailover float32 `json:"maximum_lag_on_failover"` // float32 because https://github.com/kubernetes/kubernetes/issues/30213
|
||||
Slots map[string]map[string]string `json:"slots"`
|
||||
}
|
||||
|
||||
// CloneDescription describes which cluster the new should clone and up to which point in time
|
||||
|
||||
@@ -8,15 +8,20 @@ import (
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/apis/acid.zalan.do"
|
||||
)
|
||||
|
||||
// APIVersion of the `postgresql` and `operator` CRDs
|
||||
const (
|
||||
APIVersion = "v1"
|
||||
)
|
||||
|
||||
var (
|
||||
// localSchemeBuilder and AddToScheme will stay in k8s.io/kubernetes.
|
||||
|
||||
// An instance of runtime.SchemeBuilder, global for this package
|
||||
SchemeBuilder runtime.SchemeBuilder
|
||||
localSchemeBuilder = &SchemeBuilder
|
||||
AddToScheme = localSchemeBuilder.AddToScheme
|
||||
//AddToScheme is localSchemeBuilder.AddToScheme
|
||||
AddToScheme = localSchemeBuilder.AddToScheme
|
||||
//SchemeGroupVersion has GroupName and APIVersion
|
||||
SchemeGroupVersion = schema.GroupVersion{Group: acidzalando.GroupName, Version: APIVersion}
|
||||
)
|
||||
|
||||
|
||||
@@ -14,6 +14,7 @@ var (
|
||||
serviceNameRegex = regexp.MustCompile(serviceNameRegexString)
|
||||
)
|
||||
|
||||
// Clone convenience wrapper around DeepCopy
|
||||
func (p *Postgresql) Clone() *Postgresql {
|
||||
if p == nil {
|
||||
return nil
|
||||
@@ -83,6 +84,7 @@ func validateCloneClusterDescription(clone *CloneDescription) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Success of the current Status
|
||||
func (status PostgresStatus) Success() bool {
|
||||
return status != ClusterStatusAddFailed &&
|
||||
status != ClusterStatusUpdateFailed &&
|
||||
|
||||
@@ -132,7 +132,7 @@ var unmarshalCluster = []struct {
|
||||
// This error message can vary between Go versions, so compute it for the current version.
|
||||
Error: json.Unmarshal([]byte(`{"teamId": 0}`), &PostgresSpec{}).Error(),
|
||||
},
|
||||
[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{}},"status":"Invalid"}`), nil},
|
||||
[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0,"slots":null},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{}},"status":"Invalid"}`), nil},
|
||||
{[]byte(`{
|
||||
"kind": "Postgresql",
|
||||
"apiVersion": "acid.zalan.do/v1",
|
||||
@@ -189,7 +189,14 @@ var unmarshalCluster = []struct {
|
||||
"ttl": 30,
|
||||
"loop_wait": 10,
|
||||
"retry_timeout": 10,
|
||||
"maximum_lag_on_failover": 33554432
|
||||
"maximum_lag_on_failover": 33554432,
|
||||
"slots" : {
|
||||
"permanent_logical_1" : {
|
||||
"type" : "logical",
|
||||
"database" : "foo",
|
||||
"plugin" : "pgoutput"
|
||||
}
|
||||
}
|
||||
},
|
||||
"maintenanceWindows": [
|
||||
"Mon:01:00-06:00",
|
||||
@@ -230,10 +237,11 @@ var unmarshalCluster = []struct {
|
||||
LoopWait: 10,
|
||||
RetryTimeout: 10,
|
||||
MaximumLagOnFailover: 33554432,
|
||||
Slots: map[string]map[string]string{"permanent_logical_1": {"type": "logical", "database": "foo", "plugin": "pgoutput"}},
|
||||
},
|
||||
Resources: Resources{
|
||||
ResourceRequest: ResourceDescription{CPU: "10m", Memory: "50Mi"},
|
||||
ResourceLimits: ResourceDescription{CPU: "300m", Memory: "3000Mi"},
|
||||
ResourceRequests: ResourceDescription{CPU: "10m", Memory: "50Mi"},
|
||||
ResourceLimits: ResourceDescription{CPU: "300m", Memory: "3000Mi"},
|
||||
},
|
||||
|
||||
TeamID: "ACID",
|
||||
@@ -265,7 +273,7 @@ var unmarshalCluster = []struct {
|
||||
},
|
||||
Error: "",
|
||||
},
|
||||
[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"9.6","parameters":{"log_statement":"all","max_connections":"10","shared_buffers":"32MB"}},"volume":{"size":"5Gi","storageClass":"SSD"},"patroni":{"initdb":{"data-checksums":"true","encoding":"UTF8","locale":"en_US.UTF-8"},"pg_hba":["hostssl all all 0.0.0.0/0 md5","host all all 0.0.0.0/0 md5"],"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432},"resources":{"requests":{"cpu":"10m","memory":"50Mi"},"limits":{"cpu":"300m","memory":"3000Mi"}},"teamId":"ACID","allowedSourceRanges":["127.0.0.1/32"],"numberOfInstances":2,"users":{"zalando":["superuser","createdb"]},"maintenanceWindows":["Mon:01:00-06:00","Sat:00:00-04:00","05:00-05:15"],"clone":{"cluster":"acid-batman"}}}`), nil},
|
||||
[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"9.6","parameters":{"log_statement":"all","max_connections":"10","shared_buffers":"32MB"}},"volume":{"size":"5Gi","storageClass":"SSD"},"patroni":{"initdb":{"data-checksums":"true","encoding":"UTF8","locale":"en_US.UTF-8"},"pg_hba":["hostssl all all 0.0.0.0/0 md5","host all all 0.0.0.0/0 md5"],"ttl":30,"loop_wait":10,"retry_timeout":10,"maximum_lag_on_failover":33554432,"slots":{"permanent_logical_1":{"database":"foo","plugin":"pgoutput","type":"logical"}}},"resources":{"requests":{"cpu":"10m","memory":"50Mi"},"limits":{"cpu":"300m","memory":"3000Mi"}},"teamId":"ACID","allowedSourceRanges":["127.0.0.1/32"],"numberOfInstances":2,"users":{"zalando":["superuser","createdb"]},"maintenanceWindows":["Mon:01:00-06:00","Sat:00:00-04:00","05:00-05:15"],"clone":{"cluster":"acid-batman"}}}`), nil},
|
||||
{
|
||||
[]byte(`{"kind": "Postgresql","apiVersion": "acid.zalan.do/v1","metadata": {"name": "teapot-testcluster1"}, "spec": {"teamId": "acid"}}`),
|
||||
Postgresql{
|
||||
@@ -280,7 +288,7 @@ var unmarshalCluster = []struct {
|
||||
Status: ClusterStatusInvalid,
|
||||
Error: errors.New("name must match {TEAM}-{NAME} format").Error(),
|
||||
},
|
||||
[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"teapot-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"acid","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{}},"status":"Invalid"}`), nil},
|
||||
[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"teapot-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0,"slots":null},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"acid","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{}},"status":"Invalid"}`), nil},
|
||||
{
|
||||
in: []byte(`{"kind": "Postgresql","apiVersion": "acid.zalan.do/v1","metadata": {"name": "acid-testcluster1"}, "spec": {"teamId": "acid", "clone": {"cluster": "team-batman"}}}`),
|
||||
out: Postgresql{
|
||||
@@ -300,12 +308,12 @@ var unmarshalCluster = []struct {
|
||||
},
|
||||
Error: "",
|
||||
},
|
||||
marshal: []byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"acid","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{"cluster":"team-batman"}}}`), err: nil},
|
||||
marshal: []byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster1","creationTimestamp":null},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0,"slots":null},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"acid","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{"cluster":"team-batman"}}}`), err: nil},
|
||||
{[]byte(`{"kind": "Postgresql","apiVersion": "acid.zalan.do/v1"`),
|
||||
Postgresql{},
|
||||
[]byte{},
|
||||
errors.New("unexpected end of JSON input")},
|
||||
{[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster","creationTimestamp":qaz},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"acid","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{}},"status":"Invalid"}`),
|
||||
{[]byte(`{"kind":"Postgresql","apiVersion":"acid.zalan.do/v1","metadata":{"name":"acid-testcluster","creationTimestamp":qaz},"spec":{"postgresql":{"version":"","parameters":null},"volume":{"size":"","storageClass":""},"patroni":{"initdb":null,"pg_hba":null,"ttl":0,"loop_wait":0,"retry_timeout":0,"maximum_lag_on_failover":0,"slots":null},"resources":{"requests":{"cpu":"","memory":""},"limits":{"cpu":"","memory":""}},"teamId":"acid","allowedSourceRanges":null,"numberOfInstances":0,"users":null,"clone":{}},"status":"Invalid"}`),
|
||||
Postgresql{},
|
||||
[]byte{},
|
||||
errors.New("invalid character 'q' looking for beginning of value")}}
|
||||
@@ -491,7 +499,7 @@ func TestMarshal(t *testing.T) {
|
||||
t.Errorf("Marshal error: %v", err)
|
||||
}
|
||||
if !bytes.Equal(m, tt.marshal) {
|
||||
t.Errorf("Marshal Postgresql expected: %q, got: %q", string(tt.marshal), string(m))
|
||||
t.Errorf("Marshal Postgresql \nexpected: %q, \ngot: %q", string(tt.marshal), string(m))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -499,11 +507,11 @@ func TestMarshal(t *testing.T) {
|
||||
func TestPostgresMeta(t *testing.T) {
|
||||
for _, tt := range unmarshalCluster {
|
||||
if a := tt.out.GetObjectKind(); a != &tt.out.TypeMeta {
|
||||
t.Errorf("GetObjectKindMeta expected: %v, got: %v", tt.out.TypeMeta, a)
|
||||
t.Errorf("GetObjectKindMeta \nexpected: %v, \ngot: %v", tt.out.TypeMeta, a)
|
||||
}
|
||||
|
||||
if a := tt.out.GetObjectMeta(); reflect.DeepEqual(a, tt.out.ObjectMeta) {
|
||||
t.Errorf("GetObjectMeta expected: %v, got: %v", tt.out.ObjectMeta, a)
|
||||
t.Errorf("GetObjectMeta \nexpected: %v, \ngot: %v", tt.out.ObjectMeta, a)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -320,6 +320,23 @@ func (in *Patroni) DeepCopyInto(out *Patroni) {
|
||||
*out = make([]string, len(*in))
|
||||
copy(*out, *in)
|
||||
}
|
||||
if in.Slots != nil {
|
||||
in, out := &in.Slots, &out.Slots
|
||||
*out = make(map[string]map[string]string, len(*in))
|
||||
for key, val := range *in {
|
||||
var outVal map[string]string
|
||||
if val == nil {
|
||||
(*out)[key] = nil
|
||||
} else {
|
||||
in, out := &val, &outVal
|
||||
*out = make(map[string]string, len(*in))
|
||||
for key, val := range *in {
|
||||
(*out)[key] = val
|
||||
}
|
||||
}
|
||||
(*out)[key] = outVal
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
@@ -556,7 +573,7 @@ func (in *ResourceDescription) DeepCopy() *ResourceDescription {
|
||||
// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil.
|
||||
func (in *Resources) DeepCopyInto(out *Resources) {
|
||||
*out = *in
|
||||
out.ResourceRequest = in.ResourceRequest
|
||||
out.ResourceRequests = in.ResourceRequests
|
||||
out.ResourceLimits = in.ResourceLimits
|
||||
return
|
||||
}
|
||||
|
||||
@@ -48,11 +48,22 @@ type Server struct {
|
||||
controller controllerInformer
|
||||
}
|
||||
|
||||
const (
|
||||
teamRe = `(?P<team>[a-zA-Z][a-zA-Z0-9\-_]*)`
|
||||
namespaceRe = `(?P<namespace>[a-z0-9]([-a-z0-9\-_]*[a-z0-9])?)`
|
||||
clusterRe = `(?P<cluster>[a-zA-Z][a-zA-Z0-9\-_]*)`
|
||||
)
|
||||
|
||||
var (
|
||||
clusterStatusURL = regexp.MustCompile(`^/clusters/(?P<team>[a-zA-Z][a-zA-Z0-9]*)/(?P<namespace>[a-z0-9]([-a-z0-9]*[a-z0-9])?)/(?P<cluster>[a-zA-Z][a-zA-Z0-9-]*)/?$`)
|
||||
clusterLogsURL = regexp.MustCompile(`^/clusters/(?P<team>[a-zA-Z][a-zA-Z0-9]*)/(?P<namespace>[a-z0-9]([-a-z0-9]*[a-z0-9])?)/(?P<cluster>[a-zA-Z][a-zA-Z0-9-]*)/logs/?$`)
|
||||
clusterHistoryURL = regexp.MustCompile(`^/clusters/(?P<team>[a-zA-Z][a-zA-Z0-9]*)/(?P<namespace>[a-z0-9]([-a-z0-9]*[a-z0-9])?)/(?P<cluster>[a-zA-Z][a-zA-Z0-9-]*)/history/?$`)
|
||||
teamURL = regexp.MustCompile(`^/clusters/(?P<team>[a-zA-Z][a-zA-Z0-9]*)/?$`)
|
||||
clusterStatusRe = fmt.Sprintf(`^/clusters/%s/%s/%s/?$`, teamRe, namespaceRe, clusterRe)
|
||||
clusterLogsRe = fmt.Sprintf(`^/clusters/%s/%s/%s/logs/?$`, teamRe, namespaceRe, clusterRe)
|
||||
clusterHistoryRe = fmt.Sprintf(`^/clusters/%s/%s/%s/history/?$`, teamRe, namespaceRe, clusterRe)
|
||||
teamURLRe = fmt.Sprintf(`^/clusters/%s/?$`, teamRe)
|
||||
|
||||
clusterStatusURL = regexp.MustCompile(clusterStatusRe)
|
||||
clusterLogsURL = regexp.MustCompile(clusterLogsRe)
|
||||
clusterHistoryURL = regexp.MustCompile(clusterHistoryRe)
|
||||
teamURL = regexp.MustCompile(teamURLRe)
|
||||
workerLogsURL = regexp.MustCompile(`^/workers/(?P<id>\d+)/logs/?$`)
|
||||
workerEventsQueueURL = regexp.MustCompile(`^/workers/(?P<id>\d+)/queue/?$`)
|
||||
workerStatusURL = regexp.MustCompile(`^/workers/(?P<id>\d+)/status/?$`)
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
package apiserver
|
||||
|
||||
import (
|
||||
"testing"
|
||||
)
|
||||
|
||||
const (
|
||||
clusterStatusTest = "/clusters/test-id/test_namespace/testcluster/"
|
||||
clusterStatusNumericTest = "/clusters/test-id-1/test_namespace/testcluster/"
|
||||
clusterLogsTest = "/clusters/test-id/test_namespace/testcluster/logs/"
|
||||
teamTest = "/clusters/test-id/"
|
||||
)
|
||||
|
||||
func TestUrlRegexps(t *testing.T) {
|
||||
if clusterStatusURL.FindStringSubmatch(clusterStatusTest) == nil {
|
||||
t.Errorf("clusterStatusURL can't match %s", clusterStatusTest)
|
||||
}
|
||||
|
||||
if clusterStatusURL.FindStringSubmatch(clusterStatusNumericTest) == nil {
|
||||
t.Errorf("clusterStatusURL can't match %s", clusterStatusNumericTest)
|
||||
}
|
||||
|
||||
if clusterLogsURL.FindStringSubmatch(clusterLogsTest) == nil {
|
||||
t.Errorf("clusterLogsURL can't match %s", clusterLogsTest)
|
||||
}
|
||||
|
||||
if teamURL.FindStringSubmatch(teamTest) == nil {
|
||||
t.Errorf("teamURL can't match %s", teamTest)
|
||||
}
|
||||
}
|
||||
+25
-19
@@ -12,7 +12,7 @@ import (
|
||||
|
||||
"github.com/sirupsen/logrus"
|
||||
"k8s.io/api/apps/v1beta1"
|
||||
"k8s.io/api/core/v1"
|
||||
v1 "k8s.io/api/core/v1"
|
||||
policybeta1 "k8s.io/api/policy/v1beta1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
@@ -367,7 +367,9 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *v1beta1.StatefulSet) *comp
|
||||
needsRollUpdate = true
|
||||
reasons = append(reasons, "new statefulset's container specification doesn't match the current one")
|
||||
} else {
|
||||
needsRollUpdate, reasons = c.compareContainers(c.Statefulset, statefulSet)
|
||||
var containerReasons []string
|
||||
needsRollUpdate, containerReasons = c.compareContainers(c.Statefulset, statefulSet)
|
||||
reasons = append(reasons, containerReasons...)
|
||||
}
|
||||
if len(c.Statefulset.Spec.Template.Spec.Containers) == 0 {
|
||||
c.logger.Warningf("statefulset %q has no container", util.NameFromMeta(c.Statefulset.ObjectMeta))
|
||||
@@ -375,7 +377,6 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *v1beta1.StatefulSet) *comp
|
||||
}
|
||||
// In the comparisons below, the needsReplace and needsRollUpdate flags are never reset, since checks fall through
|
||||
// and the combined effect of all the changes should be applied.
|
||||
// TODO: log all reasons for changing the statefulset, not just the last one.
|
||||
// TODO: make sure this is in sync with generatePodTemplate, ideally by using the same list of fields to generate
|
||||
// the template and the diff
|
||||
if c.Statefulset.Spec.Template.Spec.ServiceAccountName != statefulSet.Spec.Template.Spec.ServiceAccountName {
|
||||
@@ -386,7 +387,7 @@ func (c *Cluster) compareStatefulSetWith(statefulSet *v1beta1.StatefulSet) *comp
|
||||
if *c.Statefulset.Spec.Template.Spec.TerminationGracePeriodSeconds != *statefulSet.Spec.Template.Spec.TerminationGracePeriodSeconds {
|
||||
needsReplace = true
|
||||
needsRollUpdate = true
|
||||
reasons = append(reasons, "new statefulset's terminationGracePeriodSeconds doesn't match the current one")
|
||||
reasons = append(reasons, "new statefulset's terminationGracePeriodSeconds doesn't match the current one")
|
||||
}
|
||||
if !reflect.DeepEqual(c.Statefulset.Spec.Template.Spec.Affinity, statefulSet.Spec.Template.Spec.Affinity) {
|
||||
needsReplace = true
|
||||
@@ -462,23 +463,23 @@ func newCheck(msg string, cond containerCondition) containerCheck {
|
||||
|
||||
// compareContainers: compare containers from two stateful sets
|
||||
// and return:
|
||||
// * whether or not roll update is needed
|
||||
// * whether or not a rolling update is needed
|
||||
// * a list of reasons in a human readable format
|
||||
func (c *Cluster) compareContainers(setA, setB *v1beta1.StatefulSet) (bool, []string) {
|
||||
reasons := make([]string, 0)
|
||||
needsRollUpdate := false
|
||||
checks := []containerCheck{
|
||||
newCheck("new statefulset's container %d name doesn't match the current one",
|
||||
newCheck("new statefulset's container %s (index %d) name doesn't match the current one",
|
||||
func(a, b v1.Container) bool { return a.Name != b.Name }),
|
||||
newCheck("new statefulset's container %d image doesn't match the current one",
|
||||
newCheck("new statefulset's container %s (index %d) image doesn't match the current one",
|
||||
func(a, b v1.Container) bool { return a.Image != b.Image }),
|
||||
newCheck("new statefulset's container %d ports don't match the current one",
|
||||
newCheck("new statefulset's container %s (index %d) ports don't match the current one",
|
||||
func(a, b v1.Container) bool { return !reflect.DeepEqual(a.Ports, b.Ports) }),
|
||||
newCheck("new statefulset's container %d resources don't match the current ones",
|
||||
newCheck("new statefulset's container %s (index %d) resources don't match the current ones",
|
||||
func(a, b v1.Container) bool { return !compareResources(&a.Resources, &b.Resources) }),
|
||||
newCheck("new statefulset's container %d environment doesn't match the current one",
|
||||
newCheck("new statefulset's container %s (index %d) environment doesn't match the current one",
|
||||
func(a, b v1.Container) bool { return !reflect.DeepEqual(a.Env, b.Env) }),
|
||||
newCheck("new statefulset's container %d environment sources don't match the current one",
|
||||
newCheck("new statefulset's container %s (index %d) environment sources don't match the current one",
|
||||
func(a, b v1.Container) bool { return !reflect.DeepEqual(a.EnvFrom, b.EnvFrom) }),
|
||||
}
|
||||
|
||||
@@ -487,7 +488,7 @@ func (c *Cluster) compareContainers(setA, setB *v1beta1.StatefulSet) (bool, []st
|
||||
for _, check := range checks {
|
||||
if check.condition(containerA, containerB) {
|
||||
needsRollUpdate = true
|
||||
reasons = append(reasons, fmt.Sprintf(check.reason, index))
|
||||
reasons = append(reasons, fmt.Sprintf(check.reason, containerA.Name, index))
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -754,11 +755,16 @@ func (c *Cluster) initRobotUsers() error {
|
||||
if err != nil {
|
||||
return fmt.Errorf("invalid flags for user %q: %v", username, err)
|
||||
}
|
||||
adminRole := ""
|
||||
if c.OpConfig.EnableAdminRoleForUsers {
|
||||
adminRole = c.OpConfig.TeamAdminRole
|
||||
}
|
||||
newRole := spec.PgUser{
|
||||
Origin: spec.RoleOriginManifest,
|
||||
Name: username,
|
||||
Password: util.RandomPassword(constants.PasswordLength),
|
||||
Flags: flags,
|
||||
Origin: spec.RoleOriginManifest,
|
||||
Name: username,
|
||||
Password: util.RandomPassword(constants.PasswordLength),
|
||||
Flags: flags,
|
||||
AdminRole: adminRole,
|
||||
}
|
||||
if currentRole, present := c.pgUsers[username]; present {
|
||||
c.pgUsers[username] = c.resolveNameConflict(¤tRole, &newRole)
|
||||
@@ -917,7 +923,7 @@ func (c *Cluster) GetStatus() *ClusterStatus {
|
||||
func (c *Cluster) Switchover(curMaster *v1.Pod, candidate spec.NamespacedName) error {
|
||||
|
||||
var err error
|
||||
c.logger.Debugf("failing over from %q to %q", curMaster.Name, candidate)
|
||||
c.logger.Debugf("switching over from %q to %q", curMaster.Name, candidate)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
|
||||
@@ -943,12 +949,12 @@ func (c *Cluster) Switchover(curMaster *v1.Pod, candidate spec.NamespacedName) e
|
||||
}()
|
||||
|
||||
if err = c.patroni.Switchover(curMaster, candidate.Name); err == nil {
|
||||
c.logger.Debugf("successfully failed over from %q to %q", curMaster.Name, candidate)
|
||||
c.logger.Debugf("successfully switched over from %q to %q", curMaster.Name, candidate)
|
||||
if err = <-podLabelErr; err != nil {
|
||||
err = fmt.Errorf("could not get master pod label: %v", err)
|
||||
}
|
||||
} else {
|
||||
err = fmt.Errorf("could not failover: %v", err)
|
||||
err = fmt.Errorf("could not switch over: %v", err)
|
||||
}
|
||||
|
||||
// signal the role label waiting goroutine to close the shop and go home
|
||||
|
||||
+145
-25
@@ -18,6 +18,7 @@ import (
|
||||
acidv1 "github.com/zalando-incubator/postgres-operator/pkg/apis/acid.zalan.do/v1"
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/spec"
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/util"
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/util/config"
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/util/constants"
|
||||
"k8s.io/apimachinery/pkg/labels"
|
||||
)
|
||||
@@ -36,11 +37,12 @@ type pgUser struct {
|
||||
}
|
||||
|
||||
type patroniDCS struct {
|
||||
TTL uint32 `json:"ttl,omitempty"`
|
||||
LoopWait uint32 `json:"loop_wait,omitempty"`
|
||||
RetryTimeout uint32 `json:"retry_timeout,omitempty"`
|
||||
MaximumLagOnFailover float32 `json:"maximum_lag_on_failover,omitempty"`
|
||||
PGBootstrapConfiguration map[string]interface{} `json:"postgresql,omitempty"`
|
||||
TTL uint32 `json:"ttl,omitempty"`
|
||||
LoopWait uint32 `json:"loop_wait,omitempty"`
|
||||
RetryTimeout uint32 `json:"retry_timeout,omitempty"`
|
||||
MaximumLagOnFailover float32 `json:"maximum_lag_on_failover,omitempty"`
|
||||
PGBootstrapConfiguration map[string]interface{} `json:"postgresql,omitempty"`
|
||||
Slots map[string]map[string]string `json:"slots,omitempty"`
|
||||
}
|
||||
|
||||
type pgBootstrap struct {
|
||||
@@ -91,18 +93,18 @@ func (c *Cluster) makeDefaultResources() acidv1.Resources {
|
||||
defaultRequests := acidv1.ResourceDescription{CPU: config.DefaultCPURequest, Memory: config.DefaultMemoryRequest}
|
||||
defaultLimits := acidv1.ResourceDescription{CPU: config.DefaultCPULimit, Memory: config.DefaultMemoryLimit}
|
||||
|
||||
return acidv1.Resources{ResourceRequest: defaultRequests, ResourceLimits: defaultLimits}
|
||||
return acidv1.Resources{ResourceRequests: defaultRequests, ResourceLimits: defaultLimits}
|
||||
}
|
||||
|
||||
func generateResourceRequirements(resources acidv1.Resources, defaultResources acidv1.Resources) (*v1.ResourceRequirements, error) {
|
||||
var err error
|
||||
|
||||
specRequests := resources.ResourceRequest
|
||||
specRequests := resources.ResourceRequests
|
||||
specLimits := resources.ResourceLimits
|
||||
|
||||
result := v1.ResourceRequirements{}
|
||||
|
||||
result.Requests, err = fillResourceList(specRequests, defaultResources.ResourceRequest)
|
||||
result.Requests, err = fillResourceList(specRequests, defaultResources.ResourceRequests)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("could not fill resource requests: %v", err)
|
||||
}
|
||||
@@ -215,6 +217,9 @@ PatroniInitDBParams:
|
||||
if patroni.TTL != 0 {
|
||||
config.Bootstrap.DCS.TTL = patroni.TTL
|
||||
}
|
||||
if patroni.Slots != nil {
|
||||
config.Bootstrap.DCS.Slots = patroni.Slots
|
||||
}
|
||||
|
||||
config.PgLocalConfiguration = make(map[string]interface{})
|
||||
config.PgLocalConfiguration[patroniPGBinariesParameterName] = fmt.Sprintf(pgBinariesLocationTemplate, pg.PgVersion)
|
||||
@@ -334,7 +339,6 @@ func generateSpiloContainer(
|
||||
envVars []v1.EnvVar,
|
||||
volumeMounts []v1.VolumeMount,
|
||||
) *v1.Container {
|
||||
|
||||
privilegedMode := true
|
||||
return &v1.Container{
|
||||
Name: name,
|
||||
@@ -373,8 +377,8 @@ func generateSidecarContainers(sidecars []acidv1.Sidecar,
|
||||
|
||||
resources, err := generateResourceRequirements(
|
||||
makeResources(
|
||||
sidecar.Resources.ResourceRequest.CPU,
|
||||
sidecar.Resources.ResourceRequest.Memory,
|
||||
sidecar.Resources.ResourceRequests.CPU,
|
||||
sidecar.Resources.ResourceRequests.Memory,
|
||||
sidecar.Resources.ResourceLimits.CPU,
|
||||
sidecar.Resources.ResourceLimits.Memory,
|
||||
),
|
||||
@@ -392,6 +396,16 @@ func generateSidecarContainers(sidecars []acidv1.Sidecar,
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// Check whether or not we're requested to mount an shm volume,
|
||||
// taking into account that PostgreSQL manifest has precedence.
|
||||
func mountShmVolumeNeeded(opConfig config.Config, pgSpec *acidv1.PostgresSpec) bool {
|
||||
if pgSpec.ShmVolume != nil {
|
||||
return *pgSpec.ShmVolume
|
||||
}
|
||||
|
||||
return opConfig.ShmVolume
|
||||
}
|
||||
|
||||
func generatePodTemplate(
|
||||
namespace string,
|
||||
labels labels.Set,
|
||||
@@ -403,6 +417,7 @@ func generatePodTemplate(
|
||||
podServiceAccountName string,
|
||||
kubeIAMRole string,
|
||||
priorityClassName string,
|
||||
shmVolume bool,
|
||||
) (*v1.PodTemplateSpec, error) {
|
||||
|
||||
terminateGracePeriodSeconds := terminateGracePeriod
|
||||
@@ -416,6 +431,10 @@ func generatePodTemplate(
|
||||
Tolerations: *tolerationsSpec,
|
||||
}
|
||||
|
||||
if shmVolume {
|
||||
addShmVolume(&podSpec)
|
||||
}
|
||||
|
||||
if nodeAffinity != nil {
|
||||
podSpec.Affinity = nodeAffinity
|
||||
}
|
||||
@@ -471,6 +490,18 @@ func (c *Cluster) generateSpiloPodEnvVars(uid types.UID, spiloConfiguration stri
|
||||
Name: "PGUSER_SUPERUSER",
|
||||
Value: c.OpConfig.SuperUsername,
|
||||
},
|
||||
{
|
||||
Name: "KUBERNETES_SCOPE_LABEL",
|
||||
Value: c.OpConfig.ClusterNameLabel,
|
||||
},
|
||||
{
|
||||
Name: "KUBERNETES_ROLE_LABEL",
|
||||
Value: c.OpConfig.PodRoleLabel,
|
||||
},
|
||||
{
|
||||
Name: "KUBERNETES_LABELS",
|
||||
Value: labels.Set(c.OpConfig.ClusterLabels).String(),
|
||||
},
|
||||
{
|
||||
Name: "PGPASSWORD_SUPERUSER",
|
||||
ValueFrom: &v1.EnvVarSource{
|
||||
@@ -621,7 +652,7 @@ func getBucketScopeSuffix(uid string) string {
|
||||
|
||||
func makeResources(cpuRequest, memoryRequest, cpuLimit, memoryLimit string) acidv1.Resources {
|
||||
return acidv1.Resources{
|
||||
ResourceRequest: acidv1.ResourceDescription{
|
||||
ResourceRequests: acidv1.ResourceDescription{
|
||||
CPU: cpuRequest,
|
||||
Memory: memoryRequest,
|
||||
},
|
||||
@@ -640,6 +671,61 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*v1beta1.State
|
||||
podTemplate *v1.PodTemplateSpec
|
||||
volumeClaimTemplate *v1.PersistentVolumeClaim
|
||||
)
|
||||
|
||||
// Improve me. Please.
|
||||
if c.OpConfig.SetMemoryRequestToLimit {
|
||||
|
||||
// controller adjusts the default memory request at operator startup
|
||||
|
||||
request := spec.Resources.ResourceRequests.Memory
|
||||
if request == "" {
|
||||
request = c.OpConfig.DefaultMemoryRequest
|
||||
}
|
||||
|
||||
limit := spec.Resources.ResourceLimits.Memory
|
||||
if limit == "" {
|
||||
limit = c.OpConfig.DefaultMemoryLimit
|
||||
}
|
||||
|
||||
isSmaller, err := util.RequestIsSmallerThanLimit(request, limit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if isSmaller {
|
||||
c.logger.Warningf("The memory request of %v for the Postgres container is increased to match the memory limit of %v.", request, limit)
|
||||
spec.Resources.ResourceRequests.Memory = limit
|
||||
|
||||
}
|
||||
|
||||
// controller adjusts the Scalyr sidecar request at operator startup
|
||||
// as this sidecar is managed separately
|
||||
|
||||
// adjust sidecar containers defined for that particular cluster
|
||||
for _, sidecar := range spec.Sidecars {
|
||||
|
||||
// TODO #413
|
||||
sidecarRequest := sidecar.Resources.ResourceRequests.Memory
|
||||
if request == "" {
|
||||
request = c.OpConfig.DefaultMemoryRequest
|
||||
}
|
||||
|
||||
sidecarLimit := sidecar.Resources.ResourceLimits.Memory
|
||||
if limit == "" {
|
||||
limit = c.OpConfig.DefaultMemoryLimit
|
||||
}
|
||||
|
||||
isSmaller, err := util.RequestIsSmallerThanLimit(sidecarRequest, sidecarLimit)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if isSmaller {
|
||||
c.logger.Warningf("The memory request of %v for the %v sidecar container is increased to match the memory limit of %v.", sidecar.Resources.ResourceRequests.Memory, sidecar.Name, sidecar.Resources.ResourceLimits.Memory)
|
||||
sidecar.Resources.ResourceRequests.Memory = sidecar.Resources.ResourceLimits.Memory
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
defaultResources := c.makeDefaultResources()
|
||||
|
||||
resourceRequirements, err := generateResourceRequirements(spec.Resources, defaultResources)
|
||||
@@ -666,8 +752,8 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*v1beta1.State
|
||||
|
||||
// generate environment variables for the spilo container
|
||||
spiloEnvVars := deduplicateEnvVars(
|
||||
c.generateSpiloPodEnvVars(c.Postgresql.GetUID(), spiloConfiguration, &spec.Clone, customPodEnvVarsList),
|
||||
c.containerName(), c.logger)
|
||||
c.generateSpiloPodEnvVars(c.Postgresql.GetUID(), spiloConfiguration, &spec.Clone,
|
||||
customPodEnvVarsList), c.containerName(), c.logger)
|
||||
|
||||
// pickup the docker image for the spilo container
|
||||
effectiveDockerImage := util.Coalesce(spec.DockerImage, c.OpConfig.DockerImage)
|
||||
@@ -675,9 +761,15 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*v1beta1.State
|
||||
volumeMounts := generateVolumeMounts()
|
||||
|
||||
// generate the spilo container
|
||||
spiloContainer := generateSpiloContainer(c.containerName(), &effectiveDockerImage, resourceRequirements, spiloEnvVars, volumeMounts)
|
||||
c.logger.Debugf("Generating Spilo container, environment variables: %v", spiloEnvVars)
|
||||
spiloContainer := generateSpiloContainer(c.containerName(),
|
||||
&effectiveDockerImage,
|
||||
resourceRequirements,
|
||||
spiloEnvVars,
|
||||
volumeMounts,
|
||||
)
|
||||
|
||||
// resolve conflicts between operator-global and per-cluster sidecards
|
||||
// resolve conflicts between operator-global and per-cluster sidecars
|
||||
sideCars := c.mergeSidecars(spec.Sidecars)
|
||||
|
||||
resourceRequirementsScalyrSidecar := makeResources(
|
||||
@@ -706,7 +798,7 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*v1beta1.State
|
||||
tolerationSpec := tolerations(&spec.Tolerations, c.OpConfig.PodToleration)
|
||||
effectivePodPriorityClassName := util.Coalesce(spec.PodPriorityClassName, c.OpConfig.PodPriorityClassName)
|
||||
|
||||
// generate pod template for the statefulset, based on the spilo container and sidecards
|
||||
// generate pod template for the statefulset, based on the spilo container and sidecars
|
||||
if podTemplate, err = generatePodTemplate(
|
||||
c.Namespace,
|
||||
c.labelsSet(true),
|
||||
@@ -717,7 +809,8 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*v1beta1.State
|
||||
int64(c.OpConfig.PodTerminateGracePeriod.Seconds()),
|
||||
c.OpConfig.PodServiceAccountName,
|
||||
c.OpConfig.KubeIAMRole,
|
||||
effectivePodPriorityClassName); err != nil {
|
||||
effectivePodPriorityClassName,
|
||||
mountShmVolumeNeeded(c.OpConfig, spec)); err != nil {
|
||||
return nil, fmt.Errorf("could not generate pod template: %v", err)
|
||||
}
|
||||
|
||||
@@ -824,6 +917,32 @@ func (c *Cluster) getNumberOfInstances(spec *acidv1.PostgresSpec) int32 {
|
||||
return newcur
|
||||
}
|
||||
|
||||
// To avoid issues with limited /dev/shm inside docker environment, when
|
||||
// PostgreSQL can't allocate enough of dsa segments from it, we can
|
||||
// mount an extra memory volume
|
||||
//
|
||||
// see https://docs.okd.io/latest/dev_guide/shared_memory.html
|
||||
func addShmVolume(podSpec *v1.PodSpec) {
|
||||
volumes := append(podSpec.Volumes, v1.Volume{
|
||||
Name: constants.ShmVolumeName,
|
||||
VolumeSource: v1.VolumeSource{
|
||||
EmptyDir: &v1.EmptyDirVolumeSource{
|
||||
Medium: "Memory",
|
||||
},
|
||||
},
|
||||
})
|
||||
|
||||
pgIdx := constants.PostgresContainerIdx
|
||||
mounts := append(podSpec.Containers[pgIdx].VolumeMounts,
|
||||
v1.VolumeMount{
|
||||
Name: constants.ShmVolumeName,
|
||||
MountPath: constants.ShmVolumePath,
|
||||
})
|
||||
|
||||
podSpec.Containers[0].VolumeMounts = mounts
|
||||
podSpec.Volumes = volumes
|
||||
}
|
||||
|
||||
func generatePersistentVolumeClaimTemplate(volumeSize, volumeStorageClass string) (*v1.PersistentVolumeClaim, error) {
|
||||
|
||||
var storageClassName *string
|
||||
@@ -958,16 +1077,17 @@ func (c *Cluster) generateService(role PostgresRole, spec *acidv1.PostgresSpec)
|
||||
|
||||
if c.shouldCreateLoadBalancerForService(role, spec) {
|
||||
|
||||
// safe default value: lock load balancer to only local address unless overridden explicitly.
|
||||
sourceRanges := []string{localHost}
|
||||
|
||||
allowedSourceRanges := spec.AllowedSourceRanges
|
||||
if len(allowedSourceRanges) >= 0 {
|
||||
sourceRanges = allowedSourceRanges
|
||||
// spec.AllowedSourceRanges evaluates to the empty slice of zero length
|
||||
// when omitted or set to 'null'/empty sequence in the PG manifest
|
||||
if len(spec.AllowedSourceRanges) > 0 {
|
||||
serviceSpec.LoadBalancerSourceRanges = spec.AllowedSourceRanges
|
||||
} else {
|
||||
// safe default value: lock a load balancer only to the local address unless overridden explicitly
|
||||
serviceSpec.LoadBalancerSourceRanges = []string{localHost}
|
||||
}
|
||||
|
||||
c.logger.Debugf("final load balancer source ranges as seen in a service spec (not necessarily applied): %q", serviceSpec.LoadBalancerSourceRanges)
|
||||
serviceSpec.Type = v1.ServiceTypeLoadBalancer
|
||||
serviceSpec.LoadBalancerSourceRanges = sourceRanges
|
||||
|
||||
annotations = map[string]string{
|
||||
constants.ZalandoDNSNameAnnotation: dnsName,
|
||||
|
||||
@@ -1,8 +1,11 @@
|
||||
package cluster
|
||||
|
||||
import (
|
||||
"k8s.io/api/core/v1"
|
||||
|
||||
acidv1 "github.com/zalando-incubator/postgres-operator/pkg/apis/acid.zalan.do/v1"
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/util/config"
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/util/constants"
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/util/k8sutil"
|
||||
"testing"
|
||||
)
|
||||
@@ -75,3 +78,54 @@ func TestCreateLoadBalancerLogic(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestShmVolume(t *testing.T) {
|
||||
testName := "TestShmVolume"
|
||||
tests := []struct {
|
||||
subTest string
|
||||
podSpec *v1.PodSpec
|
||||
shmPos int
|
||||
}{
|
||||
{
|
||||
subTest: "empty PodSpec",
|
||||
podSpec: &v1.PodSpec{
|
||||
Volumes: []v1.Volume{},
|
||||
Containers: []v1.Container{
|
||||
v1.Container{
|
||||
VolumeMounts: []v1.VolumeMount{},
|
||||
},
|
||||
},
|
||||
},
|
||||
shmPos: 0,
|
||||
},
|
||||
{
|
||||
subTest: "non empty PodSpec",
|
||||
podSpec: &v1.PodSpec{
|
||||
Volumes: []v1.Volume{v1.Volume{}},
|
||||
Containers: []v1.Container{
|
||||
v1.Container{
|
||||
VolumeMounts: []v1.VolumeMount{
|
||||
v1.VolumeMount{},
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
shmPos: 1,
|
||||
},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
addShmVolume(tt.podSpec)
|
||||
|
||||
volumeName := tt.podSpec.Volumes[tt.shmPos].Name
|
||||
volumeMountName := tt.podSpec.Containers[0].VolumeMounts[tt.shmPos].Name
|
||||
|
||||
if volumeName != constants.ShmVolumeName {
|
||||
t.Errorf("%s %s: Expected volume %s was not created, have %s instead",
|
||||
testName, tt.subTest, constants.ShmVolumeName, volumeName)
|
||||
}
|
||||
if volumeMountName != constants.ShmVolumeName {
|
||||
t.Errorf("%s %s: Expected mount %s was not created, have %s instead",
|
||||
testName, tt.subTest, constants.ShmVolumeName, volumeMountName)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+15
-19
@@ -4,7 +4,7 @@ import (
|
||||
"fmt"
|
||||
"math/rand"
|
||||
|
||||
"k8s.io/api/core/v1"
|
||||
v1 "k8s.io/api/core/v1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/spec"
|
||||
@@ -77,11 +77,7 @@ func (c *Cluster) deletePod(podName spec.NamespacedName) error {
|
||||
return err
|
||||
}
|
||||
|
||||
if err := c.waitForPodDeletion(ch); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
return c.waitForPodDeletion(ch)
|
||||
}
|
||||
|
||||
func (c *Cluster) unregisterPodSubscriber(podName spec.NamespacedName) {
|
||||
@@ -122,7 +118,7 @@ func (c *Cluster) movePodFromEndOfLifeNode(pod *v1.Pod) (*v1.Pod, error) {
|
||||
if eol, err = c.podIsEndOfLife(pod); err != nil {
|
||||
return nil, fmt.Errorf("could not get node %q: %v", pod.Spec.NodeName, err)
|
||||
} else if !eol {
|
||||
c.logger.Infof("pod %q is already on a live node", podName)
|
||||
c.logger.Infof("check failed: pod %q is already on a live node", podName)
|
||||
return pod, nil
|
||||
}
|
||||
|
||||
@@ -162,7 +158,7 @@ func (c *Cluster) masterCandidate(oldNodeName string) (*v1.Pod, error) {
|
||||
}
|
||||
|
||||
if len(replicas) == 0 {
|
||||
c.logger.Warningf("no available master candidates, migration will cause longer downtime of the master instance")
|
||||
c.logger.Warningf("no available master candidates, migration will cause longer downtime of Postgres cluster")
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
@@ -193,18 +189,18 @@ func (c *Cluster) MigrateMasterPod(podName spec.NamespacedName) error {
|
||||
return fmt.Errorf("could not get pod: %v", err)
|
||||
}
|
||||
|
||||
c.logger.Infof("migrating master pod %q", podName)
|
||||
c.logger.Infof("starting process to migrate master pod %q", podName)
|
||||
|
||||
if eol, err = c.podIsEndOfLife(oldMaster); err != nil {
|
||||
return fmt.Errorf("could not get node %q: %v", oldMaster.Spec.NodeName, err)
|
||||
}
|
||||
if !eol {
|
||||
c.logger.Debugf("pod is already on a live node")
|
||||
c.logger.Debugf("no action needed: master pod is already on a live node")
|
||||
return nil
|
||||
}
|
||||
|
||||
if role := PostgresRole(oldMaster.Labels[c.OpConfig.PodRoleLabel]); role != Master {
|
||||
c.logger.Warningf("pod %q is not a master", podName)
|
||||
c.logger.Warningf("no action needed: pod %q is not the master (anymore)", podName)
|
||||
return nil
|
||||
}
|
||||
// we must have a statefulset in the cluster for the migration to work
|
||||
@@ -219,10 +215,10 @@ 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.
|
||||
if *c.Statefulset.Spec.Replicas > 1 {
|
||||
if masterCandidatePod, err = c.masterCandidate(oldMaster.Spec.NodeName); err != nil {
|
||||
return fmt.Errorf("could not get new master candidate: %v", err)
|
||||
return fmt.Errorf("could not find suitable replica pod as candidate for failover: %v", err)
|
||||
}
|
||||
} else {
|
||||
c.logger.Warningf("single master pod for cluster %q, migration will cause longer downtime of the master instance", c.clusterName())
|
||||
c.logger.Warningf("migrating single pod cluster %q, this will cause downtime of the Postgres cluster until pod is back", c.clusterName())
|
||||
}
|
||||
|
||||
// there are two cases for each postgres cluster that has its master pod on the node to migrate from:
|
||||
@@ -256,15 +252,15 @@ func (c *Cluster) MigrateReplicaPod(podName spec.NamespacedName, fromNodeName st
|
||||
return fmt.Errorf("could not get pod: %v", err)
|
||||
}
|
||||
|
||||
c.logger.Infof("migrating replica pod %q", podName)
|
||||
c.logger.Infof("migrating replica pod %q to live node", podName)
|
||||
|
||||
if replicaPod.Spec.NodeName != fromNodeName {
|
||||
c.logger.Infof("pod %q has already migrated to node %q", podName, replicaPod.Spec.NodeName)
|
||||
c.logger.Infof("check failed: pod %q has already migrated to node %q", podName, replicaPod.Spec.NodeName)
|
||||
return nil
|
||||
}
|
||||
|
||||
if role := PostgresRole(replicaPod.Labels[c.OpConfig.PodRoleLabel]); role != Replica {
|
||||
return fmt.Errorf("pod %q is not a replica", podName)
|
||||
return fmt.Errorf("check failed: pod %q is not a replica", podName)
|
||||
}
|
||||
|
||||
_, err = c.movePodFromEndOfLifeNode(replicaPod)
|
||||
@@ -296,7 +292,7 @@ func (c *Cluster) recreatePod(podName spec.NamespacedName) (*v1.Pod, error) {
|
||||
}
|
||||
|
||||
func (c *Cluster) recreatePods() error {
|
||||
c.setProcessName("recreating pods")
|
||||
c.setProcessName("starting to recreate pods")
|
||||
ls := c.labelsSet(false)
|
||||
namespace := c.Namespace
|
||||
|
||||
@@ -337,10 +333,10 @@ func (c *Cluster) recreatePods() error {
|
||||
// failover if we have not observed a master pod when re-creating former replicas.
|
||||
if newMasterPod == nil && len(replicas) > 0 {
|
||||
if err := c.Switchover(masterPod, masterCandidate(replicas)); err != nil {
|
||||
c.logger.Warningf("could not perform failover: %v", err)
|
||||
c.logger.Warningf("could not perform switch over: %v", err)
|
||||
}
|
||||
} else if newMasterPod == nil && len(replicas) == 0 {
|
||||
c.logger.Warningf("cannot switch master role before re-creating the pod: no replicas")
|
||||
c.logger.Warningf("cannot perform switch over before re-creating the pod: no replicas")
|
||||
}
|
||||
c.logger.Infof("recreating old master pod %q", util.NameFromMeta(masterPod.ObjectMeta))
|
||||
|
||||
|
||||
@@ -132,16 +132,17 @@ func (c *Cluster) preScaleDown(newStatefulSet *v1beta1.StatefulSet) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// setRollingUpdateFlagForStatefulSet sets the indicator or the rolling upgrade requirement
|
||||
// setRollingUpdateFlagForStatefulSet sets the indicator or the rolling update requirement
|
||||
// in the StatefulSet annotation.
|
||||
func (c *Cluster) setRollingUpdateFlagForStatefulSet(sset *v1beta1.StatefulSet, val bool) {
|
||||
anno := sset.GetAnnotations()
|
||||
c.logger.Debugf("rolling upgrade flag has been set to %t", val)
|
||||
if anno == nil {
|
||||
anno = make(map[string]string)
|
||||
}
|
||||
|
||||
anno[rollingUpdateStatefulsetAnnotationKey] = strconv.FormatBool(val)
|
||||
sset.SetAnnotations(anno)
|
||||
c.logger.Debugf("statefulset's rolling update annotation has been set to %t", val)
|
||||
}
|
||||
|
||||
// applyRollingUpdateFlagforStatefulSet sets the rolling update flag for the cluster's StatefulSet
|
||||
@@ -176,9 +177,9 @@ func (c *Cluster) getRollingUpdateFlagFromStatefulSet(sset *v1beta1.StatefulSet,
|
||||
return flag
|
||||
}
|
||||
|
||||
// mergeRollingUpdateFlagUsingCache return the value of the rollingUpdate flag from the passed
|
||||
// mergeRollingUpdateFlagUsingCache returns the value of the rollingUpdate flag from the passed
|
||||
// statefulset, however, the value can be cleared if there is a cached flag in the cluster that
|
||||
// is set to false (the disrepancy could be a result of a failed StatefulSet update).s
|
||||
// is set to false (the discrepancy could be a result of a failed StatefulSet update)
|
||||
func (c *Cluster) mergeRollingUpdateFlagUsingCache(runningStatefulSet *v1beta1.StatefulSet) bool {
|
||||
var (
|
||||
cachedStatefulsetExists, clearRollingUpdateFromCache, podsRollingUpdateRequired bool
|
||||
@@ -198,7 +199,7 @@ func (c *Cluster) mergeRollingUpdateFlagUsingCache(runningStatefulSet *v1beta1.S
|
||||
c.logger.Infof("clearing the rolling update flag based on the cached information")
|
||||
podsRollingUpdateRequired = false
|
||||
} else {
|
||||
c.logger.Infof("found a statefulset with an unfinished pods rolling update")
|
||||
c.logger.Infof("found a statefulset with an unfinished rolling update of the pods")
|
||||
|
||||
}
|
||||
}
|
||||
@@ -436,7 +437,11 @@ func (c *Cluster) updateService(role PostgresRole, newService *v1.Service) error
|
||||
func (c *Cluster) deleteService(role PostgresRole) error {
|
||||
c.logger.Debugf("deleting service %s", role)
|
||||
|
||||
service := c.Services[role]
|
||||
service, ok := c.Services[role]
|
||||
if !ok {
|
||||
c.logger.Debugf("No service for %s role was found, nothing to delete", role)
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := c.KubeClient.Services(service.Namespace).Delete(service.Name, c.deleteOptions); err != nil {
|
||||
return err
|
||||
|
||||
@@ -2,6 +2,7 @@ package cluster
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"k8s.io/api/core/v1"
|
||||
policybeta1 "k8s.io/api/policy/v1beta1"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
@@ -280,6 +281,7 @@ func (c *Cluster) syncStatefulSet() error {
|
||||
podsRollingUpdateRequired = true
|
||||
c.setRollingUpdateFlagForStatefulSet(desiredSS, podsRollingUpdateRequired)
|
||||
}
|
||||
|
||||
c.logStatefulSetChanges(c.Statefulset, desiredSS, false, cmp.reasons)
|
||||
|
||||
if !cmp.replace {
|
||||
|
||||
@@ -1,12 +1,13 @@
|
||||
package cluster
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
acidv1 "github.com/zalando-incubator/postgres-operator/pkg/apis/acid.zalan.do/v1"
|
||||
"k8s.io/api/apps/v1beta1"
|
||||
"k8s.io/api/core/v1"
|
||||
policybeta1 "k8s.io/api/policy/v1beta1"
|
||||
"k8s.io/apimachinery/pkg/types"
|
||||
"time"
|
||||
)
|
||||
|
||||
// PostgresRole describes role of the node
|
||||
@@ -20,6 +21,7 @@ const (
|
||||
Replica PostgresRole = "replica"
|
||||
)
|
||||
|
||||
// PodEventType represents the type of a pod-related event
|
||||
type PodEventType string
|
||||
|
||||
// Possible values for the EventType
|
||||
|
||||
+2
-1
@@ -179,7 +179,7 @@ func (c *Cluster) logStatefulSetChanges(old, new *v1beta1.StatefulSet, isUpdate
|
||||
if !reflect.DeepEqual(old.Annotations, new.Annotations) {
|
||||
c.logger.Debugf("metadata.annotation diff\n%s\n", util.PrettyDiff(old.Annotations, new.Annotations))
|
||||
}
|
||||
c.logger.Debugf("spec diff\n%s\n", util.PrettyDiff(old.Spec, new.Spec))
|
||||
c.logger.Debugf("spec diff between old and new statefulsets: \n%s\n", util.PrettyDiff(old.Spec, new.Spec))
|
||||
|
||||
if len(reasons) > 0 {
|
||||
for _, reason := range reasons {
|
||||
@@ -460,6 +460,7 @@ func (c *Cluster) setSpec(newSpec *acidv1.Postgresql) {
|
||||
c.specMu.Unlock()
|
||||
}
|
||||
|
||||
// GetSpec returns a copy of the operator-side spec of a Postgres cluster in a thread-safe manner
|
||||
func (c *Cluster) GetSpec() (*acidv1.Postgresql, error) {
|
||||
c.specMu.RLock()
|
||||
defer c.specMu.RUnlock()
|
||||
|
||||
@@ -128,6 +128,29 @@ func (c *Controller) initOperatorConfig() {
|
||||
c.opConfig = config.NewFromMap(configMapData)
|
||||
c.warnOnDeprecatedOperatorParameters()
|
||||
|
||||
if c.opConfig.SetMemoryRequestToLimit {
|
||||
|
||||
isSmaller, err := util.RequestIsSmallerThanLimit(c.opConfig.DefaultMemoryRequest, c.opConfig.DefaultMemoryLimit)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
if isSmaller {
|
||||
c.logger.Warningf("The default memory request of %v for Postgres containers is increased to match the default memory limit of %v.", c.opConfig.DefaultMemoryRequest, c.opConfig.DefaultMemoryLimit)
|
||||
c.opConfig.DefaultMemoryRequest = c.opConfig.DefaultMemoryLimit
|
||||
}
|
||||
|
||||
isSmaller, err = util.RequestIsSmallerThanLimit(c.opConfig.ScalyrMemoryRequest, c.opConfig.ScalyrMemoryLimit)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
if isSmaller {
|
||||
c.logger.Warningf("The memory request of %v for the Scalyr sidecar container is increased to match the memory limit of %v.", c.opConfig.ScalyrMemoryRequest, c.opConfig.ScalyrMemoryLimit)
|
||||
c.opConfig.ScalyrMemoryRequest = c.opConfig.ScalyrMemoryLimit
|
||||
}
|
||||
|
||||
// generateStatefulSet adjusts values for individual Postgres clusters
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (c *Controller) modifyConfigFromEnvironment() {
|
||||
|
||||
@@ -55,6 +55,7 @@ func (c *Controller) importConfigurationFromCRD(fromCRD *acidv1.OperatorConfigur
|
||||
result.DefaultMemoryRequest = fromCRD.PostgresPodResources.DefaultMemoryRequest
|
||||
result.DefaultCPULimit = fromCRD.PostgresPodResources.DefaultCPULimit
|
||||
result.DefaultMemoryLimit = fromCRD.PostgresPodResources.DefaultMemoryLimit
|
||||
result.SetMemoryRequestToLimit = fromCRD.SetMemoryRequestToLimit
|
||||
|
||||
result.ResourceCheckInterval = time.Duration(fromCRD.Timeouts.ResourceCheckInterval)
|
||||
result.ResourceCheckTimeout = time.Duration(fromCRD.Timeouts.ResourceCheckTimeout)
|
||||
|
||||
@@ -452,8 +452,14 @@ func (c *Controller) queueClusterEvent(informerOldSpec, informerNewSpec *acidv1.
|
||||
if informerOldSpec != nil { //update, delete
|
||||
uid = informerOldSpec.GetUID()
|
||||
clusterName = util.NameFromMeta(informerOldSpec.ObjectMeta)
|
||||
|
||||
// user is fixing previously incorrect spec
|
||||
if eventType == EventUpdate && informerNewSpec.Error == "" && informerOldSpec.Error != "" {
|
||||
eventType = EventSync
|
||||
}
|
||||
|
||||
// set current error to be one of the new spec if present
|
||||
if informerNewSpec != nil {
|
||||
clusterError = informerNewSpec.Error
|
||||
} else {
|
||||
clusterError = informerOldSpec.Error
|
||||
|
||||
@@ -49,6 +49,7 @@ type PgUser struct {
|
||||
Flags []string `yaml:"user_flags"`
|
||||
MemberOf []string `yaml:"inrole"`
|
||||
Parameters map[string]string `yaml:"db_parameters"`
|
||||
AdminRole string `yaml:"admin_role"`
|
||||
}
|
||||
|
||||
// PgUserMap maps user names to the definitions.
|
||||
@@ -125,6 +126,7 @@ func (n *NamespacedName) Decode(value string) error {
|
||||
return n.DecodeWorker(value, GetOperatorNamespace())
|
||||
}
|
||||
|
||||
// UnmarshalJSON converts a byte slice to NamespacedName
|
||||
func (n *NamespacedName) UnmarshalJSON(data []byte) error {
|
||||
result := NamespacedName{}
|
||||
var tmp string
|
||||
|
||||
@@ -38,6 +38,7 @@ type Resources struct {
|
||||
NodeReadinessLabel map[string]string `name:"node_readiness_label" default:""`
|
||||
MaxInstances int32 `name:"max_instances" default:"-1"`
|
||||
MinInstances int32 `name:"min_instances" default:"-1"`
|
||||
ShmVolume bool `name:"enable_shm_volume" default:"true"`
|
||||
}
|
||||
|
||||
// Auth describes authentication specific configuration parameters
|
||||
@@ -89,6 +90,7 @@ type Config struct {
|
||||
EnableTeamsAPI bool `name:"enable_teams_api" default:"true"`
|
||||
EnableTeamSuperuser bool `name:"enable_team_superuser" default:"false"`
|
||||
TeamAdminRole string `name:"team_admin_role" default:"admin"`
|
||||
EnableAdminRoleForUsers bool `name:"enable_admin_role_for_users" default:"true"`
|
||||
EnableMasterLoadBalancer bool `name:"enable_master_load_balancer" default:"true"`
|
||||
EnableReplicaLoadBalancer bool `name:"enable_replica_load_balancer" default:"false"`
|
||||
// deprecated and kept for backward compatibility
|
||||
@@ -104,6 +106,7 @@ type Config struct {
|
||||
PodTerminateGracePeriod time.Duration `name:"pod_terminate_grace_period" default:"5m"`
|
||||
ProtectedRoles []string `name:"protected_role_names" default:"admin"`
|
||||
PostgresSuperuserTeams []string `name:"postgres_superuser_teams" default:""`
|
||||
SetMemoryRequestToLimit bool `name:"set_memory_request_to_limit" defaults:"false"`
|
||||
}
|
||||
|
||||
// MustMarshal marshals the config or panics
|
||||
|
||||
@@ -19,6 +19,7 @@ type fieldInfo struct {
|
||||
Field reflect.Value
|
||||
}
|
||||
|
||||
// StringTemplate is a convenience alias
|
||||
type StringTemplate string
|
||||
|
||||
func decoderFrom(field reflect.Value) (d decoder) {
|
||||
@@ -221,12 +222,14 @@ func getMapPairsFromString(value string) (pairs []string, err error) {
|
||||
return
|
||||
}
|
||||
|
||||
// Decode cast value to StringTemplate
|
||||
func (f *StringTemplate) Decode(value string) error {
|
||||
*f = StringTemplate(value)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Format formatted string from StringTemplate
|
||||
func (f *StringTemplate) Format(a ...string) string {
|
||||
res := string(*f)
|
||||
|
||||
@@ -237,6 +240,7 @@ func (f *StringTemplate) Format(a ...string) string {
|
||||
return res
|
||||
}
|
||||
|
||||
// MarshalJSON converts a StringTemplate to byte slice
|
||||
func (f StringTemplate) MarshalJSON() ([]byte, error) {
|
||||
return json.Marshal(string(f))
|
||||
}
|
||||
|
||||
@@ -5,6 +5,7 @@ import "time"
|
||||
// General kubernetes-related constants
|
||||
const (
|
||||
PostgresContainerName = "postgres"
|
||||
PostgresContainerIdx = 0
|
||||
K8sAPIPath = "/apis"
|
||||
StatefulsetDeletionInterval = 1 * time.Second
|
||||
StatefulsetDeletionTimeout = 30 * time.Second
|
||||
|
||||
@@ -10,4 +10,7 @@ const (
|
||||
|
||||
PostgresConnectRetryTimeout = 2 * time.Minute
|
||||
PostgresConnectTimeout = 15 * time.Second
|
||||
|
||||
ShmVolumeName = "dshm"
|
||||
ShmVolumePath = "/dev/shm"
|
||||
)
|
||||
|
||||
@@ -17,8 +17,10 @@ type Ticker struct {
|
||||
ticker *time.Ticker
|
||||
}
|
||||
|
||||
// Stop is a convenience wrapper around ticker.Stop
|
||||
func (t *Ticker) Stop() { t.ticker.Stop() }
|
||||
|
||||
// Tick is a convenience wrapper around ticker.C
|
||||
func (t *Ticker) Tick() { <-t.ticker.C }
|
||||
|
||||
// Retry is a wrapper around RetryWorker that provides a real RetryTicker
|
||||
|
||||
@@ -43,6 +43,7 @@ type httpClient interface {
|
||||
Do(req *http.Request) (*http.Response, error)
|
||||
}
|
||||
|
||||
// Interface to the TeamsAPIClient
|
||||
type Interface interface {
|
||||
TeamInfo(teamID, token string) (tm *Team, err error)
|
||||
}
|
||||
|
||||
@@ -5,9 +5,10 @@ import (
|
||||
"fmt"
|
||||
"strings"
|
||||
|
||||
"reflect"
|
||||
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/spec"
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/util"
|
||||
"reflect"
|
||||
)
|
||||
|
||||
const (
|
||||
@@ -19,6 +20,7 @@ const (
|
||||
doBlockStmt = `SET LOCAL synchronous_commit = 'local'; DO $$ BEGIN %s; END;$$;`
|
||||
passwordTemplate = "ENCRYPTED PASSWORD '%s'"
|
||||
inRoleTemplate = `IN ROLE %s`
|
||||
adminTemplate = `ADMIN %s`
|
||||
)
|
||||
|
||||
// DefaultUserSyncStrategy implements a user sync strategy that merges already existing database users
|
||||
@@ -113,6 +115,9 @@ func (strategy DefaultUserSyncStrategy) createPgUser(user spec.PgUser, db *sql.D
|
||||
if len(user.MemberOf) > 0 {
|
||||
userFlags = append(userFlags, fmt.Sprintf(inRoleTemplate, quoteMemberList(user)))
|
||||
}
|
||||
if user.AdminRole != "" {
|
||||
userFlags = append(userFlags, fmt.Sprintf(adminTemplate, user.AdminRole))
|
||||
}
|
||||
|
||||
if user.Password == "" {
|
||||
userPassword = "PASSWORD NULL"
|
||||
|
||||
@@ -3,12 +3,14 @@ package util
|
||||
import (
|
||||
"crypto/md5" // #nosec we need it to for PostgreSQL md5 passwords
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"math/rand"
|
||||
"regexp"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/motomux/pretty"
|
||||
resource "k8s.io/apimachinery/pkg/api/resource"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/spec"
|
||||
@@ -127,3 +129,19 @@ func Coalesce(val, defaultVal string) string {
|
||||
}
|
||||
return val
|
||||
}
|
||||
|
||||
// RequestIsSmallerThanLimit
|
||||
func RequestIsSmallerThanLimit(requestStr, limitStr string) (bool, error) {
|
||||
|
||||
request, err := resource.ParseQuantity(requestStr)
|
||||
if err != nil {
|
||||
return false, fmt.Errorf("could not parse memory request %v : %v", requestStr, err)
|
||||
}
|
||||
|
||||
limit, err2 := resource.ParseQuantity(limitStr)
|
||||
if err2 != nil {
|
||||
return false, fmt.Errorf("could not parse memory limit %v : %v", limitStr, err2)
|
||||
}
|
||||
|
||||
return request.Cmp(limit) == -1, nil
|
||||
}
|
||||
|
||||
@@ -69,6 +69,17 @@ var substringMatch = []struct {
|
||||
{regexp.MustCompile(`aaaa (\d+) bbbb`), "aaaa 123 bbbb", nil},
|
||||
}
|
||||
|
||||
var requestIsSmallerThanLimitTests = []struct {
|
||||
request string
|
||||
limit string
|
||||
out bool
|
||||
}{
|
||||
{"1G", "2G", true},
|
||||
{"1G", "1Gi", true}, // G is 1000^3 bytes, Gi is 1024^3 bytes
|
||||
{"1024Mi", "1G", false},
|
||||
{"1e9", "1G", false}, // 1e9 bytes == 1G
|
||||
}
|
||||
|
||||
func TestRandomPassword(t *testing.T) {
|
||||
const pwdLength = 10
|
||||
pwd := RandomPassword(pwdLength)
|
||||
@@ -143,3 +154,15 @@ func TestMapContains(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestRequestIsSmallerThanLimit(t *testing.T) {
|
||||
for _, tt := range requestIsSmallerThanLimitTests {
|
||||
res, err := RequestIsSmallerThanLimit(tt.request, tt.limit)
|
||||
if err != nil {
|
||||
t.Errorf("RequestIsSmallerThanLimit returned unexpected error: %#v", err)
|
||||
}
|
||||
if res != tt.out {
|
||||
t.Errorf("RequestIsSmallerThanLimit expected: %#v, got: %#v", tt.out, res)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user