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

Use code-generation for CRD API and deepcopy methods (#369)

Browse Source 
Client-go provides a https://github.com/kubernetes/code-generator package in order to provide the API to work with CRDs similar to the one available for built-in types, i.e. Pods, Statefulsets and so on.

Use this package to generate deepcopy methods (required for CRDs), instead of using an external deepcopy package; we also generate APIs used to manipulate both Postgres and OperatorConfiguration CRDs, as well as informers and listers for the Postgres CRD, instead of using generic informers and CRD REST API; by using generated code we can get rid of some custom and obscure CRD-related code and use a better API.

All generated code resides in /pkg/generated, with an exception of zz_deepcopy.go in apis/acid.zalan.do/v1

Rename postgres-operator-configuration CRD to OperatorConfiguration, since the former broke naming convention in the code-generator.

Moved Postgresql, PostgresqlList, OperatorConfiguration and OperatorConfigurationList and other types used by them into

Change the type of  the Error field in the Postgresql crd to a string, so that client-go could generate a deepcopy for it.

Use generated code to set status of CRD objects as well. Right now this is done with patch, however, Kubernetes 1.11 introduces the /status subresources, allowing us to set the status with
the special updateStatus call in the future. For now, we keep the code that is compatible with earlier versions of Kubernetes.

Rename postgresql.go to database.go and status.go to logs_and_api.go to reflect the purpose of each of those files.

Update client-go dependencies.

Minor reformatting and renaming.
This commit is contained in:
Oleksii Kliukin
2018-08-15 17:22:25 +02:00
committed by GitHub
parent 6e8dcabac7
commit e1ed4b847d
66 changed files with 3285 additions and 966 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/controller/controller.go
+12 -14
View File
@@ -21,6 +21,8 @@ import (
"github.com/zalando-incubator/postgres-operator/pkg/util/constants"
"github.com/zalando-incubator/postgres-operator/pkg/util/k8sutil"
"github.com/zalando-incubator/postgres-operator/pkg/util/ringlog"
acidv1informer "github.com/zalando-incubator/postgres-operator/pkg/generated/informers/externalversions/acid.zalan.do/v1"
)
// Controller represents operator controller
@@ -46,7 +48,7 @@ type Controller struct {
postgresqlInformer cache.SharedIndexInformer
podInformer cache.SharedIndexInformer
nodesInformer cache.SharedIndexInformer
podCh chan spec.PodEvent
podCh chan cluster.PodEvent
clusterEventQueues []*cache.FIFO // [workerID]Queue
lastClusterSyncTime int64
@@ -74,7 +76,7 @@ func NewController(controllerConfig *spec.ControllerConfig) *Controller {
clusterHistory: make(map[spec.NamespacedName]ringlog.RingLogger),
teamClusters: make(map[string][]spec.NamespacedName),
stopCh: make(chan struct{}),
podCh: make(chan spec.PodEvent),
podCh: make(chan cluster.PodEvent),
}
logger.Hooks.Add(c)
@@ -227,9 +229,9 @@ func (c *Controller) initController() {
}
} else {
c.initOperatorConfig()
c.initPodServiceAccount()
c.initRoleBinding()
}
c.initPodServiceAccount()
c.initRoleBinding()
c.modifyConfigFromEnvironment()
@@ -256,7 +258,7 @@ func (c *Controller) initController() {
c.workerLogs = make(map[uint32]ringlog.RingLogger, c.opConfig.Workers)
for i := range c.clusterEventQueues {
c.clusterEventQueues[i] = cache.NewFIFO(func(obj interface{}) (string, error) {
e, ok := obj.(spec.ClusterEvent)
e, ok := obj.(ClusterEvent)
if !ok {
return "", fmt.Errorf("could not cast to ClusterEvent")
}
@@ -269,13 +271,10 @@ func (c *Controller) initController() {
}
func (c *Controller) initSharedInformers() {
// Postgresqls
c.postgresqlInformer = cache.NewSharedIndexInformer(
&cache.ListWatch{
ListFunc: c.clusterListFunc,
WatchFunc: c.clusterWatchFunc,
},
&spec.Postgresql{},
c.postgresqlInformer = acidv1informer.NewPostgresqlInformer(
c.KubeClient.AcidV1ClientSet,
c.opConfig.WatchedNamespace,
constants.QueueResyncPeriodTPR,
cache.Indexers{})
@@ -345,7 +344,6 @@ func (c *Controller) Run(stopCh <-chan struct{}, wg *sync.WaitGroup) {
go c.apiserver.Run(stopCh, wg)
go c.kubeNodesInformer(stopCh, wg)
c.logger.Info("started working in background")
}
@@ -361,7 +359,7 @@ func (c *Controller) runPostgresqlInformer(stopCh <-chan struct{}, wg *sync.Wait
c.postgresqlInformer.Run(stopCh)
}
func queueClusterKey(eventType spec.EventType, uid types.UID) string {
func queueClusterKey(eventType EventType, uid types.UID) string {
return fmt.Sprintf("%s-%s", eventType, uid)
}
pkg/controller/status.go → pkg/controller/logs_and_api.go
+5 -4
View File
@@ -11,10 +11,11 @@ import (
"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"
"k8s.io/apimachinery/pkg/types"
)
// ClusterStatus provides status of the cluster
func (c *Controller) ClusterStatus(team, namespace, cluster string) (*spec.ClusterStatus, error) {
func (c *Controller) ClusterStatus(team, namespace, cluster string) (*cluster.ClusterStatus, error) {
clusterName := spec.NamespacedName{
Namespace: namespace,
@@ -196,7 +197,7 @@ func (c *Controller) GetWorkersCnt() uint32 {
}
//WorkerStatus provides status of the worker
func (c *Controller) WorkerStatus(workerID uint32) (*spec.WorkerStatus, error) {
func (c *Controller) WorkerStatus(workerID uint32) (*cluster.WorkerStatus, error) {
obj, ok := c.curWorkerCluster.Load(workerID)
if !ok || obj == nil {
return nil, nil
@@ -207,8 +208,8 @@ func (c *Controller) WorkerStatus(workerID uint32) (*spec.WorkerStatus, error) {
return nil, fmt.Errorf("could not cast to Cluster struct")
}
return &spec.WorkerStatus{
CurrentCluster: util.NameFromMeta(cl.ObjectMeta),
return &cluster.WorkerStatus{
CurrentCluster: types.NamespacedName(util.NameFromMeta(cl.ObjectMeta)),
CurrentProcess: cl.GetCurrentProcess(),
}, nil
}
pkg/controller/operator_config.go
+6 -19
View File
@@ -1,40 +1,27 @@
package controller
import (
"encoding/json"
"fmt"
"time"
acidv1 "github.com/zalando-incubator/postgres-operator/pkg/apis/acid.zalan.do/v1"
"github.com/zalando-incubator/postgres-operator/pkg/util/config"
"github.com/zalando-incubator/postgres-operator/pkg/util/constants"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
)
func (c *Controller) readOperatorConfigurationFromCRD(configObjectNamespace, configObjectName string) (*config.OperatorConfiguration, error) {
var (
opConfig config.OperatorConfiguration
)
func (c *Controller) readOperatorConfigurationFromCRD(configObjectNamespace, configObjectName string) (*acidv1.OperatorConfiguration, error) {
req := c.KubeClient.CRDREST.Get().
Name(configObjectName).
Namespace(configObjectNamespace).
Resource(constants.OperatorConfigCRDResource).
VersionedParams(&metav1.ListOptions{ResourceVersion: "0"}, metav1.ParameterCodec)
data, err := req.DoRaw()
config, err := c.KubeClient.AcidV1ClientSet.AcidV1().OperatorConfigurations(configObjectNamespace).Get(configObjectName, metav1.GetOptions{})
if err != nil {
return nil, fmt.Errorf("could not get operator configuration object %s: %v", configObjectName, err)
}
if err = json.Unmarshal(data, &opConfig); err != nil {
return nil, fmt.Errorf("could not unmarshal operator configuration object %s, %v", configObjectName, err)
return nil, fmt.Errorf("could not get operator configuration object %q: %v", configObjectName, err)
}
return &opConfig, nil
return config, nil
}
// importConfigurationFromCRD is a transitional function that converts CRD configuration to the one based on the configmap
func (c *Controller) importConfigurationFromCRD(fromCRD *config.OperatorConfigurationData) *config.Config {
func (c *Controller) importConfigurationFromCRD(fromCRD *acidv1.OperatorConfigurationData) *config.Config {
result := &config.Config{}
result.EtcdHost = fromCRD.EtcdHost
pkg/controller/pod.go
+9 -7
View File
@@ -6,8 +6,10 @@ import (
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/watch"
"github.com/zalando-incubator/postgres-operator/pkg/cluster"
"github.com/zalando-incubator/postgres-operator/pkg/spec"
"github.com/zalando-incubator/postgres-operator/pkg/util"
"k8s.io/apimachinery/pkg/types"
)
func (c *Controller) podListFunc(options metav1.ListOptions) (runtime.Object, error) {
@@ -30,7 +32,7 @@ func (c *Controller) podWatchFunc(options metav1.ListOptions) (watch.Interface,
return c.KubeClient.Pods(c.opConfig.WatchedNamespace).Watch(opts)
}
func (c *Controller) dispatchPodEvent(clusterName spec.NamespacedName, event spec.PodEvent) {
func (c *Controller) dispatchPodEvent(clusterName spec.NamespacedName, event cluster.PodEvent) {
c.clustersMu.RLock()
cluster, ok := c.clusters[clusterName]
c.clustersMu.RUnlock()
@@ -41,7 +43,7 @@ func (c *Controller) dispatchPodEvent(clusterName spec.NamespacedName, event spe
func (c *Controller) podAdd(obj interface{}) {
if pod, ok := obj.(*v1.Pod); ok {
c.preparePodEventForDispatch(pod, nil, spec.EventAdd)
c.preparePodEventForDispatch(pod, nil, cluster.PodEventAdd)
}
}
@@ -56,19 +58,19 @@ func (c *Controller) podUpdate(prev, cur interface{}) {
return
}
c.preparePodEventForDispatch(curPod, prevPod, spec.EventUpdate)
c.preparePodEventForDispatch(curPod, prevPod, cluster.PodEventUpdate)
}
func (c *Controller) podDelete(obj interface{}) {
if pod, ok := obj.(*v1.Pod); ok {
c.preparePodEventForDispatch(pod, nil, spec.EventDelete)
c.preparePodEventForDispatch(pod, nil, cluster.PodEventDelete)
}
}
func (c *Controller) preparePodEventForDispatch(curPod, prevPod *v1.Pod, event spec.EventType) {
podEvent := spec.PodEvent{
PodName: util.NameFromMeta(curPod.ObjectMeta),
func (c *Controller) preparePodEventForDispatch(curPod, prevPod *v1.Pod, event cluster.PodEventType) {
podEvent := cluster.PodEvent{
PodName: types.NamespacedName(util.NameFromMeta(curPod.ObjectMeta)),
CurPod: curPod,
PrevPod: prevPod,
EventType: event,
pkg/controller/postgresql.go
+56 -119
View File
@@ -1,7 +1,6 @@
package controller
import (
"encoding/json"
"fmt"
"reflect"
"strings"
@@ -12,15 +11,13 @@ import (
"github.com/Sirupsen/logrus"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/tools/cache"
acidv1 "github.com/zalando-incubator/postgres-operator/pkg/apis/acid.zalan.do/v1"
"github.com/zalando-incubator/postgres-operator/pkg/cluster"
"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"
"github.com/zalando-incubator/postgres-operator/pkg/util/k8sutil"
"github.com/zalando-incubator/postgres-operator/pkg/util/ringlog"
)
@@ -42,40 +39,20 @@ func (c *Controller) clusterResync(stopCh <-chan struct{}, wg *sync.WaitGroup) {
}
// clusterListFunc obtains a list of all PostgreSQL clusters
func (c *Controller) listClusters(options metav1.ListOptions) (*spec.PostgresqlList, error) {
var (
list spec.PostgresqlList
)
req := c.KubeClient.CRDREST.
Get().
Namespace(c.opConfig.WatchedNamespace).
Resource(constants.PostgresCRDResource).
VersionedParams(&options, metav1.ParameterCodec)
b, err := req.DoRaw()
func (c *Controller) listClusters(options metav1.ListOptions) (*acidv1.PostgresqlList, error) {
// TODO: use the SharedInformer cache instead of quering Kubernetes API directly.
list, err := c.KubeClient.AcidV1ClientSet.AcidV1().Postgresqls(c.opConfig.WatchedNamespace).List(options)
if err != nil {
c.logger.Errorf("could not get the list of postgresql CRD objects: %v", err)
return nil, err
c.logger.Errorf("could not list postgresql objects: %v", err)
}
if err = json.Unmarshal(b, &list); err != nil {
c.logger.Warningf("could not unmarshal list of clusters: %v", err)
}
return &list, err
}
// A separate function to be called from InitSharedInformers
func (c *Controller) clusterListFunc(options metav1.ListOptions) (runtime.Object, error) {
return c.listClusters(options)
return list, err
}
// clusterListAndSync lists all manifests and decides whether to run the sync or repair.
func (c *Controller) clusterListAndSync() error {
var (
err error
event spec.EventType
event EventType
)
currentTime := time.Now().Unix()
@@ -83,12 +60,12 @@ func (c *Controller) clusterListAndSync() error {
timeFromPreviousRepair := currentTime - atomic.LoadInt64(&c.lastClusterRepairTime)
if timeFromPreviousSync >= int64(c.opConfig.ResyncPeriod.Seconds()) {
event = spec.EventSync
event = EventSync
} else if timeFromPreviousRepair >= int64(c.opConfig.RepairPeriod.Seconds()) {
event = spec.EventRepair
event = EventRepair
}
if event != "" {
var list *spec.PostgresqlList
var list *acidv1.PostgresqlList
if list, err = c.listClusters(metav1.ListOptions{ResourceVersion: "0"}); err != nil {
return err
}
@@ -101,16 +78,17 @@ func (c *Controller) clusterListAndSync() error {
}
// queueEvents queues a sync or repair event for every cluster with a valid manifest
func (c *Controller) queueEvents(list *spec.PostgresqlList, event spec.EventType) {
func (c *Controller) queueEvents(list *acidv1.PostgresqlList, event EventType) {
var activeClustersCnt, failedClustersCnt, clustersToRepair int
for i, pg := range list.Items {
if pg.Error != nil {
// XXX: check the cluster status field instead
if pg.Error != "" {
failedClustersCnt++
continue
}
activeClustersCnt++
// check if that cluster needs repair
if event == spec.EventRepair {
if event == EventRepair {
if pg.Status.Success() {
continue
} else {
@@ -133,9 +111,9 @@ func (c *Controller) queueEvents(list *spec.PostgresqlList, event spec.EventType
} else {
c.logger.Infof("no clusters running")
}
if event == spec.EventRepair || event == spec.EventSync {
if event == EventRepair || event == EventSync {
atomic.StoreInt64(&c.lastClusterRepairTime, time.Now().Unix())
if event == spec.EventSync {
if event == EventSync {
atomic.StoreInt64(&c.lastClusterSyncTime, time.Now().Unix())
}
}
@@ -143,7 +121,7 @@ func (c *Controller) queueEvents(list *spec.PostgresqlList, event spec.EventType
func (c *Controller) acquireInitialListOfClusters() error {
var (
list *spec.PostgresqlList
list *acidv1.PostgresqlList
err error
clusterName spec.NamespacedName
)
@@ -153,7 +131,8 @@ func (c *Controller) acquireInitialListOfClusters() error {
}
c.logger.Debugf("acquiring initial list of clusters")
for _, pg := range list.Items {
if pg.Error != nil {
// XXX: check the cluster status field instead
if pg.Error != "" {
continue
}
clusterName = util.NameFromMeta(pg.ObjectMeta)
@@ -161,54 +140,11 @@ func (c *Controller) acquireInitialListOfClusters() error {
c.logger.Debugf("added new cluster: %q", clusterName)
}
// initiate initial sync of all clusters.
c.queueEvents(list, spec.EventSync)
c.queueEvents(list, EventSync)
return nil
}
type crdDecoder struct {
dec *json.Decoder
close func() error
}
func (d *crdDecoder) Close() {
if err := d.close(); err != nil {
fmt.Printf("error when closing CRDDecorer: %v\n", err)
}
}
func (d *crdDecoder) Decode() (action watch.EventType, object runtime.Object, err error) {
var e struct {
Type watch.EventType
Object spec.Postgresql
}
if err := d.dec.Decode(&e); err != nil {
return watch.Error, nil, err
}
return e.Type, &e.Object, nil
}
func (c *Controller) clusterWatchFunc(options metav1.ListOptions) (watch.Interface, error) {
options.Watch = true
// MIGRATION: FieldsSelectorParam(nil)
r, err := c.KubeClient.CRDREST.
Get().
Namespace(c.opConfig.WatchedNamespace).
Resource(constants.PostgresCRDResource).
VersionedParams(&options, metav1.ParameterCodec).
Stream()
if err != nil {
return nil, err
}
return watch.NewStreamWatcher(&crdDecoder{
dec: json.NewDecoder(r),
close: r.Close,
}), nil
}
func (c *Controller) addCluster(lg *logrus.Entry, clusterName spec.NamespacedName, pgSpec *spec.Postgresql) *cluster.Cluster {
func (c *Controller) addCluster(lg *logrus.Entry, clusterName spec.NamespacedName, pgSpec *acidv1.Postgresql) *cluster.Cluster {
cl := cluster.New(c.makeClusterConfig(), c.KubeClient, *pgSpec, lg)
cl.Run(c.stopCh)
teamName := strings.ToLower(cl.Spec.TeamID)
@@ -224,13 +160,13 @@ func (c *Controller) addCluster(lg *logrus.Entry, clusterName spec.NamespacedNam
return cl
}
func (c *Controller) processEvent(event spec.ClusterEvent) {
func (c *Controller) processEvent(event ClusterEvent) {
var clusterName spec.NamespacedName
var clHistory ringlog.RingLogger
lg := c.logger.WithField("worker", event.WorkerID)
if event.EventType == spec.EventAdd || event.EventType == spec.EventSync || event.EventType == spec.EventRepair {
if event.EventType == EventAdd || event.EventType == EventSync || event.EventType == EventRepair {
clusterName = util.NameFromMeta(event.NewSpec.ObjectMeta)
} else {
clusterName = util.NameFromMeta(event.OldSpec.ObjectMeta)
@@ -246,17 +182,17 @@ func (c *Controller) processEvent(event spec.ClusterEvent) {
defer c.curWorkerCluster.Store(event.WorkerID, nil)
if event.EventType == spec.EventRepair {
if event.EventType == EventRepair {
runRepair, lastOperationStatus := cl.NeedsRepair()
if !runRepair {
lg.Debugf("Observed cluster status %s, repair is not required", lastOperationStatus)
return
}
lg.Debugf("Observed cluster status %s, running sync scan to repair the cluster", lastOperationStatus)
event.EventType = spec.EventSync
event.EventType = EventSync
}
if event.EventType == spec.EventAdd || event.EventType == spec.EventUpdate || event.EventType == spec.EventSync {
if event.EventType == EventAdd || event.EventType == EventUpdate || event.EventType == EventSync {
// handle deprecated parameters by possibly assigning their values to the new ones.
if event.OldSpec != nil {
c.mergeDeprecatedPostgreSQLSpecParameters(&event.OldSpec.Spec)
@@ -273,7 +209,7 @@ func (c *Controller) processEvent(event spec.ClusterEvent) {
}
switch event.EventType {
case spec.EventAdd:
case EventAdd:
if clusterFound {
lg.Debugf("cluster already exists")
return
@@ -286,14 +222,14 @@ func (c *Controller) processEvent(event spec.ClusterEvent) {
c.curWorkerCluster.Store(event.WorkerID, cl)
if err := cl.Create(); err != nil {
cl.Error = fmt.Errorf("could not create cluster: %v", err)
cl.Error = fmt.Sprintf("could not create cluster: %v", err)
lg.Error(cl.Error)
return
}
lg.Infoln("cluster has been created")
case spec.EventUpdate:
case EventUpdate:
lg.Infoln("update of the cluster started")
if !clusterFound {
@@ -302,12 +238,12 @@ func (c *Controller) processEvent(event spec.ClusterEvent) {
}
c.curWorkerCluster.Store(event.WorkerID, cl)
if err := cl.Update(event.OldSpec, event.NewSpec); err != nil {
cl.Error = fmt.Errorf("could not update cluster: %v", err)
cl.Error = fmt.Sprintf("could not update cluster: %v", err)
lg.Error(cl.Error)
return
}
cl.Error = nil
cl.Error = ""
lg.Infoln("cluster has been updated")
clHistory.Insert(&spec.Diff{
@@ -315,7 +251,7 @@ func (c *Controller) processEvent(event spec.ClusterEvent) {
ProcessTime: time.Now(),
Diff: util.Diff(event.OldSpec, event.NewSpec),
})
case spec.EventDelete:
case EventDelete:
if !clusterFound {
lg.Errorf("unknown cluster: %q", clusterName)
return
@@ -345,7 +281,7 @@ func (c *Controller) processEvent(event spec.ClusterEvent) {
}()
lg.Infof("cluster has been deleted")
case spec.EventSync:
case EventSync:
lg.Infof("syncing of the cluster started")
// no race condition because a cluster is always processed by single worker
@@ -355,11 +291,11 @@ func (c *Controller) processEvent(event spec.ClusterEvent) {
c.curWorkerCluster.Store(event.WorkerID, cl)
if err := cl.Sync(event.NewSpec); err != nil {
cl.Error = fmt.Errorf("could not sync cluster: %v", err)
cl.Error = fmt.Sprintf("could not sync cluster: %v", err)
lg.Error(cl.Error)
return
}
cl.Error = nil
cl.Error = ""
lg.Infof("cluster has been synced")
}
@@ -382,7 +318,7 @@ func (c *Controller) processClusterEventsQueue(idx int, stopCh <-chan struct{},
c.logger.Errorf("error when processing cluster events queue: %v", err)
continue
}
event, ok := obj.(spec.ClusterEvent)
event, ok := obj.(ClusterEvent)
if !ok {
c.logger.Errorf("could not cast to ClusterEvent")
}
@@ -391,7 +327,7 @@ func (c *Controller) processClusterEventsQueue(idx int, stopCh <-chan struct{},
}
}
func (c *Controller) warnOnDeprecatedPostgreSQLSpecParameters(spec *spec.PostgresSpec) {
func (c *Controller) warnOnDeprecatedPostgreSQLSpecParameters(spec *acidv1.PostgresSpec) {
deprecate := func(deprecated, replacement string) {
c.logger.Warningf("Parameter %q is deprecated. Consider setting %q instead", deprecated, replacement)
@@ -421,7 +357,7 @@ func (c *Controller) warnOnDeprecatedPostgreSQLSpecParameters(spec *spec.Postgre
// mergeDeprecatedPostgreSQLSpecParameters modifies the spec passed to the cluster by setting current parameter
// values from the obsolete ones. Note: while the spec that is modified is a copy made in queueClusterEvent, it is
// still a shallow copy, so be extra careful not to modify values pointer fields point to, but copy them instead.
func (c *Controller) mergeDeprecatedPostgreSQLSpecParameters(spec *spec.PostgresSpec) *spec.PostgresSpec {
func (c *Controller) mergeDeprecatedPostgreSQLSpecParameters(spec *acidv1.PostgresSpec) *acidv1.PostgresSpec {
if (spec.UseLoadBalancer != nil || spec.ReplicaLoadBalancer != nil) &&
(spec.EnableReplicaLoadBalancer == nil && spec.EnableMasterLoadBalancer == nil) {
if spec.UseLoadBalancer != nil {
@@ -439,18 +375,18 @@ func (c *Controller) mergeDeprecatedPostgreSQLSpecParameters(spec *spec.Postgres
return spec
}
func (c *Controller) queueClusterEvent(informerOldSpec, informerNewSpec *spec.Postgresql, eventType spec.EventType) {
func (c *Controller) queueClusterEvent(informerOldSpec, informerNewSpec *acidv1.Postgresql, eventType EventType) {
var (
uid types.UID
clusterName spec.NamespacedName
clusterError error
clusterError string
)
if informerOldSpec != nil { //update, delete
uid = informerOldSpec.GetUID()
clusterName = util.NameFromMeta(informerOldSpec.ObjectMeta)
if eventType == spec.EventUpdate && informerNewSpec.Error == nil && informerOldSpec.Error != nil {
eventType = spec.EventSync
if eventType == EventUpdate && informerNewSpec.Error == "" && informerOldSpec.Error != "" {
eventType = EventSync
clusterError = informerNewSpec.Error
} else {
clusterError = informerOldSpec.Error
@@ -461,10 +397,10 @@ func (c *Controller) queueClusterEvent(informerOldSpec, informerNewSpec *spec.Po
clusterError = informerNewSpec.Error
}
if clusterError != nil && eventType != spec.EventDelete {
if clusterError != "" && eventType != EventDelete {
c.logger.
WithField("cluster-name", clusterName).
Debugf("skipping %q event for the invalid cluster: %v", eventType, clusterError)
Debugf("skipping %q event for the invalid cluster: %s", eventType, clusterError)
return
}
@@ -473,7 +409,7 @@ func (c *Controller) queueClusterEvent(informerOldSpec, informerNewSpec *spec.Po
// effect, the modified state will be returned together with subsequent events).
workerID := c.clusterWorkerID(clusterName)
clusterEvent := spec.ClusterEvent{
clusterEvent := ClusterEvent{
EventTime: time.Now(),
EventType: eventType,
UID: uid,
@@ -488,11 +424,11 @@ func (c *Controller) queueClusterEvent(informerOldSpec, informerNewSpec *spec.Po
}
lg.Infof("%q event has been queued", eventType)
if eventType != spec.EventDelete {
if eventType != EventDelete {
return
}
// A delete event discards all prior requests for that cluster.
for _, evType := range []spec.EventType{spec.EventAdd, spec.EventSync, spec.EventUpdate, spec.EventRepair} {
for _, evType := range []EventType{EventAdd, EventSync, EventUpdate, EventRepair} {
obj, exists, err := c.clusterEventQueues[workerID].GetByKey(queueClusterKey(evType, uid))
if err != nil {
lg.Warningf("could not get event from the queue: %v", err)
@@ -513,40 +449,41 @@ func (c *Controller) queueClusterEvent(informerOldSpec, informerNewSpec *spec.Po
}
func (c *Controller) postgresqlAdd(obj interface{}) {
pg, ok := obj.(*spec.Postgresql)
pg, ok := obj.(*acidv1.Postgresql)
if !ok {
c.logger.Errorf("could not cast to postgresql spec")
return
}
// We will not get multiple Add events for the same cluster
c.queueClusterEvent(nil, pg, spec.EventAdd)
c.queueClusterEvent(nil, pg, EventAdd)
}
func (c *Controller) postgresqlUpdate(prev, cur interface{}) {
pgOld, ok := prev.(*spec.Postgresql)
pgOld, ok := prev.(*acidv1.Postgresql)
if !ok {
c.logger.Errorf("could not cast to postgresql spec")
}
pgNew, ok := cur.(*spec.Postgresql)
pgNew, ok := cur.(*acidv1.Postgresql)
if !ok {
c.logger.Errorf("could not cast to postgresql spec")
}
// Avoid the inifinite recursion for status updates
if reflect.DeepEqual(pgOld.Spec, pgNew.Spec) {
return
}
c.queueClusterEvent(pgOld, pgNew, spec.EventUpdate)
c.queueClusterEvent(pgOld, pgNew, EventUpdate)
}
func (c *Controller) postgresqlDelete(obj interface{}) {
pg, ok := obj.(*spec.Postgresql)
pg, ok := obj.(*acidv1.Postgresql)
if !ok {
c.logger.Errorf("could not cast to postgresql spec")
return
}
c.queueClusterEvent(pg, nil, spec.EventDelete)
c.queueClusterEvent(pg, nil, EventDelete)
}
/*
@@ -555,7 +492,7 @@ func (c *Controller) postgresqlDelete(obj interface{}) {
The operator does not sync accounts/role bindings after creation.
*/
func (c *Controller) submitRBACCredentials(event spec.ClusterEvent) error {
func (c *Controller) submitRBACCredentials(event ClusterEvent) error {
namespace := event.NewSpec.GetNamespace()
if _, ok := c.namespacesWithDefinedRBAC.Load(namespace); ok {
pkg/controller/postgresql_test.go
+7 -6
View File
@@ -1,6 +1,7 @@
package controller
import (
acidv1 "github.com/zalando-incubator/postgres-operator/pkg/apis/acid.zalan.do/v1"
"github.com/zalando-incubator/postgres-operator/pkg/spec"
"reflect"
"testing"
@@ -16,21 +17,21 @@ func TestMergeDeprecatedPostgreSQLSpecParameters(t *testing.T) {
tests := []struct {
name string
in *spec.PostgresSpec
out *spec.PostgresSpec
in *acidv1.PostgresSpec
out *acidv1.PostgresSpec
error string
}{
{
"Check that old parameters propagate values to the new ones",
&spec.PostgresSpec{UseLoadBalancer: &True, ReplicaLoadBalancer: &True},
&spec.PostgresSpec{UseLoadBalancer: nil, ReplicaLoadBalancer: nil,
&acidv1.PostgresSpec{UseLoadBalancer: &True, ReplicaLoadBalancer: &True},
&acidv1.PostgresSpec{UseLoadBalancer: nil, ReplicaLoadBalancer: nil,
EnableMasterLoadBalancer: &True, EnableReplicaLoadBalancer: &True},
"New parameters should be set from the values of old ones",
},
{
"Check that new parameters are not set when both old and new ones are present",
&spec.PostgresSpec{UseLoadBalancer: &True, EnableMasterLoadBalancer: &False},
&spec.PostgresSpec{UseLoadBalancer: nil, EnableMasterLoadBalancer: &False},
&acidv1.PostgresSpec{UseLoadBalancer: &True, EnableMasterLoadBalancer: &False},
&acidv1.PostgresSpec{UseLoadBalancer: nil, EnableMasterLoadBalancer: &False},
"New parameters should remain unchanged when both old and new are present",
},
}
pkg/controller/types.go
+30
View File
@@ -0,0 +1,30 @@
package controller
import (
"k8s.io/apimachinery/pkg/types"
"time"
acidv1 "github.com/zalando-incubator/postgres-operator/pkg/apis/acid.zalan.do/v1"
)
// EventType contains type of the events for the TPRs and Pods received from Kubernetes
type EventType string
// Possible values for the EventType
const (
EventAdd EventType = "ADD"
EventUpdate EventType = "UPDATE"
EventDelete EventType = "DELETE"
EventSync EventType = "SYNC"
EventRepair EventType = "REPAIR"
)
// ClusterEvent carries the payload of the Cluster TPR events.
type ClusterEvent struct {
EventTime time.Time
UID types.UID
EventType EventType
OldSpec *acidv1.Postgresql
NewSpec *acidv1.Postgresql
WorkerID uint32
}
pkg/controller/util.go
+18 -10
View File
@@ -8,10 +8,10 @@ import (
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/util/wait"
acidv1 "github.com/zalando-incubator/postgres-operator/pkg/apis/acid.zalan.do/v1"
"github.com/zalando-incubator/postgres-operator/pkg/cluster"
"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/constants"
"github.com/zalando-incubator/postgres-operator/pkg/util/k8sutil"
"gopkg.in/yaml.v2"
)
@@ -47,22 +47,24 @@ func (c *Controller) clusterWorkerID(clusterName spec.NamespacedName) uint32 {
return c.clusterWorkers[clusterName]
}
func (c *Controller) createOperatorCRD(plural, singular, short string) error {
func (c *Controller) createOperatorCRD(name, kind, plural, short string) error {
subResourceStatus := apiextv1beta1.CustomResourceSubresourceStatus{}
crd := &apiextv1beta1.CustomResourceDefinition{
ObjectMeta: metav1.ObjectMeta{
Name: plural + "." + constants.CRDGroup,
Name: name,
},
Spec: apiextv1beta1.CustomResourceDefinitionSpec{
Group: constants.CRDGroup,
Version: constants.CRDApiVersion,
Group: acidv1.SchemeGroupVersion.Group,
Version: acidv1.SchemeGroupVersion.Version,
Names: apiextv1beta1.CustomResourceDefinitionNames{
Plural: plural,
Singular: singular,
ShortNames: []string{short},
Kind: singular,
ListKind: singular + "List",
Kind: kind,
},
Scope: apiextv1beta1.NamespaceScoped,
Subresources: &apiextv1beta1.CustomResourceSubresources{
Status: &subResourceStatus,
},
},
}
@@ -99,11 +101,17 @@ func (c *Controller) createOperatorCRD(plural, singular, short string) error {
}
func (c *Controller) createPostgresCRD() error {
return c.createOperatorCRD(constants.PostgresCRDResource, constants.PostgresCRDKind, constants.PostgresCRDShort)
return c.createOperatorCRD(acidv1.PostgresCRDResouceName,
acidv1.PostgresCRDResourceKind,
acidv1.PostgresCRDResourcePlural,
acidv1.PostgresCRDResourceShort)
}
func (c *Controller) createConfigurationCRD() error {
return c.createOperatorCRD(constants.OperatorConfigCRDResource, constants.OperatorConfigCRDKind, constants.OperatorConfigCRDShort)
return c.createOperatorCRD(acidv1.OperatorConfigCRDResourceName,
acidv1.OperatorConfigCRDResouceKind,
acidv1.OperatorConfigCRDResourcePlural,
acidv1.OperatorConfigCRDResourceShort)
}
func readDecodedRole(s string) (*spec.PgUser, error) {