public_git/postgres-operator
1
0
Fork 0
mirror of https://github.com/zalando/postgres-operator.git synced 2026-10-08 15:32:53 +02:00
Code Issues Projects Releases Wiki Activity

Merge branch 'master' into feature/actions

Browse Source
This commit is contained in:
erthalion
2018-08-30 16:51:28 +02:00
parent 332db3bbda 25fa45fd58
commit 87eee1ed72
91 changed files with 5275 additions and 1561 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/cluster/cluster.go
+140 -128
View File
@@ -4,7 +4,6 @@ package cluster
import (
"database/sql"
"encoding/json"
"fmt"
"reflect"
"regexp"
@@ -12,14 +11,17 @@ import (
"time"
"github.com/Sirupsen/logrus"
"k8s.io/api/apps/v1beta1"
"k8s.io/api/core/v1"
policybeta1 "k8s.io/api/policy/v1beta1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/pkg/api/v1"
"k8s.io/client-go/pkg/apis/apps/v1beta1"
policybeta1 "k8s.io/client-go/pkg/apis/policy/v1beta1"
"k8s.io/client-go/rest"
"k8s.io/client-go/tools/cache"
"encoding/json"
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"
@@ -28,6 +30,7 @@ import (
"github.com/zalando-incubator/postgres-operator/pkg/util/patroni"
"github.com/zalando-incubator/postgres-operator/pkg/util/teams"
"github.com/zalando-incubator/postgres-operator/pkg/util/users"
rbacv1beta1 "k8s.io/api/rbac/v1beta1"
)
var (
@@ -39,10 +42,11 @@ var (
// Config contains operator-wide clients and configuration used from a cluster. TODO: remove struct duplication.
type Config struct {
OpConfig config.Config
RestConfig *rest.Config
InfrastructureRoles map[string]spec.PgUser // inherited from the controller
PodServiceAccount *v1.ServiceAccount
OpConfig config.Config
RestConfig *rest.Config
InfrastructureRoles map[string]spec.PgUser // inherited from the controller
PodServiceAccount *v1.ServiceAccount
PodServiceAccountRoleBinding *rbacv1beta1.RoleBinding
}
type kubeResources struct {
@@ -58,13 +62,13 @@ type kubeResources struct {
// Cluster describes postgresql cluster
type Cluster struct {
kubeResources
spec.Postgresql
acidv1.Postgresql
Config
logger *logrus.Entry
patroni patroni.Interface
pgUsers map[string]spec.PgUser
systemUsers map[string]spec.PgUser
podSubscribers map[spec.NamespacedName]chan spec.PodEvent
podSubscribers map[spec.NamespacedName]chan PodEvent
podSubscribersMu sync.RWMutex
pgDb *sql.DB
mu sync.Mutex
@@ -75,7 +79,7 @@ type Cluster struct {
teamsAPIClient teams.Interface
oauthTokenGetter OAuthTokenGetter
KubeClient k8sutil.KubernetesClient //TODO: move clients to the better place?
currentProcess spec.Process
currentProcess Process
processMu sync.RWMutex // protects the current operation for reporting, no need to hold the master mutex
specMu sync.RWMutex // protects the spec for reporting, no need to hold the master mutex
dryRunMode bool
@@ -89,11 +93,11 @@ type compareStatefulsetResult struct {
}
// New creates a new cluster. This function should be called from a controller.
func New(cfg Config, kubeClient k8sutil.KubernetesClient, pgSpec spec.Postgresql, logger *logrus.Entry) *Cluster {
orphanDependents := true
func New(cfg Config, kubeClient k8sutil.KubernetesClient, pgSpec acidv1.Postgresql, logger *logrus.Entry) *Cluster {
deletePropagationPolicy := metav1.DeletePropagationOrphan
podEventsQueue := cache.NewFIFO(func(obj interface{}) (string, error) {
e, ok := obj.(spec.PodEvent)
e, ok := obj.(PodEvent)
if !ok {
return "", fmt.Errorf("could not cast to PodEvent")
}
@@ -106,20 +110,20 @@ func New(cfg Config, kubeClient k8sutil.KubernetesClient, pgSpec spec.Postgresql
Postgresql: pgSpec,
pgUsers: make(map[string]spec.PgUser),
systemUsers: make(map[string]spec.PgUser),
podSubscribers: make(map[spec.NamespacedName]chan spec.PodEvent),
podSubscribers: make(map[spec.NamespacedName]chan PodEvent),
kubeResources: kubeResources{
Secrets: make(map[types.UID]*v1.Secret),
Services: make(map[PostgresRole]*v1.Service),
Endpoints: make(map[PostgresRole]*v1.Endpoints)},
userSyncStrategy: users.DefaultUserSyncStrategy{},
deleteOptions: &metav1.DeleteOptions{OrphanDependents: &orphanDependents},
deleteOptions: &metav1.DeleteOptions{PropagationPolicy: &deletePropagationPolicy},
podEventsQueue: podEventsQueue,
KubeClient: kubeClient,
dryRunMode: false,
}
cluster.logger = logger.WithField("pkg", "cluster").WithField("cluster-name", cluster.clusterName())
cluster.teamsAPIClient = teams.NewTeamsAPI(cfg.OpConfig.TeamsAPIUrl, logger)
cluster.oauthTokenGetter = NewSecretOauthTokenGetter(&kubeClient, cfg.OpConfig.OAuthTokenSecretName)
cluster.oauthTokenGetter = newSecretOauthTokenGetter(&kubeClient, cfg.OpConfig.OAuthTokenSecretName)
cluster.patroni = patroni.New(cluster.logger)
return cluster
@@ -141,39 +145,36 @@ func (c *Cluster) teamName() string {
func (c *Cluster) setProcessName(procName string, args ...interface{}) {
c.processMu.Lock()
defer c.processMu.Unlock()
c.currentProcess = spec.Process{
c.currentProcess = Process{
Name: fmt.Sprintf(procName, args...),
StartTime: time.Now(),
}
}
func (c *Cluster) setStatus(status spec.PostgresStatus) {
c.Status = status
b, err := json.Marshal(status)
func (c *Cluster) setStatus(status acidv1.PostgresStatus) {
// TODO: eventually switch to updateStatus() for kubernetes 1.11 and above
var (
err error
b []byte
)
if b, err = json.Marshal(status); err != nil {
c.logger.Errorf("could not marshal status: %v", err)
}
patch := []byte(fmt.Sprintf(`{"status": %s}`, string(b)))
// we cannot do a full scale update here without fetching the previous manifest (as the resourceVersion may differ),
// however, we could do patch without it. In the future, once /status subresource is there (starting Kubernets 1.11)
// we should take advantage of it.
newspec, err := c.KubeClient.AcidV1ClientSet.AcidV1().Postgresqls(c.clusterNamespace()).Patch(c.Name, types.MergePatchType, patch)
if err != nil {
c.logger.Fatalf("could not marshal status: %v", err)
}
request := []byte(fmt.Sprintf(`{"status": %s}`, string(b))) //TODO: Look into/wait for k8s go client methods
_, err = c.KubeClient.CRDREST.Patch(types.MergePatchType).
Namespace(c.Namespace).
Resource(constants.CRDResource).
Name(c.Name).
Body(request).
DoRaw()
if k8sutil.ResourceNotFound(err) {
c.logger.Warningf("could not set %q status for the non-existing cluster", status)
return
}
if err != nil {
c.logger.Warningf("could not set %q status for the cluster: %v", status, err)
c.logger.Errorf("could not update status: %v", err)
}
// update the spec, maintaining the new resourceVersion.
c.setSpec(newspec)
}
func (c *Cluster) isNewCluster() bool {
return c.Status == spec.ClusterStatusCreating
return c.Status == acidv1.ClusterStatusCreating
}
// initUsers populates c.systemUsers and c.pgUsers maps.
@@ -201,39 +202,6 @@ func (c *Cluster) initUsers() error {
return nil
}
/*
Ensures the service account required by StatefulSets to create pods exists in a namespace before a PG cluster is created there so that a user does not have to deploy the account manually.
The operator does not sync these accounts after creation.
*/
func (c *Cluster) createPodServiceAccounts() error {
podServiceAccountName := c.Config.OpConfig.PodServiceAccountName
_, err := c.KubeClient.ServiceAccounts(c.Namespace).Get(podServiceAccountName, metav1.GetOptions{})
if err != nil {
c.setProcessName(fmt.Sprintf("creating pod service account in the namespace %v", c.Namespace))
c.logger.Infof("the pod service account %q cannot be retrieved in the namespace %q. Trying to deploy the account.", podServiceAccountName, c.Namespace)
// get a separate copy of service account
// to prevent a race condition when setting a namespace for many clusters
sa := *c.PodServiceAccount
_, err = c.KubeClient.ServiceAccounts(c.Namespace).Create(&sa)
if err != nil {
return fmt.Errorf("cannot deploy the pod service account %q defined in the config map to the %q namespace: %v", podServiceAccountName, c.Namespace, err)
}
c.logger.Infof("successfully deployed the pod service account %q to the %q namespace", podServiceAccountName, c.Namespace)
} else {
c.logger.Infof("successfully found the service account %q used to create pods to the namespace %q", podServiceAccountName, c.Namespace)
}
return nil
}
// Create creates the new kubernetes objects associated with the cluster.
func (c *Cluster) Create() error {
c.mu.Lock()
@@ -248,13 +216,13 @@ func (c *Cluster) Create() error {
defer func() {
if err == nil {
c.setStatus(spec.ClusterStatusRunning) //TODO: are you sure it's running?
c.setStatus(acidv1.ClusterStatusRunning) //TODO: are you sure it's running?
} else {
c.setStatus(spec.ClusterStatusAddFailed)
c.setStatus(acidv1.ClusterStatusAddFailed)
}
}()
c.setStatus(spec.ClusterStatusCreating)
c.setStatus(acidv1.ClusterStatusCreating)
for _, role := range []PostgresRole{Master, Replica} {
@@ -302,11 +270,6 @@ func (c *Cluster) Create() error {
}
c.logger.Infof("pod disruption budget %q has been successfully created", util.NameFromMeta(pdb.ObjectMeta))
if err = c.createPodServiceAccounts(); err != nil {
return fmt.Errorf("could not create pod service account %v : %v", c.OpConfig.PodServiceAccountName, err)
}
c.logger.Infof("pod service accounts have been successfully synced")
if c.Statefulset != nil {
return fmt.Errorf("statefulset already exists in the cluster")
}
@@ -324,7 +287,7 @@ func (c *Cluster) Create() error {
}
c.logger.Infof("pods are ready")
// create database objects unless we are running without pods or disabled that feature explicitely
// create database objects unless we are running without pods or disabled that feature explicitly
if !(c.databaseAccessDisabled() || c.getNumberOfInstances(&c.Spec) <= 0) {
if err = c.createRoles(); err != nil {
return fmt.Errorf("could not create users: %v", err)
@@ -439,7 +402,6 @@ func (c *Cluster) compareVolumeClaimTemplates(setA, setB *v1beta1.StatefulSet) (
func (c *Cluster) compareContainers(setA, setB *v1beta1.StatefulSet) (bool, []string) {
reasons := make([]string, 0)
needsRollUpdate := false
for index, containerA := range setA.Spec.Template.Spec.Containers {
containerB := setB.Spec.Template.Spec.Containers[index]
for _, check := range c.getContainerChecks() {
@@ -461,8 +423,8 @@ func (c *Cluster) compareContainers(setA, setB *v1beta1.StatefulSet) (bool, []st
return needsRollUpdate, reasons
}
func compareResources(a *v1.ResourceRequirements, b *v1.ResourceRequirements) (equal bool) {
equal = true
func compareResources(a *v1.ResourceRequirements, b *v1.ResourceRequirements) bool {
equal := true
if a != nil {
equal = compareResoucesAssumeFirstNotNil(a, b)
}
@@ -470,7 +432,7 @@ func compareResources(a *v1.ResourceRequirements, b *v1.ResourceRequirements) (e
equal = compareResoucesAssumeFirstNotNil(b, a)
}
return
return equal
}
func compareResoucesAssumeFirstNotNil(a *v1.ResourceRequirements, b *v1.ResourceRequirements) bool {
@@ -493,20 +455,20 @@ func compareResoucesAssumeFirstNotNil(a *v1.ResourceRequirements, b *v1.Resource
// Update changes Kubernetes objects according to the new specification. Unlike the sync case, the missing object.
// (i.e. service) is treated as an error.
func (c *Cluster) Update(oldSpec, newSpec *spec.Postgresql) error {
func (c *Cluster) Update(oldSpec, newSpec *acidv1.Postgresql) error {
updateFailed := false
c.mu.Lock()
defer c.mu.Unlock()
c.setStatus(spec.ClusterStatusUpdating)
c.setStatus(acidv1.ClusterStatusUpdating)
c.setSpec(newSpec)
defer func() {
if updateFailed {
c.setStatus(spec.ClusterStatusUpdateFailed)
} else if c.Status != spec.ClusterStatusRunning {
c.setStatus(spec.ClusterStatusRunning)
c.setStatus(acidv1.ClusterStatusUpdateFailed)
} else if c.Status != acidv1.ClusterStatusRunning {
c.setStatus(acidv1.ClusterStatusRunning)
}
}()
@@ -616,7 +578,7 @@ func (c *Cluster) Delete() {
}
for _, obj := range c.Secrets {
if delete, user := c.shouldDeleteSecret(obj); !delete {
if doDelete, user := c.shouldDeleteSecret(obj); !doDelete {
c.logger.Warningf("not removing secret %q for the system user %q", obj.GetName(), user)
continue
}
@@ -647,21 +609,29 @@ func (c *Cluster) Delete() {
}
}
//NeedsRepair returns true if the cluster should be included in the repair scan (based on its in-memory status).
func (c *Cluster) NeedsRepair() (bool, acidv1.PostgresStatus) {
c.specMu.RLock()
defer c.specMu.RUnlock()
return !c.Status.Success(), c.Status
}
// ReceivePodEvent is called back by the controller in order to add the cluster's pod event to the queue.
func (c *Cluster) ReceivePodEvent(event spec.PodEvent) {
func (c *Cluster) ReceivePodEvent(event PodEvent) {
if err := c.podEventsQueue.Add(event); err != nil {
c.logger.Errorf("error when receiving pod events: %v", err)
}
}
func (c *Cluster) processPodEvent(obj interface{}) error {
event, ok := obj.(spec.PodEvent)
event, ok := obj.(PodEvent)
if !ok {
return fmt.Errorf("could not cast to PodEvent")
}
c.podSubscribersMu.RLock()
subscriber, ok := c.podSubscribers[event.PodName]
subscriber, ok := c.podSubscribers[spec.NamespacedName(event.PodName)]
c.podSubscribersMu.RUnlock()
if ok {
subscriber <- event
@@ -733,11 +703,13 @@ func (c *Cluster) initRobotUsers() error {
return nil
}
func (c *Cluster) initHumanUsers() error {
teamMembers, err := c.getTeamMembers()
func (c *Cluster) initTeamMembers(teamID string, isPostgresSuperuserTeam bool) error {
teamMembers, err := c.getTeamMembers(teamID)
if err != nil {
return fmt.Errorf("could not get list of team members: %v", err)
return fmt.Errorf("could not get list of team members for team %q: %v", teamID, err)
}
for _, username := range teamMembers {
flags := []string{constants.RoleFlagLogin}
memberOf := []string{c.OpConfig.PamRoleName}
@@ -745,7 +717,7 @@ func (c *Cluster) initHumanUsers() error {
if c.shouldAvoidProtectedOrSystemRole(username, "API role") {
continue
}
if c.OpConfig.EnableTeamSuperuser {
if c.OpConfig.EnableTeamSuperuser || isPostgresSuperuserTeam {
flags = append(flags, constants.RoleFlagSuperuser)
} else {
if c.OpConfig.TeamAdminRole != "" {
@@ -771,6 +743,33 @@ func (c *Cluster) initHumanUsers() error {
return nil
}
func (c *Cluster) initHumanUsers() error {
var clusterIsOwnedBySuperuserTeam bool
for _, postgresSuperuserTeam := range c.OpConfig.PostgresSuperuserTeams {
err := c.initTeamMembers(postgresSuperuserTeam, true)
if err != nil {
return fmt.Errorf("Cannot create a team %q of Postgres superusers: %v", postgresSuperuserTeam, err)
}
if postgresSuperuserTeam == c.Spec.TeamID {
clusterIsOwnedBySuperuserTeam = true
}
}
if clusterIsOwnedBySuperuserTeam {
c.logger.Infof("Team %q owning the cluster is also a team of superusers. Created superuser roles for its members instead of admin roles.", c.Spec.TeamID)
return nil
}
err := c.initTeamMembers(c.Spec.TeamID, false)
if err != nil {
return fmt.Errorf("Cannot create a team %q of admins owning the PG cluster: %v", c.Spec.TeamID, err)
}
return nil
}
func (c *Cluster) initInfrastructureRoles() error {
// add infrastructure roles from the operator's definition
for username, newRole := range c.InfrastructureRoles {
@@ -796,7 +795,8 @@ func (c *Cluster) initInfrastructureRoles() error {
}
// resolves naming conflicts between existing and new roles by chosing either of them.
func (c *Cluster) resolveNameConflict(currentRole, newRole *spec.PgUser) (result spec.PgUser) {
func (c *Cluster) resolveNameConflict(currentRole, newRole *spec.PgUser) spec.PgUser {
var result spec.PgUser
if newRole.Origin >= currentRole.Origin {
result = *newRole
} else {
@@ -804,7 +804,7 @@ func (c *Cluster) resolveNameConflict(currentRole, newRole *spec.PgUser) (result
}
c.logger.Debugf("resolved a conflict of role %q between %s and %s to %s",
newRole.Name, newRole.Origin, currentRole.Origin, result.Origin)
return
return result
}
func (c *Cluster) shouldAvoidProtectedOrSystemRole(username, purpose string) bool {
@@ -820,7 +820,7 @@ func (c *Cluster) shouldAvoidProtectedOrSystemRole(username, purpose string) boo
}
// GetCurrentProcess provides name of the last process of the cluster
func (c *Cluster) GetCurrentProcess() spec.Process {
func (c *Cluster) GetCurrentProcess() Process {
c.processMu.RLock()
defer c.processMu.RUnlock()
@@ -828,8 +828,8 @@ func (c *Cluster) GetCurrentProcess() spec.Process {
}
// GetStatus provides status of the cluster
func (c *Cluster) GetStatus() *spec.ClusterStatus {
return &spec.ClusterStatus{
func (c *Cluster) GetStatus() *ClusterStatus {
return &ClusterStatus{
Cluster: c.Spec.ClusterName,
Team: c.Spec.TeamID,
Status: c.Status,
@@ -843,19 +843,25 @@ func (c *Cluster) GetStatus() *spec.ClusterStatus {
PodDisruptionBudget: c.GetPodDisruptionBudget(),
CurrentProcess: c.GetCurrentProcess(),
Error: c.Error,
Error: fmt.Errorf("error: %s", c.Error),
}
}
// Switchover does a switchover (via Patroni) to a candidate pod
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)
var wg sync.WaitGroup
podLabelErr := make(chan error)
stopCh := make(chan struct{})
defer close(podLabelErr)
wg.Add(1)
go func() {
defer wg.Done()
ch := c.registerPodSubscriber(candidate)
defer c.unregisterPodSubscriber(candidate)
@@ -863,26 +869,32 @@ func (c *Cluster) Switchover(curMaster *v1.Pod, candidate spec.NamespacedName) e
select {
case <-stopCh:
case podLabelErr <- func() error {
_, err := c.waitForPodLabel(ch, stopCh, &role)
return err
case podLabelErr <- func() (err2 error) {
_, err2 = c.waitForPodLabel(ch, stopCh, &role)
return
}():
}
}()
if err := c.patroni.Switchover(curMaster, candidate.Name); err != nil {
close(stopCh)
return fmt.Errorf("could not failover: %v", err)
}
c.logger.Debugf("successfully failed over from %q to %q", curMaster.Name, candidate)
defer close(stopCh)
if err := <-podLabelErr; err != nil {
return fmt.Errorf("could not get master pod label: %v", err)
if err = c.patroni.Switchover(curMaster, candidate.Name); err == nil {
c.logger.Debugf("successfully failed 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)
}
return nil
// signal the role label waiting goroutine to close the shop and go home
close(stopCh)
// wait until the goroutine terminates, since unregisterPodSubscriber
// must be called before the outer return; otherwsise we risk subscribing to the same pod twice.
wg.Wait()
// close the label waiting channel no sooner than the waiting goroutine terminates.
close(podLabelErr)
return err
}
// Lock locks the cluster
@@ -902,9 +914,9 @@ func (c *Cluster) shouldDeleteSecret(secret *v1.Secret) (delete bool, userName s
type simpleActionWithResult func() error
type ClusterObjectGet func(name string) (spec.NamespacedName, error)
type clusterObjectGet func(name string) (spec.NamespacedName, error)
type ClusterObjectDelete func(name string) error
type clusterObjectDelete func(name string) error
func (c *Cluster) deletePatroniClusterObjects() error {
// TODO: figure out how to remove leftover patroni objects in other cases
@@ -921,8 +933,8 @@ func (c *Cluster) deletePatroniClusterObjects() error {
}
func (c *Cluster) deleteClusterObject(
get ClusterObjectGet,
del ClusterObjectDelete,
get clusterObjectGet,
del clusterObjectDelete,
objType string) error {
for _, suffix := range patroniObjectSuffixes {
name := fmt.Sprintf("%s-%s", c.Name, suffix)
@@ -950,11 +962,11 @@ func (c *Cluster) deletePatroniClusterEndpoints() error {
return util.NameFromMeta(ep.ObjectMeta), err
}
delete := func(name string) error {
deleteEndpointFn := func(name string) error {
return c.KubeClient.Endpoints(c.Namespace).Delete(name, c.deleteOptions)
}
return c.deleteClusterObject(get, delete, "endpoint")
return c.deleteClusterObject(get, deleteEndpointFn, "endpoint")
}
func (c *Cluster) deletePatroniClusterConfigMaps() error {
@@ -963,9 +975,9 @@ func (c *Cluster) deletePatroniClusterConfigMaps() error {
return util.NameFromMeta(cm.ObjectMeta), err
}
delete := func(name string) error {
deleteConfigMapFn := func(name string) error {
return c.KubeClient.ConfigMaps(c.Namespace).Delete(name, c.deleteOptions)
}
return c.deleteClusterObject(get, delete, "configmap")
return c.deleteClusterObject(get, deleteConfigMapFn, "configmap")
}
pkg/cluster/cluster_test.go
+155 -11
View File
@@ -2,14 +2,16 @@ package cluster
import (
"fmt"
"reflect"
"testing"
"github.com/Sirupsen/logrus"
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/config"
"github.com/zalando-incubator/postgres-operator/pkg/util/k8sutil"
"github.com/zalando-incubator/postgres-operator/pkg/util/teams"
"k8s.io/client-go/pkg/api/v1"
"reflect"
"testing"
"k8s.io/api/core/v1"
)
const (
@@ -21,43 +23,43 @@ var logger = logrus.New().WithField("test", "cluster")
var cl = New(Config{OpConfig: config.Config{ProtectedRoles: []string{"admin"},
Auth: config.Auth{SuperUsername: superUserName,
ReplicationUsername: replicationUserName}}},
k8sutil.KubernetesClient{}, spec.Postgresql{}, logger)
k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
func TestInitRobotUsers(t *testing.T) {
testName := "TestInitRobotUsers"
tests := []struct {
manifestUsers map[string]spec.UserFlags
manifestUsers map[string]acidv1.UserFlags
infraRoles map[string]spec.PgUser
result map[string]spec.PgUser
err error
}{
{
manifestUsers: map[string]spec.UserFlags{"foo": {"superuser", "createdb"}},
manifestUsers: map[string]acidv1.UserFlags{"foo": {"superuser", "createdb"}},
infraRoles: map[string]spec.PgUser{"foo": {Origin: spec.RoleOriginInfrastructure, Name: "foo", Password: "bar"}},
result: map[string]spec.PgUser{"foo": {Origin: spec.RoleOriginInfrastructure, Name: "foo", Password: "bar"}},
err: nil,
},
{
manifestUsers: map[string]spec.UserFlags{"!fooBar": {"superuser", "createdb"}},
manifestUsers: map[string]acidv1.UserFlags{"!fooBar": {"superuser", "createdb"}},
err: fmt.Errorf(`invalid username: "!fooBar"`),
},
{
manifestUsers: map[string]spec.UserFlags{"foobar": {"!superuser", "createdb"}},
manifestUsers: map[string]acidv1.UserFlags{"foobar": {"!superuser", "createdb"}},
err: fmt.Errorf(`invalid flags for user "foobar": ` +
`user flag "!superuser" is not alphanumeric`),
},
{
manifestUsers: map[string]spec.UserFlags{"foobar": {"superuser1", "createdb"}},
manifestUsers: map[string]acidv1.UserFlags{"foobar": {"superuser1", "createdb"}},
err: fmt.Errorf(`invalid flags for user "foobar": ` +
`user flag "SUPERUSER1" is not valid`),
},
{
manifestUsers: map[string]spec.UserFlags{"foobar": {"inherit", "noinherit"}},
manifestUsers: map[string]acidv1.UserFlags{"foobar": {"inherit", "noinherit"}},
err: fmt.Errorf(`invalid flags for user "foobar": ` +
`conflicting user flags: "NOINHERIT" and "INHERIT"`),
},
{
manifestUsers: map[string]spec.UserFlags{"admin": {"superuser"}, superUserName: {"createdb"}},
manifestUsers: map[string]acidv1.UserFlags{"admin": {"superuser"}, superUserName: {"createdb"}},
infraRoles: map[string]spec.PgUser{},
result: map[string]spec.PgUser{},
err: nil,
@@ -100,6 +102,7 @@ func (m *mockTeamsAPIClient) setMembers(members []string) {
m.members = members
}
// Test adding a member of a product team owning a particular DB cluster
func TestInitHumanUsers(t *testing.T) {
var mockTeamsAPI mockTeamsAPIClient
@@ -107,7 +110,9 @@ func TestInitHumanUsers(t *testing.T) {
cl.teamsAPIClient = &mockTeamsAPI
testName := "TestInitHumanUsers"
// members of a product team are granted superuser rights for DBs of their team
cl.OpConfig.EnableTeamSuperuser = true
cl.OpConfig.EnableTeamsAPI = true
cl.OpConfig.PamRoleName = "zalandos"
cl.Spec.TeamID = "test"
@@ -145,6 +150,145 @@ func TestInitHumanUsers(t *testing.T) {
}
}
type mockTeam struct {
teamID string
members []string
isPostgresSuperuserTeam bool
}
type mockTeamsAPIClientMultipleTeams struct {
teams []mockTeam
}
func (m *mockTeamsAPIClientMultipleTeams) TeamInfo(teamID, token string) (tm *teams.Team, err error) {
for _, team := range m.teams {
if team.teamID == teamID {
return &teams.Team{Members: team.members}, nil
}
}
// should not be reached if a slice with teams is populated correctly
return nil, nil
}
// Test adding members of maintenance teams that get superuser rights for all PG databases
func TestInitHumanUsersWithSuperuserTeams(t *testing.T) {
var mockTeamsAPI mockTeamsAPIClientMultipleTeams
cl.oauthTokenGetter = &mockOAuthTokenGetter{}
cl.teamsAPIClient = &mockTeamsAPI
cl.OpConfig.EnableTeamSuperuser = false
testName := "TestInitHumanUsersWithSuperuserTeams"
cl.OpConfig.EnableTeamsAPI = true
cl.OpConfig.PamRoleName = "zalandos"
teamA := mockTeam{
teamID: "postgres_superusers",
members: []string{"postgres_superuser"},
isPostgresSuperuserTeam: true,
}
userA := spec.PgUser{
Name: "postgres_superuser",
Origin: spec.RoleOriginTeamsAPI,
MemberOf: []string{cl.OpConfig.PamRoleName},
Flags: []string{"LOGIN", "SUPERUSER"},
}
teamB := mockTeam{
teamID: "postgres_admins",
members: []string{"postgres_admin"},
isPostgresSuperuserTeam: true,
}
userB := spec.PgUser{
Name: "postgres_admin",
Origin: spec.RoleOriginTeamsAPI,
MemberOf: []string{cl.OpConfig.PamRoleName},
Flags: []string{"LOGIN", "SUPERUSER"},
}
teamTest := mockTeam{
teamID: "test",
members: []string{"test_user"},
isPostgresSuperuserTeam: false,
}
userTest := spec.PgUser{
Name: "test_user",
Origin: spec.RoleOriginTeamsAPI,
MemberOf: []string{cl.OpConfig.PamRoleName},
Flags: []string{"LOGIN"},
}
tests := []struct {
ownerTeam string
existingRoles map[string]spec.PgUser
superuserTeams []string
teams []mockTeam
result map[string]spec.PgUser
}{
// case 1: there are two different teams of PG maintainers and one product team
{
ownerTeam: "test",
existingRoles: map[string]spec.PgUser{},
superuserTeams: []string{"postgres_superusers", "postgres_admins"},
teams: []mockTeam{teamA, teamB, teamTest},
result: map[string]spec.PgUser{
"postgres_superuser": userA,
"postgres_admin": userB,
"test_user": userTest,
},
},
// case 2: the team of superusers creates a new PG cluster
{
ownerTeam: "postgres_superusers",
existingRoles: map[string]spec.PgUser{},
superuserTeams: []string{"postgres_superusers"},
teams: []mockTeam{teamA},
result: map[string]spec.PgUser{
"postgres_superuser": userA,
},
},
// case 3: the team owning the cluster is promoted to the maintainers' status
{
ownerTeam: "postgres_superusers",
existingRoles: map[string]spec.PgUser{
// role with the name exists before w/o superuser privilege
"postgres_superuser": spec.PgUser{
Origin: spec.RoleOriginTeamsAPI,
Name: "postgres_superuser",
Password: "",
Flags: []string{"LOGIN"},
MemberOf: []string{cl.OpConfig.PamRoleName},
Parameters: map[string]string(nil)}},
superuserTeams: []string{"postgres_superusers"},
teams: []mockTeam{teamA},
result: map[string]spec.PgUser{
"postgres_superuser": userA,
},
},
}
for _, tt := range tests {
mockTeamsAPI.teams = tt.teams
cl.Spec.TeamID = tt.ownerTeam
cl.pgUsers = tt.existingRoles
cl.OpConfig.PostgresSuperuserTeams = tt.superuserTeams
if err := cl.initHumanUsers(); err != nil {
t.Errorf("%s got an unexpected error %v", testName, err)
}
if !reflect.DeepEqual(cl.pgUsers, tt.result) {
t.Errorf("%s expects %#v, got %#v", testName, tt.result, cl.pgUsers)
}
}
}
func TestShouldDeleteSecret(t *testing.T) {
testName := "TestShouldDeleteSecret"
pkg/cluster/pg.go → pkg/cluster/database.go
+19 -18
View File
@@ -153,12 +153,10 @@ func (c *Cluster) readPgUsersFromDatabase(userNames []string) (users spec.PgUser
// getDatabases returns the map of current databases with owners
// The caller is responsible for opening and closing the database connection
func (c *Cluster) getDatabases() (map[string]string, error) {
func (c *Cluster) getDatabases() (dbs map[string]string, err error) {
var (
rows *sql.Rows
err error
)
dbs := make(map[string]string)
if rows, err = c.pgDb.Query(getDatabasesSQL); err != nil {
return nil, fmt.Errorf("could not query database: %v", err)
@@ -166,15 +164,20 @@ func (c *Cluster) getDatabases() (map[string]string, error) {
defer func() {
if err2 := rows.Close(); err2 != nil {
err = fmt.Errorf("error when closing query cursor: %v", err2)
if err != nil {
err = fmt.Errorf("error when closing query cursor: %v, previous error: %v", err2, err)
} else {
err = fmt.Errorf("error when closing query cursor: %v", err2)
}
}
}()
dbs = make(map[string]string)
for rows.Next() {
var datname, owner string
err := rows.Scan(&datname, &owner)
if err != nil {
if err = rows.Scan(&datname, &owner); err != nil {
return nil, fmt.Errorf("error when processing row: %v", err)
}
dbs[datname] = owner
@@ -186,26 +189,24 @@ func (c *Cluster) getDatabases() (map[string]string, error) {
// executeCreateDatabase creates new database with the given owner.
// The caller is responsible for openinging and closing the database connection.
func (c *Cluster) executeCreateDatabase(datname, owner string) error {
if !c.databaseNameOwnerValid(datname, owner) {
return nil
}
c.logger.Infof("creating database %q with owner %q", datname, owner)
if _, err := c.pgDb.Exec(fmt.Sprintf(createDatabaseSQL, datname, owner)); err != nil {
return fmt.Errorf("could not execute create database: %v", err)
}
return nil
return c.execCreateOrAlterDatabase(datname, owner, createDatabaseSQL,
"creating database", "create database")
}
// executeCreateDatabase changes the owner of the given database.
// The caller is responsible for openinging and closing the database connection.
func (c *Cluster) executeAlterDatabaseOwner(datname string, owner string) error {
return c.execCreateOrAlterDatabase(datname, owner, alterDatabaseOwnerSQL,
"changing owner for database", "alter database owner")
}
func (c *Cluster) execCreateOrAlterDatabase(datname, owner, statement, doing, operation string) error {
if !c.databaseNameOwnerValid(datname, owner) {
return nil
}
c.logger.Infof("changing database %q owner to %q", datname, owner)
if _, err := c.pgDb.Exec(fmt.Sprintf(alterDatabaseOwnerSQL, datname, owner)); err != nil {
return fmt.Errorf("could not execute alter database owner: %v", err)
c.logger.Infof("%s %q owner %q", doing, datname, owner)
if _, err := c.pgDb.Exec(fmt.Sprintf(statement, datname, owner)); err != nil {
return fmt.Errorf("could not execute %s: %v", operation, err)
}
return nil
}
pkg/cluster/exec.go
+5 -6
View File
@@ -5,10 +5,9 @@ import (
"fmt"
"strings"
"k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
remotecommandconsts "k8s.io/apimachinery/pkg/util/remotecommand"
"k8s.io/client-go/kubernetes/scheme"
"k8s.io/client-go/pkg/api/v1"
"k8s.io/client-go/tools/remotecommand"
"github.com/zalando-incubator/postgres-operator/pkg/spec"
@@ -54,15 +53,15 @@ func (c *Cluster) ExecCommand(podName *spec.NamespacedName, command ...string) (
Stderr: true,
}, scheme.ParameterCodec)
exec, err := remotecommand.NewExecutor(c.RestConfig, "POST", req.URL())
exec, err := remotecommand.NewSPDYExecutor(c.RestConfig, "POST", req.URL())
if err != nil {
return "", fmt.Errorf("failed to init executor: %v", err)
}
err = exec.Stream(remotecommand.StreamOptions{
SupportedProtocols: remotecommandconsts.SupportedStreamingProtocols,
Stdout: &execOut,
Stderr: &execErr,
Stdout: &execOut,
Stderr: &execErr,
Tty: false,
})
if err != nil {
pkg/cluster/k8sres.go
+408 -207
View File
@@ -5,22 +5,28 @@ import (
"fmt"
"sort"
"github.com/Sirupsen/logrus"
"k8s.io/api/apps/v1beta1"
"k8s.io/api/core/v1"
policybeta1 "k8s.io/api/policy/v1beta1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/util/intstr"
"k8s.io/client-go/pkg/api/v1"
"k8s.io/client-go/pkg/apis/apps/v1beta1"
policybeta1 "k8s.io/client-go/pkg/apis/policy/v1beta1"
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/constants"
"k8s.io/apimachinery/pkg/labels"
)
const (
pgBinariesLocationTemplate = "/usr/lib/postgresql/%s/bin"
patroniPGBinariesParameterName = "bin_dir"
patroniPGParametersParameterName = "parameters"
patroniPGHBAConfParameterName = "pg_hba"
localHost = "127.0.0.1/32"
)
@@ -40,7 +46,6 @@ type patroniDCS struct {
type pgBootstrap struct {
Initdb []interface{} `json:"initdb"`
Users map[string]pgUser `json:"users"`
PgHBA []string `json:"pg_hba"`
DCS patroniDCS `json:"dcs,omitempty"`
}
@@ -79,25 +84,30 @@ func (c *Cluster) podDisruptionBudgetName() string {
return c.OpConfig.PDBNameFormat.Format("cluster", c.Name)
}
func (c *Cluster) resourceRequirements(resources spec.Resources) (*v1.ResourceRequirements, error) {
func (c *Cluster) makeDefaultResources() acidv1.Resources {
config := c.OpConfig
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}
}
func generateResourceRequirements(resources acidv1.Resources, defaultResources acidv1.Resources) (*v1.ResourceRequirements, error) {
var err error
specRequests := resources.ResourceRequest
specLimits := resources.ResourceLimits
config := c.OpConfig
defaultRequests := spec.ResourceDescription{CPU: config.DefaultCPURequest, Memory: config.DefaultMemoryRequest}
defaultLimits := spec.ResourceDescription{CPU: config.DefaultCPULimit, Memory: config.DefaultMemoryLimit}
result := v1.ResourceRequirements{}
result.Requests, err = fillResourceList(specRequests, defaultRequests)
result.Requests, err = fillResourceList(specRequests, defaultResources.ResourceRequest)
if err != nil {
return nil, fmt.Errorf("could not fill resource requests: %v", err)
}
result.Limits, err = fillResourceList(specLimits, defaultLimits)
result.Limits, err = fillResourceList(specLimits, defaultResources.ResourceLimits)
if err != nil {
return nil, fmt.Errorf("could not fill resource limits: %v", err)
}
@@ -105,7 +115,7 @@ func (c *Cluster) resourceRequirements(resources spec.Resources) (*v1.ResourceRe
return &result, nil
}
func fillResourceList(spec spec.ResourceDescription, defaults spec.ResourceDescription) (v1.ResourceList, error) {
func fillResourceList(spec acidv1.ResourceDescription, defaults acidv1.ResourceDescription) (v1.ResourceList, error) {
var err error
requests := v1.ResourceList{}
@@ -135,7 +145,7 @@ func fillResourceList(spec spec.ResourceDescription, defaults spec.ResourceDescr
return requests, nil
}
func (c *Cluster) generateSpiloJSONConfiguration(pg *spec.PostgresqlParam, patroni *spec.Patroni) string {
func generateSpiloJSONConfiguration(pg *acidv1.PostgresqlParam, patroni *acidv1.Patroni, pamRoleName string, logger *logrus.Entry) string {
config := spiloConfiguration{}
config.Bootstrap = pgBootstrap{}
@@ -178,7 +188,7 @@ PatroniInitDBParams:
}
}
default:
c.logger.Warningf("unsupported type for initdb configuration item %s: %T", defaultParam, defaultParam)
logger.Warningf("unsupported type for initdb configuration item %s: %T", defaultParam, defaultParam)
continue PatroniInitDBParams
}
}
@@ -193,19 +203,6 @@ PatroniInitDBParams:
config.Bootstrap.Initdb = append(config.Bootstrap.Initdb, map[string]string{k: v})
}
// pg_hba parameters in the manifest replace the default ones. We cannot
// reasonably merge them automatically, because pg_hba parsing stops on
// a first successfully matched rule.
if len(patroni.PgHba) > 0 {
config.Bootstrap.PgHBA = patroni.PgHba
} else {
config.Bootstrap.PgHBA = []string{
"hostnossl all all all reject",
fmt.Sprintf("hostssl all +%s all pam", c.OpConfig.PamRoleName),
"hostssl all all all md5",
}
}
if patroni.MaximumLagOnFailover >= 0 {
config.Bootstrap.DCS.MaximumLagOnFailover = patroni.MaximumLagOnFailover
}
@@ -222,43 +219,56 @@ PatroniInitDBParams:
config.PgLocalConfiguration = make(map[string]interface{})
config.PgLocalConfiguration[patroniPGBinariesParameterName] = fmt.Sprintf(pgBinariesLocationTemplate, pg.PgVersion)
if len(pg.Parameters) > 0 {
localParameters := make(map[string]string)
bootstrapParameters := make(map[string]string)
for param, val := range pg.Parameters {
if isBootstrapOnlyParameter(param) {
bootstrapParameters[param] = val
} else {
localParameters[param] = val
}
local, bootstrap := getLocalAndBoostrapPostgreSQLParameters(pg.Parameters)
if len(local) > 0 {
config.PgLocalConfiguration[patroniPGParametersParameterName] = local
}
if len(localParameters) > 0 {
config.PgLocalConfiguration[patroniPGParametersParameterName] = localParameters
}
if len(bootstrapParameters) > 0 {
if len(bootstrap) > 0 {
config.Bootstrap.DCS.PGBootstrapConfiguration = make(map[string]interface{})
config.Bootstrap.DCS.PGBootstrapConfiguration[patroniPGParametersParameterName] = bootstrapParameters
config.Bootstrap.DCS.PGBootstrapConfiguration[patroniPGParametersParameterName] = bootstrap
}
}
// Patroni gives us a choice of writing pg_hba.conf to either the bootstrap section or to the local postgresql one.
// We choose the local one, because we need Patroni to change pg_hba.conf in PostgreSQL after the user changes the
// relevant section in the manifest.
if len(patroni.PgHba) > 0 {
config.PgLocalConfiguration[patroniPGHBAConfParameterName] = patroni.PgHba
}
config.Bootstrap.Users = map[string]pgUser{
c.OpConfig.PamRoleName: {
pamRoleName: {
Password: "",
Options: []string{constants.RoleFlagCreateDB, constants.RoleFlagNoLogin},
},
}
result, err := json.Marshal(config)
if err != nil {
c.logger.Errorf("cannot convert spilo configuration into JSON: %v", err)
logger.Errorf("cannot convert spilo configuration into JSON: %v", err)
return ""
}
return string(result)
}
func (c *Cluster) nodeAffinity() *v1.Affinity {
func getLocalAndBoostrapPostgreSQLParameters(parameters map[string]string) (local, bootstrap map[string]string) {
local = make(map[string]string)
bootstrap = make(map[string]string)
for param, val := range parameters {
if isBootstrapOnlyParameter(param) {
bootstrap[param] = val
} else {
local[param] = val
}
}
return
}
func nodeAffinity(nodeReadinessLabel map[string]string) *v1.Affinity {
matchExpressions := make([]v1.NodeSelectorRequirement, 0)
if len(c.OpConfig.NodeReadinessLabel) == 0 {
if len(nodeReadinessLabel) == 0 {
return nil
}
for k, v := range c.OpConfig.NodeReadinessLabel {
for k, v := range nodeReadinessLabel {
matchExpressions = append(matchExpressions, v1.NodeSelectorRequirement{
Key: k,
Operator: v1.NodeSelectorOpIn,
@@ -275,13 +285,12 @@ func (c *Cluster) nodeAffinity() *v1.Affinity {
}
}
func (c *Cluster) tolerations(tolerationsSpec *[]v1.Toleration) []v1.Toleration {
func tolerations(tolerationsSpec *[]v1.Toleration, podToleration map[string]string) []v1.Toleration {
// allow to override tolerations by postgresql manifest
if len(*tolerationsSpec) > 0 {
return *tolerationsSpec
}
podToleration := c.Config.OpConfig.PodToleration
if len(podToleration["key"]) > 0 || len(podToleration["operator"]) > 0 || len(podToleration["value"]) > 0 || len(podToleration["effect"]) > 0 {
return []v1.Toleration{
{
@@ -309,19 +318,128 @@ func isBootstrapOnlyParameter(param string) bool {
param == "track_commit_timestamp"
}
func (c *Cluster) generatePodTemplate(
uid types.UID,
resourceRequirements *v1.ResourceRequirements,
resourceRequirementsScalyrSidecar *v1.ResourceRequirements,
tolerationsSpec *[]v1.Toleration,
pgParameters *spec.PostgresqlParam,
patroniParameters *spec.Patroni,
cloneDescription *spec.CloneDescription,
dockerImage *string,
customPodEnvVars map[string]string,
) *v1.PodTemplateSpec {
spiloConfiguration := c.generateSpiloJSONConfiguration(pgParameters, patroniParameters)
func generateVolumeMounts() []v1.VolumeMount {
return []v1.VolumeMount{
{
Name: constants.DataVolumeName,
MountPath: constants.PostgresDataMount, //TODO: fetch from manifest
},
}
}
func generateSpiloContainer(
name string,
dockerImage *string,
resourceRequirements *v1.ResourceRequirements,
envVars []v1.EnvVar,
volumeMounts []v1.VolumeMount,
) *v1.Container {
privilegedMode := true
return &v1.Container{
Name: name,
Image: *dockerImage,
ImagePullPolicy: v1.PullIfNotPresent,
Resources: *resourceRequirements,
Ports: []v1.ContainerPort{
{
ContainerPort: 8008,
Protocol: v1.ProtocolTCP,
},
{
ContainerPort: 5432,
Protocol: v1.ProtocolTCP,
},
{
ContainerPort: 8080,
Protocol: v1.ProtocolTCP,
},
},
VolumeMounts: volumeMounts,
Env: envVars,
SecurityContext: &v1.SecurityContext{
Privileged: &privilegedMode,
},
}
}
func generateSidecarContainers(sidecars []acidv1.Sidecar,
volumeMounts []v1.VolumeMount, defaultResources acidv1.Resources,
superUserName string, credentialsSecretName string, logger *logrus.Entry) ([]v1.Container, error) {
if len(sidecars) > 0 {
result := make([]v1.Container, 0)
for index, sidecar := range sidecars {
resources, err := generateResourceRequirements(
makeResources(
sidecar.Resources.ResourceRequest.CPU,
sidecar.Resources.ResourceRequest.Memory,
sidecar.Resources.ResourceLimits.CPU,
sidecar.Resources.ResourceLimits.Memory,
),
defaultResources,
)
if err != nil {
return nil, err
}
sc := getSidecarContainer(sidecar, index, volumeMounts, resources, superUserName, credentialsSecretName, logger)
result = append(result, *sc)
}
return result, nil
}
return nil, nil
}
func generatePodTemplate(
namespace string,
labels labels.Set,
spiloContainer *v1.Container,
sidecarContainers []v1.Container,
tolerationsSpec *[]v1.Toleration,
nodeAffinity *v1.Affinity,
terminateGracePeriod int64,
podServiceAccountName string,
kubeIAMRole string,
priorityClassName string,
) (*v1.PodTemplateSpec, error) {
terminateGracePeriodSeconds := terminateGracePeriod
containers := []v1.Container{*spiloContainer}
containers = append(containers, sidecarContainers...)
podSpec := v1.PodSpec{
ServiceAccountName: podServiceAccountName,
TerminationGracePeriodSeconds: &terminateGracePeriodSeconds,
Containers: containers,
Tolerations: *tolerationsSpec,
}
if nodeAffinity != nil {
podSpec.Affinity = nodeAffinity
}
if priorityClassName != "" {
podSpec.PriorityClassName = priorityClassName
}
template := v1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: labels,
Namespace: namespace,
},
Spec: podSpec,
}
if kubeIAMRole != "" {
template.Annotations = map[string]string{constants.KubeIAmAnnotation: kubeIAMRole}
}
return &template, nil
}
// generatePodEnvVars generates environment variables for the Spilo Pod
func (c *Cluster) generateSpiloPodEnvVars(uid types.UID, spiloConfiguration string, cloneDescription *acidv1.CloneDescription, customPodEnvVarsList []v1.EnvVar) []v1.EnvVar {
envVars := []v1.EnvVar{
{
Name: "SCOPE",
@@ -409,134 +527,89 @@ func (c *Cluster) generatePodTemplate(
envVars = append(envVars, c.generateCloneEnvironment(cloneDescription)...)
}
var names []string
// handle environment variables from the PodEnvironmentConfigMap. We don't use envSource here as it is impossible
// to track any changes to the object envSource points to. In order to emulate the envSource behavior, however, we
// need to make sure that PodConfigMap variables doesn't override those we set explicitly from the configuration
// parameters
envVarsMap := make(map[string]string)
for _, envVar := range envVars {
envVarsMap[envVar.Name] = envVar.Value
if len(customPodEnvVarsList) > 0 {
envVars = append(envVars, customPodEnvVarsList...)
}
for name := range customPodEnvVars {
if _, ok := envVarsMap[name]; !ok {
names = append(names, name)
} else {
c.logger.Warningf("variable %q value from %q is ignored: conflict with the definition from the operator",
name, c.OpConfig.PodEnvironmentConfigMap)
return envVars
}
// deduplicateEnvVars makes sure there are no duplicate in the target envVar array. While Kubernetes already
// deduplicates variables defined in a container, it leaves the last definition in the list and this behavior is not
// well-documented, which means that the behavior can be reversed at some point (it may also start producing an error).
// Therefore, the merge is done by the operator, the entries that are ahead in the passed list take priority over those
// that are behind, and only the name is considered in order to eliminate duplicates.
func deduplicateEnvVars(input []v1.EnvVar, containerName string, logger *logrus.Entry) []v1.EnvVar {
result := make([]v1.EnvVar, 0)
names := make(map[string]int)
for i, va := range input {
if names[va.Name] == 0 {
names[va.Name]++
result = append(result, input[i])
} else if names[va.Name] == 1 {
names[va.Name]++
logger.Warningf("variable %q is defined in %q more than once, the subsequent definitions are ignored",
va.Name, containerName)
}
}
sort.Strings(names)
for _, name := range names {
envVars = append(envVars, v1.EnvVar{Name: name, Value: customPodEnvVars[name]})
return result
}
func getSidecarContainer(sidecar acidv1.Sidecar, index int, volumeMounts []v1.VolumeMount,
resources *v1.ResourceRequirements, superUserName string, credentialsSecretName string, logger *logrus.Entry) *v1.Container {
name := sidecar.Name
if name == "" {
name = fmt.Sprintf("sidecar-%d", index)
}
privilegedMode := true
containerImage := c.OpConfig.DockerImage
if dockerImage != nil && *dockerImage != "" {
containerImage = *dockerImage
}
volumeMounts := []v1.VolumeMount{
env := []v1.EnvVar{
{
Name: constants.DataVolumeName,
MountPath: constants.PostgresDataMount, //TODO: fetch from manifest
Name: "POD_NAME",
ValueFrom: &v1.EnvVarSource{
FieldRef: &v1.ObjectFieldSelector{
APIVersion: "v1",
FieldPath: "metadata.name",
},
},
},
}
container := v1.Container{
Name: c.containerName(),
Image: containerImage,
ImagePullPolicy: v1.PullIfNotPresent,
Resources: *resourceRequirements,
Ports: []v1.ContainerPort{
{
ContainerPort: 8008,
Protocol: v1.ProtocolTCP,
},
{
ContainerPort: 5432,
Protocol: v1.ProtocolTCP,
{
Name: "POD_NAMESPACE",
ValueFrom: &v1.EnvVarSource{
FieldRef: &v1.ObjectFieldSelector{
APIVersion: "v1",
FieldPath: "metadata.namespace",
},
},
{
ContainerPort: 8080,
Protocol: v1.ProtocolTCP,
},
},
VolumeMounts: volumeMounts,
Env: envVars,
SecurityContext: &v1.SecurityContext{
Privileged: &privilegedMode,
{
Name: "POSTGRES_USER",
Value: superUserName,
},
}
terminateGracePeriodSeconds := int64(c.OpConfig.PodTerminateGracePeriod.Seconds())
podSpec := v1.PodSpec{
ServiceAccountName: c.OpConfig.PodServiceAccountName,
TerminationGracePeriodSeconds: &terminateGracePeriodSeconds,
Containers: []v1.Container{container},
Tolerations: c.tolerations(tolerationsSpec),
}
if affinity := c.nodeAffinity(); affinity != nil {
podSpec.Affinity = affinity
}
if c.OpConfig.ScalyrAPIKey != "" && c.OpConfig.ScalyrImage != "" {
podSpec.Containers = append(
podSpec.Containers,
v1.Container{
Name: "scalyr-sidecar",
Image: c.OpConfig.ScalyrImage,
ImagePullPolicy: v1.PullIfNotPresent,
Resources: *resourceRequirementsScalyrSidecar,
VolumeMounts: volumeMounts,
Env: []v1.EnvVar{
{
Name: "POD_NAME",
ValueFrom: &v1.EnvVarSource{
FieldRef: &v1.ObjectFieldSelector{
APIVersion: "v1",
FieldPath: "metadata.name",
},
},
},
{
Name: "POD_NAMESPACE",
ValueFrom: &v1.EnvVarSource{
FieldRef: &v1.ObjectFieldSelector{
APIVersion: "v1",
FieldPath: "metadata.namespace",
},
},
},
{
Name: "SCALYR_API_KEY",
Value: c.OpConfig.ScalyrAPIKey,
},
{
Name: "SCALYR_SERVER_HOST",
Value: c.Name,
},
{
Name: "SCALYR_SERVER_URL",
Value: c.OpConfig.ScalyrServerURL,
{
Name: "POSTGRES_PASSWORD",
ValueFrom: &v1.EnvVarSource{
SecretKeyRef: &v1.SecretKeySelector{
LocalObjectReference: v1.LocalObjectReference{
Name: credentialsSecretName,
},
Key: "password",
},
},
)
}
template := v1.PodTemplateSpec{
ObjectMeta: metav1.ObjectMeta{
Labels: c.labelsSet(true),
Namespace: c.Namespace,
},
Spec: podSpec,
}
if c.OpConfig.KubeIAMRole != "" {
template.Annotations = map[string]string{constants.KubeIAmAnnotation: c.OpConfig.KubeIAMRole}
if len(sidecar.Env) > 0 {
env = append(env, sidecar.Env...)
}
return &v1.Container{
Name: name,
Image: sidecar.DockerImage,
ImagePullPolicy: v1.PullIfNotPresent,
Resources: *resources,
VolumeMounts: volumeMounts,
Env: deduplicateEnvVars(env, name, logger),
Ports: sidecar.Ports,
}
return &template
}
func getBucketScopeSuffix(uid string) string {
@@ -546,46 +619,114 @@ func getBucketScopeSuffix(uid string) string {
return ""
}
func makeResources(cpuRequest, memoryRequest, cpuLimit, memoryLimit string) spec.Resources {
return spec.Resources{
ResourceRequest: spec.ResourceDescription{
func makeResources(cpuRequest, memoryRequest, cpuLimit, memoryLimit string) acidv1.Resources {
return acidv1.Resources{
ResourceRequest: acidv1.ResourceDescription{
CPU: cpuRequest,
Memory: memoryRequest,
},
ResourceLimits: spec.ResourceDescription{
ResourceLimits: acidv1.ResourceDescription{
CPU: cpuLimit,
Memory: memoryLimit,
},
}
}
func (c *Cluster) generateStatefulSet(spec *spec.PostgresSpec) (*v1beta1.StatefulSet, error) {
resourceRequirements, err := c.resourceRequirements(spec.Resources)
func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*v1beta1.StatefulSet, error) {
var (
err error
sidecarContainers []v1.Container
podTemplate *v1.PodTemplateSpec
volumeClaimTemplate *v1.PersistentVolumeClaim
)
defaultResources := c.makeDefaultResources()
resourceRequirements, err := generateResourceRequirements(spec.Resources, defaultResources)
if err != nil {
return nil, fmt.Errorf("could not generate resource requirements: %v", err)
}
resourceRequirementsScalyrSidecar, err := c.resourceRequirements(
makeResources(
c.OpConfig.ScalyrCPURequest,
c.OpConfig.ScalyrMemoryRequest,
c.OpConfig.ScalyrCPULimit,
c.OpConfig.ScalyrMemoryLimit,
),
)
if err != nil {
return nil, fmt.Errorf("could not generate Scalyr sidecar resource requirements: %v", err)
}
var customPodEnvVars map[string]string
customPodEnvVarsList := make([]v1.EnvVar, 0)
if c.OpConfig.PodEnvironmentConfigMap != "" {
if cm, err := c.KubeClient.ConfigMaps(c.Namespace).Get(c.OpConfig.PodEnvironmentConfigMap, metav1.GetOptions{}); err != nil {
var cm *v1.ConfigMap
cm, err = c.KubeClient.ConfigMaps(c.Namespace).Get(c.OpConfig.PodEnvironmentConfigMap, metav1.GetOptions{})
if err != nil {
return nil, fmt.Errorf("could not read PodEnvironmentConfigMap: %v", err)
} else {
customPodEnvVars = cm.Data
}
for k, v := range cm.Data {
customPodEnvVarsList = append(customPodEnvVarsList, v1.EnvVar{Name: k, Value: v})
}
sort.Slice(customPodEnvVarsList,
func(i, j int) bool { return customPodEnvVarsList[i].Name < customPodEnvVarsList[j].Name })
}
podTemplate := c.generatePodTemplate(c.Postgresql.GetUID(), resourceRequirements, resourceRequirementsScalyrSidecar, &spec.Tolerations, &spec.PostgresqlParam, &spec.Patroni, &spec.Clone, &spec.DockerImage, customPodEnvVars)
volumeClaimTemplate, err := generatePersistentVolumeClaimTemplate(spec.Volume.Size, spec.Volume.StorageClass)
spiloConfiguration := generateSpiloJSONConfiguration(&spec.PostgresqlParam, &spec.Patroni, c.OpConfig.PamRoleName, c.logger)
// generate environment variables for the spilo container
spiloEnvVars := deduplicateEnvVars(
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)
volumeMounts := generateVolumeMounts()
// generate the spilo container
spiloContainer := generateSpiloContainer(c.containerName(), &effectiveDockerImage, resourceRequirements, spiloEnvVars, volumeMounts)
// resolve conflicts between operator-global and per-cluster sidecards
sideCars := c.mergeSidecars(spec.Sidecars)
resourceRequirementsScalyrSidecar := makeResources(
c.OpConfig.ScalyrCPURequest,
c.OpConfig.ScalyrMemoryRequest,
c.OpConfig.ScalyrCPULimit,
c.OpConfig.ScalyrMemoryLimit,
)
// generate scalyr sidecar container
if scalyrSidecar :=
generateScalyrSidecarSpec(c.Name,
c.OpConfig.ScalyrAPIKey,
c.OpConfig.ScalyrServerURL,
c.OpConfig.ScalyrImage,
&resourceRequirementsScalyrSidecar, c.logger); scalyrSidecar != nil {
sideCars = append(sideCars, *scalyrSidecar)
}
// generate sidecar containers
if sidecarContainers, err = generateSidecarContainers(sideCars, volumeMounts, defaultResources,
c.OpConfig.SuperUsername, c.credentialSecretName(c.OpConfig.SuperUsername), c.logger); err != nil {
return nil, fmt.Errorf("could not generate sidecar containers: %v", err)
}
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
if podTemplate, err = generatePodTemplate(
c.Namespace,
c.labelsSet(true),
spiloContainer,
sidecarContainers,
&tolerationSpec,
nodeAffinity(c.OpConfig.NodeReadinessLabel),
int64(c.OpConfig.PodTerminateGracePeriod.Seconds()),
c.OpConfig.PodServiceAccountName,
c.OpConfig.KubeIAMRole,
effectivePodPriorityClassName); err != nil {
return nil, fmt.Errorf("could not generate pod template: %v", err)
}
if err != nil {
return nil, fmt.Errorf("could not generate pod template: %v", err)
}
if volumeClaimTemplate, err = generatePersistentVolumeClaimTemplate(spec.Volume.Size,
spec.Volume.StorageClass); err != nil {
return nil, fmt.Errorf("could not generate volume claim template: %v", err)
}
@@ -596,7 +737,7 @@ func (c *Cluster) generateStatefulSet(spec *spec.PostgresSpec) (*v1beta1.Statefu
Name: c.statefulSetName(),
Namespace: c.Namespace,
Labels: c.labelsSet(true),
Annotations: map[string]string{RollingUpdateStatefulsetAnnotationKey: "false"},
Annotations: map[string]string{rollingUpdateStatefulsetAnnotationKey: "false"},
},
Spec: v1beta1.StatefulSetSpec{
Replicas: &numberOfInstances,
@@ -610,11 +751,65 @@ func (c *Cluster) generateStatefulSet(spec *spec.PostgresSpec) (*v1beta1.Statefu
return statefulSet, nil
}
func (c *Cluster) getNumberOfInstances(spec *spec.PostgresSpec) (newcur int32) {
func generateScalyrSidecarSpec(clusterName, APIKey, serverURL, dockerImage string,
containerResources *acidv1.Resources, logger *logrus.Entry) *acidv1.Sidecar {
if APIKey == "" || dockerImage == "" {
if APIKey == "" && dockerImage != "" {
logger.Warning("Not running Scalyr sidecar: SCALYR_API_KEY must be defined")
}
return nil
}
scalarSpec := &acidv1.Sidecar{
Name: "scalyr-sidecar",
DockerImage: dockerImage,
Env: []v1.EnvVar{
{
Name: "SCALYR_API_KEY",
Value: APIKey,
},
{
Name: "SCALYR_SERVER_HOST",
Value: clusterName,
},
},
Resources: *containerResources,
}
if serverURL != "" {
scalarSpec.Env = append(scalarSpec.Env, v1.EnvVar{Name: "SCALYR_SERVER_URL", Value: serverURL})
}
return scalarSpec
}
// mergeSidecar merges globally-defined sidecars with those defined in the cluster manifest
func (c *Cluster) mergeSidecars(sidecars []acidv1.Sidecar) []acidv1.Sidecar {
globalSidecarsToSkip := map[string]bool{}
result := make([]acidv1.Sidecar, 0)
for i, sidecar := range sidecars {
dockerImage, ok := c.OpConfig.Sidecars[sidecar.Name]
if ok {
if dockerImage != sidecar.DockerImage {
c.logger.Warningf("merging definitions for sidecar %q: "+
"ignoring %q in the global scope in favor of %q defined in the cluster",
sidecar.Name, dockerImage, sidecar.DockerImage)
}
globalSidecarsToSkip[sidecar.Name] = true
}
result = append(result, sidecars[i])
}
for name, dockerImage := range c.OpConfig.Sidecars {
if !globalSidecarsToSkip[name] {
result = append(result, acidv1.Sidecar{Name: name, DockerImage: dockerImage})
}
}
return result
}
func (c *Cluster) getNumberOfInstances(spec *acidv1.PostgresSpec) int32 {
min := c.OpConfig.MinInstances
max := c.OpConfig.MaxInstances
cur := spec.NumberOfInstances
newcur = cur
newcur := cur
if max >= 0 && newcur > max {
newcur = max
@@ -626,18 +821,23 @@ func (c *Cluster) getNumberOfInstances(spec *spec.PostgresSpec) (newcur int32) {
c.logger.Infof("adjusted number of instances from %d to %d (min: %d, max: %d)", cur, newcur, min, max)
}
return
return newcur
}
func generatePersistentVolumeClaimTemplate(volumeSize, volumeStorageClass string) (*v1.PersistentVolumeClaim, error) {
var storageClassName *string
metadata := metav1.ObjectMeta{
Name: constants.DataVolumeName,
}
if volumeStorageClass != "" {
// TODO: check if storage class exists
// TODO: remove the old annotation, switching completely to the StorageClassName field.
metadata.Annotations = map[string]string{"volume.beta.kubernetes.io/storage-class": volumeStorageClass}
storageClassName = &volumeStorageClass
} else {
metadata.Annotations = map[string]string{"volume.alpha.kubernetes.io/storage-class": "default"}
storageClassName = nil
}
quantity, err := resource.ParseQuantity(volumeSize)
@@ -654,14 +854,15 @@ func generatePersistentVolumeClaimTemplate(volumeSize, volumeStorageClass string
v1.ResourceStorage: quantity,
},
},
StorageClassName: storageClassName,
},
}
return volumeClaim, nil
}
func (c *Cluster) generateUserSecrets() (secrets map[string]*v1.Secret) {
secrets = make(map[string]*v1.Secret, len(c.pgUsers))
func (c *Cluster) generateUserSecrets() map[string]*v1.Secret {
secrets := make(map[string]*v1.Secret, len(c.pgUsers))
namespace := c.Namespace
for username, pgUser := range c.pgUsers {
//Skip users with no password i.e. human users (they'll be authenticated using pam)
@@ -678,7 +879,7 @@ func (c *Cluster) generateUserSecrets() (secrets map[string]*v1.Secret) {
}
}
return
return secrets
}
func (c *Cluster) generateSingleUserSecret(namespace string, pgUser spec.PgUser) *v1.Secret {
@@ -707,7 +908,7 @@ func (c *Cluster) generateSingleUserSecret(namespace string, pgUser spec.PgUser)
return &secret
}
func (c *Cluster) shouldCreateLoadBalancerForService(role PostgresRole, spec *spec.PostgresSpec) bool {
func (c *Cluster) shouldCreateLoadBalancerForService(role PostgresRole, spec *acidv1.PostgresSpec) bool {
switch role {
@@ -735,7 +936,7 @@ func (c *Cluster) shouldCreateLoadBalancerForService(role PostgresRole, spec *sp
}
func (c *Cluster) generateService(role PostgresRole, spec *spec.PostgresSpec) *v1.Service {
func (c *Cluster) generateService(role PostgresRole, spec *acidv1.PostgresSpec) *v1.Service {
var dnsName string
if role == Master {
@@ -806,7 +1007,7 @@ func (c *Cluster) generateEndpoint(role PostgresRole, subsets []v1.EndpointSubse
return endpoints
}
func (c *Cluster) generateCloneEnvironment(description *spec.CloneDescription) []v1.EnvVar {
func (c *Cluster) generateCloneEnvironment(description *acidv1.CloneDescription) []v1.EnvVar {
result := make([]v1.EnvVar, 0)
if description.ClusterName == "" {
@@ -841,7 +1042,7 @@ func (c *Cluster) generateCloneEnvironment(description *spec.CloneDescription) [
result = append(result, v1.EnvVar{Name: "CLONE_METHOD", Value: "CLONE_WITH_WALE"})
result = append(result, v1.EnvVar{Name: "CLONE_WAL_S3_BUCKET", Value: c.OpConfig.WALES3Bucket})
result = append(result, v1.EnvVar{Name: "CLONE_TARGET_TIME", Value: description.EndTimestamp})
result = append(result, v1.EnvVar{Name: "CLONE_WAL_BUCKET_SCOPE_SUFFIX", Value: getBucketScopeSuffix(description.Uid)})
result = append(result, v1.EnvVar{Name: "CLONE_WAL_BUCKET_SCOPE_SUFFIX", Value: getBucketScopeSuffix(description.UID)})
result = append(result, v1.EnvVar{Name: "CLONE_WAL_BUCKET_SCOPE_PREFIX", Value: ""})
}
pkg/cluster/k8sres_test.go
+7 -7
View File
@@ -1,8 +1,8 @@
package cluster
import (
"github.com/zalando-incubator/postgres-operator/pkg/spec"
u "github.com/zalando-incubator/postgres-operator/pkg/util"
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/k8sutil"
"testing"
@@ -18,41 +18,41 @@ func TestCreateLoadBalancerLogic(t *testing.T) {
ReplicationUsername: replicationUserName,
},
},
}, k8sutil.KubernetesClient{}, spec.Postgresql{}, logger)
}, k8sutil.KubernetesClient{}, acidv1.Postgresql{}, logger)
testName := "TestCreateLoadBalancerLogic"
tests := []struct {
subtest string
role PostgresRole
spec *spec.PostgresSpec
spec *acidv1.PostgresSpec
opConfig config.Config
result bool
}{
{
subtest: "new format, load balancer is enabled for replica",
role: Replica,
spec: &spec.PostgresSpec{EnableReplicaLoadBalancer: u.True()},
spec: &acidv1.PostgresSpec{EnableReplicaLoadBalancer: u.True()},
opConfig: config.Config{},
result: true,
},
{
subtest: "new format, load balancer is disabled for replica",
role: Replica,
spec: &spec.PostgresSpec{EnableReplicaLoadBalancer: u.False()},
spec: &acidv1.PostgresSpec{EnableReplicaLoadBalancer: u.False()},
opConfig: config.Config{},
result: false,
},
{
subtest: "new format, load balancer isn't specified for replica",
role: Replica,
spec: &spec.PostgresSpec{EnableReplicaLoadBalancer: nil},
spec: &acidv1.PostgresSpec{EnableReplicaLoadBalancer: nil},
opConfig: config.Config{EnableReplicaLoadBalancer: true},
result: true,
},
{
subtest: "new format, load balancer isn't specified for replica",
role: Replica,
spec: &spec.PostgresSpec{EnableReplicaLoadBalancer: nil},
spec: &acidv1.PostgresSpec{EnableReplicaLoadBalancer: nil},
opConfig: config.Config{EnableReplicaLoadBalancer: false},
result: false,
},
pkg/cluster/pod.go
+32 -26
View File
@@ -4,11 +4,12 @@ import (
"fmt"
"math/rand"
"k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/pkg/api/v1"
"github.com/zalando-incubator/postgres-operator/pkg/spec"
"github.com/zalando-incubator/postgres-operator/pkg/util"
"k8s.io/api/apps/v1beta1"
)
func (c *Cluster) listPods() ([]v1.Pod, error) {
@@ -96,12 +97,12 @@ func (c *Cluster) unregisterPodSubscriber(podName spec.NamespacedName) {
delete(c.podSubscribers, podName)
}
func (c *Cluster) registerPodSubscriber(podName spec.NamespacedName) chan spec.PodEvent {
func (c *Cluster) registerPodSubscriber(podName spec.NamespacedName) chan PodEvent {
c.logger.Debugf("subscribing to pod %q", podName)
c.podSubscribersMu.Lock()
defer c.podSubscribersMu.Unlock()
ch := make(chan spec.PodEvent)
ch := make(chan PodEvent)
if _, ok := c.podSubscribers[podName]; ok {
panic("pod '" + podName.String() + "' is already subscribed")
}
@@ -182,6 +183,8 @@ func (c *Cluster) masterCandidate(oldNodeName string) (*v1.Pod, error) {
func (c *Cluster) MigrateMasterPod(podName spec.NamespacedName) error {
var (
masterCandidatePod *v1.Pod
err error
eol bool
)
oldMaster, err := c.KubeClient.Pods(podName.Namespace).Get(podName.Name, metav1.GetOptions{})
@@ -192,9 +195,10 @@ func (c *Cluster) MigrateMasterPod(podName spec.NamespacedName) error {
c.logger.Infof("migrating master pod %q", podName)
if eol, err := c.podIsEndOfLife(oldMaster); err != nil {
if eol, err = c.podIsEndOfLife(oldMaster); err != nil {
return fmt.Errorf("could not get node %q: %v", oldMaster.Spec.NodeName, err)
} else if !eol {
}
if !eol {
c.logger.Debugf("pod is already on a live node")
return nil
}
@@ -205,41 +209,43 @@ func (c *Cluster) MigrateMasterPod(podName spec.NamespacedName) error {
}
// we must have a statefulset in the cluster for the migration to work
if c.Statefulset == nil {
sset, err := c.KubeClient.StatefulSets(c.Namespace).Get(c.statefulSetName(), metav1.GetOptions{})
if err != nil {
var sset *v1beta1.StatefulSet
if sset, err = c.KubeClient.StatefulSets(c.Namespace).Get(c.statefulSetName(),
metav1.GetOptions{}); err != nil {
return fmt.Errorf("could not retrieve cluster statefulset: %v", err)
}
c.Statefulset = sset
}
// 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 {
c.logger.Warningf("single master pod for cluster %q, migration will cause longer downtime of the master instance", c.clusterName())
} else {
masterCandidatePod, err = c.masterCandidate(oldMaster.Spec.NodeName)
if err != nil {
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)
}
} else {
c.logger.Warningf("single master pod for cluster %q, migration will cause longer downtime of the master instance", c.clusterName())
}
// there are two cases for each postgres cluster that has its master pod on the node to migrate from:
// - the cluster has some replicas - migrate one of those if necessary and failover to it
// - there are no replicas - just terminate the master and wait until it respawns
// in both cases the result is the new master up and running on a new node.
if masterCandidatePod != nil {
pod, err := c.movePodFromEndOfLifeNode(masterCandidatePod)
if err != nil {
return fmt.Errorf("could not move pod: %v", err)
}
masterCandidateName := util.NameFromMeta(pod.ObjectMeta)
if err := c.Switchover(oldMaster, masterCandidateName); err != nil {
return fmt.Errorf("could not failover to pod %q: %v", masterCandidateName, err)
}
} else {
if masterCandidatePod == nil {
if _, err = c.movePodFromEndOfLifeNode(oldMaster); err != nil {
return fmt.Errorf("could not move pod: %v", err)
}
return nil
}
if masterCandidatePod, err = c.movePodFromEndOfLifeNode(masterCandidatePod); err != nil {
return fmt.Errorf("could not move pod: %v", err)
}
masterCandidateName := util.NameFromMeta(masterCandidatePod.ObjectMeta)
if err := c.Switchover(oldMaster, masterCandidateName); err != nil {
return fmt.Errorf("could not failover to pod %q: %v", masterCandidateName, err)
}
return nil
}
@@ -281,12 +287,12 @@ func (c *Cluster) recreatePod(podName spec.NamespacedName) (*v1.Pod, error) {
if err := c.waitForPodDeletion(ch); err != nil {
return nil, err
}
if pod, err := c.waitForPodLabel(ch, stopChan, nil); err != nil {
pod, err := c.waitForPodLabel(ch, stopChan, nil)
if err != nil {
return nil, err
} else {
c.logger.Infof("pod %q has been recreated", podName)
return pod, nil
}
c.logger.Infof("pod %q has been recreated", podName)
return pod, nil
}
func (c *Cluster) recreatePods() error {
pkg/cluster/resources.go
+11 -35
View File
@@ -5,11 +5,11 @@ import (
"strconv"
"strings"
"k8s.io/api/apps/v1beta1"
"k8s.io/api/core/v1"
policybeta1 "k8s.io/api/policy/v1beta1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/pkg/api/v1"
"k8s.io/client-go/pkg/apis/apps/v1beta1"
policybeta1 "k8s.io/client-go/pkg/apis/policy/v1beta1"
"github.com/zalando-incubator/postgres-operator/pkg/util"
"github.com/zalando-incubator/postgres-operator/pkg/util/constants"
@@ -18,7 +18,7 @@ import (
)
const (
RollingUpdateStatefulsetAnnotationKey = "zalando-postgres-operator-rolling-update-required"
rollingUpdateStatefulsetAnnotationKey = "zalando-postgres-operator-rolling-update-required"
)
func (c *Cluster) listResources() error {
@@ -140,7 +140,7 @@ func (c *Cluster) setRollingUpdateFlagForStatefulSet(sset *v1beta1.StatefulSet,
if anno == nil {
anno = make(map[string]string)
}
anno[RollingUpdateStatefulsetAnnotationKey] = strconv.FormatBool(val)
anno[rollingUpdateStatefulsetAnnotationKey] = strconv.FormatBool(val)
sset.SetAnnotations(anno)
}
@@ -162,13 +162,13 @@ func (c *Cluster) getRollingUpdateFlagFromStatefulSet(sset *v1beta1.StatefulSet,
anno := sset.GetAnnotations()
flag = defaultValue
stringFlag, exists := anno[RollingUpdateStatefulsetAnnotationKey]
stringFlag, exists := anno[rollingUpdateStatefulsetAnnotationKey]
if exists {
var err error
if flag, err = strconv.ParseBool(stringFlag); err != nil {
c.logger.Warnf("error when parsing %q annotation for the statefulset %q: expected boolean value, got %q\n",
RollingUpdateStatefulsetAnnotationKey,
types.NamespacedName{sset.Namespace, sset.Name},
rollingUpdateStatefulsetAnnotationKey,
types.NamespacedName{Namespace: sset.Namespace, Name: sset.Name},
stringFlag)
flag = defaultValue
}
@@ -272,10 +272,10 @@ func (c *Cluster) replaceStatefulSet(newStatefulSet *v1beta1.StatefulSet) error
c.logger.Debugf("replacing statefulset")
// Delete the current statefulset without deleting the pods
orphanDepencies := true
deletePropagationPolicy := metav1.DeletePropagationOrphan
oldStatefulset := c.Statefulset
options := metav1.DeleteOptions{OrphanDependents: &orphanDepencies}
options := metav1.DeleteOptions{PropagationPolicy: &deletePropagationPolicy}
if err := c.KubeClient.StatefulSets(oldStatefulset.Namespace).Delete(oldStatefulset.Name, &options); err != nil {
return fmt.Errorf("could not delete statefulset %q: %v", statefulSetName, err)
}
@@ -483,7 +483,7 @@ func (c *Cluster) generateEndpointSubsets(role PostgresRole) []v1.EndpointSubset
if len(endPointAddresses) > 0 {
result = append(result, v1.EndpointSubset{
Addresses: endPointAddresses,
Ports: []v1.EndpointPort{{"postgresql", 5432, "TCP"}},
Ports: []v1.EndpointPort{{Name: "postgresql", Port: 5432, Protocol: "TCP"}},
})
} else if role == Master {
c.logger.Warningf("master is not running, generated master endpoint does not contain any addresses")
@@ -625,27 +625,3 @@ func (c *Cluster) GetStatefulSet() *v1beta1.StatefulSet {
func (c *Cluster) GetPodDisruptionBudget() *policybeta1.PodDisruptionBudget {
return c.PodDisruptionBudget
}
func (c *Cluster) createDatabases() error {
c.setProcessName("creating databases")
if len(c.Spec.Databases) == 0 {
return nil
}
if err := c.initDbConn(); err != nil {
return fmt.Errorf("could not init database connection")
}
defer func() {
if err := c.closeDbConn(); err != nil {
c.logger.Errorf("could not close database connection: %v", err)
}
}()
for datname, owner := range c.Spec.Databases {
if err := c.executeCreateDatabase(datname, owner); err != nil {
return err
}
}
return nil
}
pkg/cluster/sync.go
+118 -104
View File
@@ -2,11 +2,11 @@ package cluster
import (
"fmt"
"reflect"
"k8s.io/api/core/v1"
policybeta1 "k8s.io/api/policy/v1beta1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
policybeta1 "k8s.io/client-go/pkg/apis/policy/v1beta1"
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/constants"
@@ -16,7 +16,8 @@ import (
// Sync syncs the cluster, making sure the actual Kubernetes objects correspond to what is defined in the manifest.
// Unlike the update, sync does not error out if some objects do not exist and takes care of creating them.
func (c *Cluster) Sync(newSpec *spec.Postgresql) (err error) {
func (c *Cluster) Sync(newSpec *acidv1.Postgresql) error {
var err error
c.mu.Lock()
defer c.mu.Unlock()
var actions []Action = NoActions
@@ -26,15 +27,15 @@ func (c *Cluster) Sync(newSpec *spec.Postgresql) (err error) {
defer func() {
if err != nil {
c.logger.Warningf("error while syncing cluster state: %v", err)
c.setStatus(spec.ClusterStatusSyncFailed)
} else if c.Status != spec.ClusterStatusRunning {
c.setStatus(spec.ClusterStatusRunning)
c.setStatus(acidv1.ClusterStatusSyncFailed)
} else if c.Status != acidv1.ClusterStatusRunning {
c.setStatus(acidv1.ClusterStatusRunning)
}
}()
if err = c.initUsers(); err != nil {
err = fmt.Errorf("could not init users: %v", err)
return
return err
}
c.logger.Debugf("syncing secrets")
@@ -42,18 +43,18 @@ func (c *Cluster) Sync(newSpec *spec.Postgresql) (err error) {
//TODO: mind the secrets of the deleted/new users
if err = c.syncSecrets(); err != nil {
err = fmt.Errorf("could not sync secrets: %v", err)
return
return err
}
c.logger.Debugf("syncing services")
if actions, err = c.syncServices(); err != nil {
err = fmt.Errorf("could not resolve actions to sync services: %v", err)
return
return err
}
if err = c.applyActions(actions); err != nil {
err = fmt.Errorf("could not apply actions to sync services: %v", err)
return
return err
}
// potentially enlarge volumes before changing the statefulset. By doing that
@@ -65,38 +66,38 @@ func (c *Cluster) Sync(newSpec *spec.Postgresql) (err error) {
c.logger.Debugf("syncing persistent volumes")
if err = c.syncVolumes(); err != nil {
err = fmt.Errorf("could not sync persistent volumes: %v", err)
return
return err
}
c.logger.Debugf("syncing statefulsets")
if err = c.syncStatefulSet(); err != nil {
if !k8sutil.ResourceAlreadyExists(err) {
err = fmt.Errorf("could not sync statefulsets: %v", err)
return
return err
}
}
// create database objects unless we are running without pods or disabled that feature explicitely
// create database objects unless we are running without pods or disabled that feature explicitly
if !(c.databaseAccessDisabled() || c.getNumberOfInstances(&newSpec.Spec) <= 0) {
c.logger.Debugf("syncing roles")
if err = c.syncRoles(); err != nil {
err = fmt.Errorf("could not sync roles: %v", err)
return
return err
}
c.logger.Debugf("syncing databases")
if err = c.syncDatabases(); err != nil {
err = fmt.Errorf("could not sync databases: %v", err)
return
return err
}
}
c.logger.Debug("syncing pod disruption budgets")
if err = c.syncPodDisruptionBudget(false); err != nil {
err = fmt.Errorf("could not sync pod disruption budget: %v", err)
return
return err
}
return
return err
}
func (c *Cluster) syncServices() (actions []Action, err error) {
@@ -135,10 +136,13 @@ func (c *Cluster) applyActions(actions []Action) (err error) {
}
func (c *Cluster) syncService(role PostgresRole) ([]Action, error) {
var (
svc *v1.Service
err error
)
c.setProcessName("syncing %s service", role)
svc, err := c.KubeClient.Services(c.Namespace).Get(c.serviceName(role), metav1.GetOptions{})
if err == nil {
if svc, err = c.KubeClient.Services(c.Namespace).Get(c.serviceName(role), metav1.GetOptions{}); err == nil {
c.Services[role] = svc
desiredSvc := c.generateService(role, &c.Spec)
match, reason := k8sutil.SameService(svc, desiredSvc)
@@ -156,8 +160,8 @@ func (c *Cluster) syncService(role PostgresRole) ([]Action, error) {
} else if !k8sutil.ResourceNotFound(err) {
return NoActions, fmt.Errorf("could not get %s service: %v", role, err)
}
// no existing service, create new one
c.Services[role] = nil
c.logger.Infof("could not find the cluster's %s service", role)
actions, err := c.createService(role)
@@ -171,72 +175,78 @@ func (c *Cluster) syncService(role PostgresRole) ([]Action, error) {
}
func (c *Cluster) syncEndpoint(role PostgresRole) error {
var (
ep *v1.Endpoints
err error
)
c.setProcessName("syncing %s endpoint", role)
ep, err := c.KubeClient.Endpoints(c.Namespace).Get(c.endpointName(role), metav1.GetOptions{})
if err == nil {
if ep, err = c.KubeClient.Endpoints(c.Namespace).Get(c.endpointName(role), metav1.GetOptions{}); err == nil {
// TODO: No syncing of endpoints here, is this covered completely by updateService?
c.Endpoints[role] = ep
return nil
} else if !k8sutil.ResourceNotFound(err) {
}
if !k8sutil.ResourceNotFound(err) {
return fmt.Errorf("could not get %s endpoint: %v", role, err)
}
// no existing endpoint, create new one
c.Endpoints[role] = nil
c.logger.Infof("could not find the cluster's %s endpoint", role)
if ep, err := c.createEndpoint(role); err != nil {
if k8sutil.ResourceAlreadyExists(err) {
c.logger.Infof("%s endpoint %q already exists", role, util.NameFromMeta(ep.ObjectMeta))
ep, err := c.KubeClient.Endpoints(c.Namespace).Get(c.endpointName(role), metav1.GetOptions{})
if err == nil {
c.Endpoints[role] = ep
} else {
c.logger.Infof("could not fetch existing %s endpoint: %v", role, err)
}
} else {
if ep, err = c.createEndpoint(role); err == nil {
c.logger.Infof("created missing %s endpoint %q", role, util.NameFromMeta(ep.ObjectMeta))
} else {
if !k8sutil.ResourceAlreadyExists(err) {
return fmt.Errorf("could not create missing %s endpoint: %v", role, err)
}
} else {
c.logger.Infof("created missing %s endpoint %q", role, util.NameFromMeta(ep.ObjectMeta))
c.Endpoints[role] = ep
c.logger.Infof("%s endpoint %q already exists", role, util.NameFromMeta(ep.ObjectMeta))
if ep, err = c.KubeClient.Endpoints(c.Namespace).Get(c.endpointName(role), metav1.GetOptions{}); err != nil {
return fmt.Errorf("could not fetch existing %s endpoint: %v", role, err)
}
}
c.Endpoints[role] = ep
return nil
}
func (c *Cluster) syncPodDisruptionBudget(isUpdate bool) error {
pdb, err := c.KubeClient.PodDisruptionBudgets(c.Namespace).Get(c.podDisruptionBudgetName(), metav1.GetOptions{})
if err == nil {
var (
pdb *policybeta1.PodDisruptionBudget
err error
)
if pdb, err = c.KubeClient.PodDisruptionBudgets(c.Namespace).Get(c.podDisruptionBudgetName(), metav1.GetOptions{}); err == nil {
c.PodDisruptionBudget = pdb
newPDB := c.generatePodDisruptionBudget()
if match, reason := k8sutil.SamePDB(pdb, newPDB); !match {
c.logPDBChanges(pdb, newPDB, isUpdate, reason)
if err := c.updatePodDisruptionBudget(newPDB); err != nil {
if err = c.updatePodDisruptionBudget(newPDB); err != nil {
return err
}
} else {
c.PodDisruptionBudget = pdb
}
return nil
} else if !k8sutil.ResourceNotFound(err) {
}
if !k8sutil.ResourceNotFound(err) {
return fmt.Errorf("could not get pod disruption budget: %v", err)
}
// no existing pod disruption budget, create new one
c.PodDisruptionBudget = nil
c.logger.Infof("could not find the cluster's pod disruption budget")
if pdb, err = c.createPodDisruptionBudget(); err != nil {
if k8sutil.ResourceAlreadyExists(err) {
c.logger.Infof("pod disruption budget %q already exists", util.NameFromMeta(pdb.ObjectMeta))
} else {
if !k8sutil.ResourceAlreadyExists(err) {
return fmt.Errorf("could not create pod disruption budget: %v", err)
}
} else {
c.logger.Infof("created missing pod disruption budget %q", util.NameFromMeta(pdb.ObjectMeta))
c.PodDisruptionBudget = pdb
c.logger.Infof("pod disruption budget %q already exists", util.NameFromMeta(pdb.ObjectMeta))
if pdb, err = c.KubeClient.PodDisruptionBudgets(c.Namespace).Get(c.podDisruptionBudgetName(), metav1.GetOptions{}); err != nil {
return fmt.Errorf("could not fetch existing %q pod disruption budget", util.NameFromMeta(pdb.ObjectMeta))
}
}
c.logger.Infof("created missing pod disruption budget %q", util.NameFromMeta(pdb.ObjectMeta))
c.PodDisruptionBudget = pdb
return nil
}
@@ -270,7 +280,9 @@ func (c *Cluster) syncStatefulSet() error {
podsRollingUpdateRequired = (len(pods) > 0)
if podsRollingUpdateRequired {
c.logger.Warningf("found pods from the previous statefulset: trigger rolling update")
c.applyRollingUpdateFlagforStatefulSet(podsRollingUpdateRequired)
if err := c.applyRollingUpdateFlagforStatefulSet(podsRollingUpdateRequired); err != nil {
return fmt.Errorf("could not set rolling update flag for the statefulset: %v", err)
}
}
c.logger.Infof("created missing statefulset %q", util.NameFromMeta(sset.ObjectMeta))
@@ -330,6 +342,11 @@ func (c *Cluster) syncStatefulSet() error {
// checkAndSetGlobalPostgreSQLConfiguration checks whether cluster-wide API parameters
// (like max_connections) has changed and if necessary sets it via the Patroni API
func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration() error {
var (
err error
pods []v1.Pod
)
// we need to extract those options from the cluster manifest.
optionsToSet := make(map[string]string)
pgOptions := c.Spec.Parameters
@@ -340,47 +357,55 @@ func (c *Cluster) checkAndSetGlobalPostgreSQLConfiguration() error {
}
}
if len(optionsToSet) > 0 {
pods, err := c.listPods()
if err != nil {
return err
}
if len(pods) == 0 {
return fmt.Errorf("could not call Patroni API: cluster has no pods")
}
for _, pod := range pods {
podName := util.NameFromMeta(pod.ObjectMeta)
c.logger.Debugf("calling Patroni API on a pod %s to set the following Postgres options: %v",
podName, optionsToSet)
if err := c.patroni.SetPostgresParameters(&pod, optionsToSet); err == nil {
return nil
} else {
c.logger.Warningf("could not patch postgres parameters with a pod %s: %v", podName, err)
}
}
return fmt.Errorf("could not reach Patroni API to set Postgres options: failed on every pod (%d total)",
len(pods))
if len(optionsToSet) == 0 {
return nil
}
return nil
if pods, err = c.listPods(); err != nil {
return err
}
if len(pods) == 0 {
return fmt.Errorf("could not call Patroni API: cluster has no pods")
}
// try all pods until the first one that is successful, as it doesn't matter which pod
// carries the request to change configuration through
for _, pod := range pods {
podName := util.NameFromMeta(pod.ObjectMeta)
c.logger.Debugf("calling Patroni API on a pod %s to set the following Postgres options: %v",
podName, optionsToSet)
if err = c.patroni.SetPostgresParameters(&pod, optionsToSet); err == nil {
return nil
}
c.logger.Warningf("could not patch postgres parameters with a pod %s: %v", podName, err)
}
return fmt.Errorf("could not reach Patroni API to set Postgres options: failed on every pod (%d total)",
len(pods))
}
func (c *Cluster) syncSecrets() error {
var (
err error
secret *v1.Secret
)
c.setProcessName("syncing secrets")
secrets := c.generateUserSecrets()
for secretUsername, secretSpec := range secrets {
secret, err := c.KubeClient.Secrets(secretSpec.Namespace).Create(secretSpec)
if secret, err = c.KubeClient.Secrets(secretSpec.Namespace).Create(secretSpec); err == nil {
c.Secrets[secret.UID] = secret
c.logger.Debugf("created new secret %q, uid: %q", util.NameFromMeta(secret.ObjectMeta), secret.UID)
continue
}
if k8sutil.ResourceAlreadyExists(err) {
var userMap map[string]spec.PgUser
curSecret, err2 := c.KubeClient.Secrets(secretSpec.Namespace).Get(secretSpec.Name, metav1.GetOptions{})
if err2 != nil {
return fmt.Errorf("could not get current secret: %v", err2)
if secret, err = c.KubeClient.Secrets(secretSpec.Namespace).Get(secretSpec.Name, metav1.GetOptions{}); err != nil {
return fmt.Errorf("could not get current secret: %v", err)
}
if secretUsername != string(curSecret.Data["username"]) {
if secretUsername != string(secret.Data["username"]) {
c.logger.Warningf("secret %q does not contain the role %q", secretSpec.Name, secretUsername)
continue
}
c.logger.Debugf("secret %q already exists, fetching its password", util.NameFromMeta(curSecret.ObjectMeta))
c.logger.Debugf("secret %q already exists, fetching its password", util.NameFromMeta(secret.ObjectMeta))
if secretUsername == c.systemUsers[constants.SuperuserKeyName].Name {
secretUsername = constants.SuperuserKeyName
userMap = c.systemUsers
@@ -392,35 +417,28 @@ func (c *Cluster) syncSecrets() error {
}
pwdUser := userMap[secretUsername]
// if this secret belongs to the infrastructure role and the password has changed - replace it in the secret
if pwdUser.Password != string(curSecret.Data["password"]) && pwdUser.Origin == spec.RoleOriginInfrastructure {
if pwdUser.Password != string(secret.Data["password"]) && pwdUser.Origin == spec.RoleOriginInfrastructure {
c.logger.Debugf("updating the secret %q from the infrastructure roles", secretSpec.Name)
if _, err := c.KubeClient.Secrets(secretSpec.Namespace).Update(secretSpec); err != nil {
if _, err = c.KubeClient.Secrets(secretSpec.Namespace).Update(secretSpec); err != nil {
return fmt.Errorf("could not update infrastructure role secret for role %q: %v", secretUsername, err)
}
} else {
// for non-infrastructure role - update the role with the password from the secret
pwdUser.Password = string(curSecret.Data["password"])
pwdUser.Password = string(secret.Data["password"])
userMap[secretUsername] = pwdUser
}
continue
} else {
if err != nil {
return fmt.Errorf("could not create secret for user %q: %v", secretUsername, err)
}
c.Secrets[secret.UID] = secret
c.logger.Debugf("created new secret %q, uid: %q", util.NameFromMeta(secret.ObjectMeta), secret.UID)
return fmt.Errorf("could not create secret for user %q: %v", secretUsername, err)
}
}
return nil
}
func (c *Cluster) syncRoles() error {
func (c *Cluster) syncRoles() (err error) {
c.setProcessName("syncing roles")
var (
err error
dbUsers spec.PgUserMap
userNames []string
)
@@ -429,9 +447,14 @@ func (c *Cluster) syncRoles() error {
if err != nil {
return fmt.Errorf("could not init db connection: %v", err)
}
defer func() {
if err := c.closeDbConn(); err != nil {
c.logger.Errorf("could not close db connection: %v", err)
if err2 := c.closeDbConn(); err2 != nil {
if err == nil {
err = fmt.Errorf("could not close database connection: %v", err2)
} else {
err = fmt.Errorf("could not close database connection: %v (prior error: %v)", err2, err)
}
}
}()
@@ -462,7 +485,7 @@ func (c *Cluster) syncVolumes() error {
if !act {
return nil
}
if err := c.resizeVolumes(c.Spec.Volume, []volumes.VolumeResizer{&volumes.EBSVolumeResizer{}}); err != nil {
if err := c.resizeVolumes(c.Spec.Volume, []volumes.VolumeResizer{&volumes.EBSVolumeResizer{AWSRegion: c.OpConfig.AWSRegion}}); err != nil {
return fmt.Errorf("could not sync volumes: %v", err)
}
@@ -471,15 +494,6 @@ func (c *Cluster) syncVolumes() error {
return nil
}
func (c *Cluster) samePDBWith(pdb *policybeta1.PodDisruptionBudget) (match bool, reason string) {
match = reflect.DeepEqual(pdb.Spec, c.PodDisruptionBudget.Spec)
if !match {
reason = "new service spec doesn't match the current one"
}
return
}
func (c *Cluster) syncDatabases() error {
c.setProcessName("syncing databases")
pkg/cluster/types.go
+57
View File
@@ -1,5 +1,14 @@
package cluster
import (
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
type PostgresRole string
@@ -10,3 +19,51 @@ const (
// Replica role
Replica PostgresRole = "replica"
)
type PodEventType string
// Possible values for the EventType
const (
PodEventAdd PodEventType = "ADD"
PodEventUpdate PodEventType = "UPDATE"
PodEventDelete PodEventType = "DELETE"
)
// PodEvent describes the event for a single Pod
type PodEvent struct {
ResourceVersion string
PodName types.NamespacedName
PrevPod *v1.Pod
CurPod *v1.Pod
EventType PodEventType
}
// Process describes process of the cluster
type Process struct {
Name string
StartTime time.Time
}
// WorkerStatus describes status of the worker
type WorkerStatus struct {
CurrentCluster types.NamespacedName
CurrentProcess Process
}
// ClusterStatus describes status of the cluster
type ClusterStatus struct {
Team string
Cluster string
MasterService *v1.Service
ReplicaService *v1.Service
MasterEndpoint *v1.Endpoints
ReplicaEndpoint *v1.Endpoints
StatefulSet *v1beta1.StatefulSet
PodDisruptionBudget *policybeta1.PodDisruptionBudget
CurrentProcess Process
Worker uint32
Status acidv1.PostgresStatus
Spec acidv1.PostgresSpec
Error error
}
pkg/cluster/util.go
+27 -22
View File
@@ -11,12 +11,14 @@ import (
"strings"
"time"
"k8s.io/api/apps/v1beta1"
"k8s.io/api/core/v1"
policybeta1 "k8s.io/api/policy/v1beta1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/client-go/pkg/api/v1"
"k8s.io/client-go/pkg/apis/apps/v1beta1"
policybeta1 "k8s.io/client-go/pkg/apis/policy/v1beta1"
acidzalando "github.com/zalando-incubator/postgres-operator/pkg/apis/acid.zalan.do"
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/constants"
@@ -37,7 +39,7 @@ type SecretOauthTokenGetter struct {
OAuthTokenSecretName spec.NamespacedName
}
func NewSecretOauthTokenGetter(kubeClient *k8sutil.KubernetesClient,
func newSecretOauthTokenGetter(kubeClient *k8sutil.KubernetesClient,
OAuthTokenSecretName spec.NamespacedName) *SecretOauthTokenGetter {
return &SecretOauthTokenGetter{kubeClient, OAuthTokenSecretName}
}
@@ -80,7 +82,8 @@ func (c *Cluster) isSystemUsername(username string) bool {
func isValidFlag(flag string) bool {
for _, validFlag := range []string{constants.RoleFlagSuperuser, constants.RoleFlagLogin, constants.RoleFlagCreateDB,
constants.RoleFlagInherit, constants.RoleFlagReplication, constants.RoleFlagByPassRLS} {
constants.RoleFlagInherit, constants.RoleFlagReplication, constants.RoleFlagByPassRLS,
constants.RoleFlagCreateRole} {
if flag == validFlag || flag == "NO"+validFlag {
return true
}
@@ -215,17 +218,19 @@ func (c *Cluster) logServiceChanges(role PostgresRole, old, new *v1.Service, isU
}
}
func (c *Cluster) logVolumeChanges(old, new spec.Volume) {
func (c *Cluster) logVolumeChanges(old, new acidv1.Volume) {
c.logger.Infof("volume specification has been changed")
c.logger.Debugf("diff\n%s\n", util.PrettyDiff(old, new))
}
func (c *Cluster) getTeamMembers() ([]string, error) {
if c.Spec.TeamID == "" {
func (c *Cluster) getTeamMembers(teamID string) ([]string, error) {
if teamID == "" {
return nil, fmt.Errorf("no teamId specified")
}
if !c.OpConfig.EnableTeamsAPI {
c.logger.Debug("team API is disabled, returning empty list of members")
c.logger.Debugf("team API is disabled, returning empty list of members for team %q", teamID)
return []string{}, nil
}
@@ -235,16 +240,16 @@ func (c *Cluster) getTeamMembers() ([]string, error) {
return []string{}, nil
}
teamInfo, err := c.teamsAPIClient.TeamInfo(c.Spec.TeamID, token)
teamInfo, err := c.teamsAPIClient.TeamInfo(teamID, token)
if err != nil {
c.logger.Warnf("could not get team info, returning empty list of team members: %v", err)
c.logger.Warnf("could not get team info for team %q, returning empty list of team members: %v", teamID, err)
return []string{}, nil
}
return teamInfo.Members, nil
}
func (c *Cluster) waitForPodLabel(podEvents chan spec.PodEvent, stopChan chan struct{}, role *PostgresRole) (*v1.Pod, error) {
func (c *Cluster) waitForPodLabel(podEvents chan PodEvent, stopChan chan struct{}, role *PostgresRole) (*v1.Pod, error) {
timeout := time.After(c.OpConfig.PodLabelWaitTimeout)
for {
select {
@@ -266,12 +271,12 @@ func (c *Cluster) waitForPodLabel(podEvents chan spec.PodEvent, stopChan chan st
}
}
func (c *Cluster) waitForPodDeletion(podEvents chan spec.PodEvent) error {
func (c *Cluster) waitForPodDeletion(podEvents chan PodEvent) error {
timeout := time.After(c.OpConfig.PodDeletionWaitTimeout)
for {
select {
case podEvent := <-podEvents:
if podEvent.EventType == spec.EventDelete {
if podEvent.EventType == PodEventDelete {
return nil
}
case <-timeout:
@@ -385,7 +390,7 @@ func (c *Cluster) waitStatefulsetPodsReady() error {
}
// Returns labels used to create or list k8s objects such as pods
// For backward compatability, shouldAddExtraLabels must be false
// For backward compatibility, shouldAddExtraLabels must be false
// when listing k8s objects. See operator PR #252
func (c *Cluster) labelsSet(shouldAddExtraLabels bool) labels.Set {
lbls := make(map[string]string)
@@ -403,7 +408,7 @@ func (c *Cluster) labelsSet(shouldAddExtraLabels bool) labels.Set {
}
func (c *Cluster) labelsSelector() *metav1.LabelSelector {
return &metav1.LabelSelector{c.labelsSet(false), nil}
return &metav1.LabelSelector{MatchLabels: c.labelsSet(false), MatchExpressions: nil}
}
func (c *Cluster) roleLabelsSet(role PostgresRole) labels.Set {
@@ -437,18 +442,18 @@ func (c *Cluster) credentialSecretNameForCluster(username string, clusterName st
return c.OpConfig.SecretNameTemplate.Format(
"username", strings.Replace(username, "_", "-", -1),
"cluster", clusterName,
"tprkind", constants.CRDKind,
"tprgroup", constants.CRDGroup)
"tprkind", acidv1.PostgresCRDResourceKind,
"tprgroup", acidzalando.GroupName)
}
func masterCandidate(replicas []spec.NamespacedName) spec.NamespacedName {
return replicas[rand.Intn(len(replicas))]
}
func cloneSpec(from *spec.Postgresql) (*spec.Postgresql, error) {
func cloneSpec(from *acidv1.Postgresql) (*acidv1.Postgresql, error) {
var (
buf bytes.Buffer
result *spec.Postgresql
result *acidv1.Postgresql
err error
)
enc := gob.NewEncoder(&buf)
@@ -462,13 +467,13 @@ func cloneSpec(from *spec.Postgresql) (*spec.Postgresql, error) {
return result, nil
}
func (c *Cluster) setSpec(newSpec *spec.Postgresql) {
func (c *Cluster) setSpec(newSpec *acidv1.Postgresql) {
c.specMu.Lock()
c.Postgresql = *newSpec
c.specMu.Unlock()
}
func (c *Cluster) GetSpec() (*spec.Postgresql, error) {
func (c *Cluster) GetSpec() (*acidv1.Postgresql, error) {
c.specMu.RLock()
defer c.specMu.RUnlock()
return cloneSpec(&c.Postgresql)
pkg/cluster/volumes.go
+15 -9
View File
@@ -5,10 +5,11 @@ import (
"strconv"
"strings"
"k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/api/resource"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/client-go/pkg/api/v1"
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/constants"
@@ -88,10 +89,11 @@ func (c *Cluster) listPersistentVolumes() ([]*v1.PersistentVolume, error) {
}
// resizeVolumes resize persistent volumes compatible with the given resizer interface
func (c *Cluster) resizeVolumes(newVolume spec.Volume, resizers []volumes.VolumeResizer) error {
func (c *Cluster) resizeVolumes(newVolume acidv1.Volume, resizers []volumes.VolumeResizer) error {
c.setProcessName("resizing volumes")
totalCompatible := 0
var totalIncompatible int
newQuantity, err := resource.ParseQuantity(newVolume.Size)
if err != nil {
return fmt.Errorf("could not parse volume size: %v", err)
@@ -100,7 +102,6 @@ func (c *Cluster) resizeVolumes(newVolume spec.Volume, resizers []volumes.Volume
if err != nil {
return fmt.Errorf("could not list persistent volumes: %v", err)
}
for _, pv := range pvs {
volumeSize := quantityToGigabyte(pv.Spec.Capacity[v1.ResourceStorage])
if volumeSize >= newSize {
@@ -109,11 +110,12 @@ func (c *Cluster) resizeVolumes(newVolume spec.Volume, resizers []volumes.Volume
}
continue
}
compatible := false
for _, resizer := range resizers {
if !resizer.VolumeBelongsToProvider(pv) {
continue
}
totalCompatible++
compatible = true
if !resizer.IsConnectedToProvider() {
err := resizer.ConnectToProvider()
if err != nil {
@@ -146,14 +148,18 @@ func (c *Cluster) resizeVolumes(newVolume spec.Volume, resizers []volumes.Volume
}
c.logger.Debugf("successfully updated persistent volume %q", pv.Name)
}
if !compatible {
c.logger.Warningf("volume %q is incompatible with all available resizing providers", pv.Name)
totalIncompatible++
}
}
if len(pvs) > 0 && totalCompatible == 0 {
return fmt.Errorf("could not resize EBS volumes: persistent volumes are not compatible with existing resizing providers")
if totalIncompatible > 0 {
return fmt.Errorf("could not resize EBS volumes: some persistent volumes are not compatible with existing resizing providers")
}
return nil
}
func (c *Cluster) volumesNeedResizing(newVolume spec.Volume) (bool, error) {
func (c *Cluster) volumesNeedResizing(newVolume acidv1.Volume) (bool, error) {
vols, manifestSize, err := c.listVolumesWithManifestSize(newVolume)
if err != nil {
return false, err
@@ -167,7 +173,7 @@ func (c *Cluster) volumesNeedResizing(newVolume spec.Volume) (bool, error) {
return false, nil
}
func (c *Cluster) listVolumesWithManifestSize(newVolume spec.Volume) ([]*v1.PersistentVolume, int64, error) {
func (c *Cluster) listVolumesWithManifestSize(newVolume acidv1.Volume) ([]*v1.PersistentVolume, int64, error) {
newSize, err := resource.ParseQuantity(newVolume.Size)
if err != nil {
return nil, 0, fmt.Errorf("could not parse volume size from the manifest: %v", err)