mirror of
https://github.com/zalando/postgres-operator.git
synced 2026-10-08 15:32:53 +02:00
Merge branch 'master' into feature/diagnostic-rest-api
# Conflicts: # pkg/controller/controller.go # pkg/controller/pod.go
This commit is contained in:
+10
-25
@@ -3,27 +3,20 @@ package controller
|
||||
import (
|
||||
"sync"
|
||||
|
||||
"k8s.io/client-go/pkg/api"
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/apimachinery/pkg/runtime"
|
||||
"k8s.io/apimachinery/pkg/watch"
|
||||
"k8s.io/client-go/pkg/api/v1"
|
||||
"k8s.io/client-go/pkg/runtime"
|
||||
"k8s.io/client-go/pkg/watch"
|
||||
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/spec"
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/util"
|
||||
)
|
||||
|
||||
func (c *Controller) podListFunc(options api.ListOptions) (runtime.Object, error) {
|
||||
func (c *Controller) podListFunc(options metav1.ListOptions) (runtime.Object, error) {
|
||||
var labelSelector string
|
||||
var fieldSelector string
|
||||
|
||||
if options.LabelSelector != nil {
|
||||
labelSelector = options.LabelSelector.String()
|
||||
}
|
||||
|
||||
if options.FieldSelector != nil {
|
||||
fieldSelector = options.FieldSelector.String()
|
||||
}
|
||||
opts := v1.ListOptions{
|
||||
opts := metav1.ListOptions{
|
||||
LabelSelector: labelSelector,
|
||||
FieldSelector: fieldSelector,
|
||||
Watch: options.Watch,
|
||||
@@ -31,22 +24,14 @@ func (c *Controller) podListFunc(options api.ListOptions) (runtime.Object, error
|
||||
TimeoutSeconds: options.TimeoutSeconds,
|
||||
}
|
||||
|
||||
return c.KubeClient.CoreV1().Pods(c.opConfig.Namespace).List(opts)
|
||||
return c.KubeClient.Pods(c.opConfig.Namespace).List(opts)
|
||||
}
|
||||
|
||||
func (c *Controller) podWatchFunc(options api.ListOptions) (watch.Interface, error) {
|
||||
func (c *Controller) podWatchFunc(options metav1.ListOptions) (watch.Interface, error) {
|
||||
var labelSelector string
|
||||
var fieldSelector string
|
||||
|
||||
if options.LabelSelector != nil {
|
||||
labelSelector = options.LabelSelector.String()
|
||||
}
|
||||
|
||||
if options.FieldSelector != nil {
|
||||
fieldSelector = options.FieldSelector.String()
|
||||
}
|
||||
|
||||
opts := v1.ListOptions{
|
||||
opts := metav1.ListOptions{
|
||||
LabelSelector: labelSelector,
|
||||
FieldSelector: fieldSelector,
|
||||
Watch: options.Watch,
|
||||
@@ -54,7 +39,7 @@ func (c *Controller) podWatchFunc(options api.ListOptions) (watch.Interface, err
|
||||
TimeoutSeconds: options.TimeoutSeconds,
|
||||
}
|
||||
|
||||
return c.KubeClient.CoreV1Client.Pods(c.opConfig.Namespace).Watch(opts)
|
||||
return c.KubeClient.Pods(c.opConfig.Namespace).Watch(opts)
|
||||
}
|
||||
|
||||
func (c *Controller) podAdd(obj interface{}) {
|
||||
@@ -127,7 +112,7 @@ func (c *Controller) podEventsDispatcher(stopCh <-chan struct{}, wg *sync.WaitGr
|
||||
c.clustersMu.RUnlock()
|
||||
|
||||
if ok {
|
||||
c.logger.Debugf("Sending %s event of pod '%s' to the '%s' cluster channel", event.EventType, event.PodName, event.ClusterName)
|
||||
c.logger.Debugf("Sending %q event of pod %q to the %q cluster channel", event.EventType, event.PodName, event.ClusterName)
|
||||
cluster.ReceivePodEvent(event)
|
||||
}
|
||||
case <-stopCh:
|
||||
|
||||
@@ -1,18 +1,17 @@
|
||||
package controller
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"k8s.io/client-go/pkg/api"
|
||||
"k8s.io/client-go/pkg/api/meta"
|
||||
"k8s.io/client-go/pkg/fields"
|
||||
"k8s.io/client-go/pkg/runtime"
|
||||
"k8s.io/client-go/pkg/types"
|
||||
"k8s.io/client-go/pkg/watch"
|
||||
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"
|
||||
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/cluster"
|
||||
@@ -29,52 +28,43 @@ func (c *Controller) clusterResync(stopCh <-chan struct{}, wg *sync.WaitGroup) {
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
c.clusterListFunc(api.ListOptions{ResourceVersion: "0"})
|
||||
c.clusterListFunc(metav1.ListOptions{ResourceVersion: "0"})
|
||||
case <-stopCh:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Controller) clusterListFunc(options api.ListOptions) (runtime.Object, error) {
|
||||
c.logger.Info("Getting list of currently running clusters")
|
||||
func (c *Controller) clusterListFunc(options metav1.ListOptions) (runtime.Object, error) {
|
||||
var list spec.PostgresqlList
|
||||
var activeClustersCnt, failedClustersCnt int
|
||||
|
||||
req := c.RestClient.Get().
|
||||
RequestURI(fmt.Sprintf(constants.ListClustersURITemplate, c.opConfig.Namespace)).
|
||||
VersionedParams(&options, api.ParameterCodec).
|
||||
FieldsSelectorParam(fields.Everything())
|
||||
|
||||
object, err := req.Do().Get()
|
||||
req := c.RestClient.
|
||||
Get().
|
||||
Namespace(c.opConfig.Namespace).
|
||||
Resource(constants.ResourceName).
|
||||
VersionedParams(&options, metav1.ParameterCodec)
|
||||
|
||||
b, err := req.DoRaw()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("could not get list of postgresql objects: %v", err)
|
||||
}
|
||||
|
||||
objList, err := meta.ExtractList(object)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("could not extract list of postgresql objects: %v", err)
|
||||
return nil, err
|
||||
}
|
||||
err = json.Unmarshal(b, &list)
|
||||
|
||||
if time.Now().Unix()-atomic.LoadInt64(&c.lastClusterSyncTime) <= int64(c.opConfig.ResyncPeriod.Seconds()) {
|
||||
c.logger.Debugln("skipping resync of clusters")
|
||||
return object, err
|
||||
return &list, err
|
||||
}
|
||||
|
||||
var activeClustersCnt, failedClustersCnt int
|
||||
for _, obj := range objList {
|
||||
pg, ok := obj.(*spec.Postgresql)
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("could not cast object to postgresql")
|
||||
}
|
||||
|
||||
for _, pg := range list.Items {
|
||||
if pg.Error != nil {
|
||||
failedClustersCnt++
|
||||
continue
|
||||
}
|
||||
c.queueClusterEvent(nil, pg, spec.EventSync)
|
||||
c.queueClusterEvent(nil, &pg, spec.EventSync)
|
||||
activeClustersCnt++
|
||||
}
|
||||
if len(objList) > 0 {
|
||||
if len(list.Items) > 0 {
|
||||
if failedClustersCnt > 0 && activeClustersCnt == 0 {
|
||||
c.logger.Infof("There are no clusters running. %d are in the failed state", failedClustersCnt)
|
||||
} else if failedClustersCnt == 0 && activeClustersCnt > 0 {
|
||||
@@ -88,15 +78,48 @@ func (c *Controller) clusterListFunc(options api.ListOptions) (runtime.Object, e
|
||||
|
||||
atomic.StoreInt64(&c.lastClusterSyncTime, time.Now().Unix())
|
||||
|
||||
return object, err
|
||||
return &list, err
|
||||
}
|
||||
|
||||
func (c *Controller) clusterWatchFunc(options api.ListOptions) (watch.Interface, error) {
|
||||
req := c.RestClient.Get().
|
||||
RequestURI(fmt.Sprintf(constants.WatchClustersURITemplate, c.opConfig.Namespace)).
|
||||
VersionedParams(&options, api.ParameterCodec).
|
||||
FieldsSelectorParam(fields.Everything())
|
||||
return req.Watch()
|
||||
type tprDecoder struct {
|
||||
dec *json.Decoder
|
||||
close func() error
|
||||
}
|
||||
|
||||
func (d *tprDecoder) Close() {
|
||||
d.close()
|
||||
}
|
||||
|
||||
func (d *tprDecoder) 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
|
||||
r, err := c.RestClient.
|
||||
Get().
|
||||
Namespace(c.opConfig.Namespace).
|
||||
Resource(constants.ResourceName).
|
||||
VersionedParams(&options, metav1.ParameterCodec).
|
||||
FieldsSelectorParam(nil).
|
||||
Stream()
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return watch.NewStreamWatcher(&tprDecoder{
|
||||
dec: json.NewDecoder(r),
|
||||
close: r.Close,
|
||||
}), nil
|
||||
}
|
||||
|
||||
func (c *Controller) processEvent(obj interface{}) error {
|
||||
@@ -109,9 +132,9 @@ func (c *Controller) processEvent(obj interface{}) error {
|
||||
logger := c.logger.WithField("worker", event.WorkerID)
|
||||
|
||||
if event.EventType == spec.EventAdd || event.EventType == spec.EventSync {
|
||||
clusterName = util.NameFromMeta(event.NewSpec.Metadata)
|
||||
clusterName = util.NameFromMeta(event.NewSpec.ObjectMeta)
|
||||
} else {
|
||||
clusterName = util.NameFromMeta(event.OldSpec.Metadata)
|
||||
clusterName = util.NameFromMeta(event.OldSpec.ObjectMeta)
|
||||
}
|
||||
|
||||
c.clustersMu.RLock()
|
||||
@@ -121,14 +144,14 @@ func (c *Controller) processEvent(obj interface{}) error {
|
||||
switch event.EventType {
|
||||
case spec.EventAdd:
|
||||
if clusterFound {
|
||||
logger.Debugf("Cluster '%s' already exists", clusterName)
|
||||
logger.Debugf("Cluster %q already exists", clusterName)
|
||||
return nil
|
||||
}
|
||||
|
||||
logger.Infof("Creation of the '%s' cluster started", clusterName)
|
||||
logger.Infof("Creation of the %q cluster started", clusterName)
|
||||
|
||||
stopCh := make(chan struct{})
|
||||
cl = cluster.New(c.makeClusterConfig(), *event.NewSpec, logger)
|
||||
cl = cluster.New(c.makeClusterConfig(), c.KubeClient, *event.NewSpec, logger)
|
||||
cl.Run(stopCh)
|
||||
|
||||
c.clustersMu.Lock()
|
||||
@@ -143,31 +166,31 @@ func (c *Controller) processEvent(obj interface{}) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
logger.Infof("Cluster '%s' has been created", clusterName)
|
||||
logger.Infof("Cluster %q has been created", clusterName)
|
||||
case spec.EventUpdate:
|
||||
logger.Infof("Update of the '%s' cluster started", clusterName)
|
||||
logger.Infof("Update of the %q cluster started", clusterName)
|
||||
|
||||
if !clusterFound {
|
||||
logger.Warnf("Cluster '%s' does not exist", clusterName)
|
||||
logger.Warnf("Cluster %q does not exist", clusterName)
|
||||
return nil
|
||||
}
|
||||
if err := cl.Update(event.NewSpec); err != nil {
|
||||
cl.Error = fmt.Errorf("could not update cluster: %s", err)
|
||||
cl.Error = fmt.Errorf("could not update cluster: %v", err)
|
||||
logger.Errorf("%v", cl.Error)
|
||||
|
||||
return nil
|
||||
}
|
||||
cl.Error = nil
|
||||
logger.Infof("Cluster '%s' has been updated", clusterName)
|
||||
logger.Infof("Cluster %q has been updated", clusterName)
|
||||
case spec.EventDelete:
|
||||
logger.Infof("Deletion of the '%s' cluster started", clusterName)
|
||||
logger.Infof("Deletion of the %q cluster started", clusterName)
|
||||
if !clusterFound {
|
||||
logger.Errorf("Unknown cluster: %s", clusterName)
|
||||
logger.Errorf("Unknown cluster: %q", clusterName)
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := cl.Delete(); err != nil {
|
||||
logger.Errorf("could not delete cluster '%s': %s", clusterName, err)
|
||||
logger.Errorf("could not delete cluster %q: %v", clusterName, err)
|
||||
return nil
|
||||
}
|
||||
close(c.stopChs[clusterName])
|
||||
@@ -177,14 +200,14 @@ func (c *Controller) processEvent(obj interface{}) error {
|
||||
delete(c.stopChs, clusterName)
|
||||
c.clustersMu.Unlock()
|
||||
|
||||
logger.Infof("Cluster '%s' has been deleted", clusterName)
|
||||
logger.Infof("Cluster %q has been deleted", clusterName)
|
||||
case spec.EventSync:
|
||||
logger.Infof("Syncing of the '%s' cluster started", clusterName)
|
||||
logger.Infof("Syncing of the %q cluster started", clusterName)
|
||||
|
||||
// no race condition because a cluster is always processed by single worker
|
||||
if !clusterFound {
|
||||
stopCh := make(chan struct{})
|
||||
cl = cluster.New(c.makeClusterConfig(), *event.NewSpec, logger)
|
||||
cl = cluster.New(c.makeClusterConfig(), c.KubeClient, *event.NewSpec, logger)
|
||||
cl.Run(stopCh)
|
||||
|
||||
c.clustersMu.Lock()
|
||||
@@ -194,13 +217,13 @@ func (c *Controller) processEvent(obj interface{}) error {
|
||||
}
|
||||
|
||||
if err := cl.Sync(); err != nil {
|
||||
cl.Error = fmt.Errorf("could not sync cluster '%s': %v", clusterName, err)
|
||||
cl.Error = fmt.Errorf("could not sync cluster %q: %v", clusterName, err)
|
||||
logger.Errorf("%v", cl.Error)
|
||||
return nil
|
||||
}
|
||||
cl.Error = nil
|
||||
|
||||
logger.Infof("Cluster '%s' has been synced", clusterName)
|
||||
logger.Infof("Cluster %q has been synced", clusterName)
|
||||
}
|
||||
|
||||
return nil
|
||||
@@ -227,8 +250,8 @@ func (c *Controller) queueClusterEvent(old, new *spec.Postgresql, eventType spec
|
||||
)
|
||||
|
||||
if old != nil { //update, delete
|
||||
uid = old.Metadata.GetUID()
|
||||
clusterName = util.NameFromMeta(old.Metadata)
|
||||
uid = old.GetUID()
|
||||
clusterName = util.NameFromMeta(old.ObjectMeta)
|
||||
if eventType == spec.EventUpdate && new.Error == nil && old.Error != nil {
|
||||
eventType = spec.EventSync
|
||||
clusterError = new.Error
|
||||
@@ -236,13 +259,13 @@ func (c *Controller) queueClusterEvent(old, new *spec.Postgresql, eventType spec
|
||||
clusterError = old.Error
|
||||
}
|
||||
} else { //add, sync
|
||||
uid = new.Metadata.GetUID()
|
||||
clusterName = util.NameFromMeta(new.Metadata)
|
||||
uid = new.GetUID()
|
||||
clusterName = util.NameFromMeta(new.ObjectMeta)
|
||||
clusterError = new.Error
|
||||
}
|
||||
|
||||
if clusterError != nil && eventType != spec.EventDelete {
|
||||
c.logger.Debugf("Skipping %s event for invalid cluster %s (reason: %v)", eventType, clusterName, clusterError)
|
||||
c.logger.Debugf("Skipping %q event for invalid cluster %q (reason: %v)", eventType, clusterName, clusterError)
|
||||
return
|
||||
}
|
||||
|
||||
@@ -259,7 +282,7 @@ func (c *Controller) queueClusterEvent(old, new *spec.Postgresql, eventType spec
|
||||
if err := c.clusterEventQueues[workerID].Add(clusterEvent); err != nil {
|
||||
c.logger.WithField("worker", workerID).Errorf("error when queueing cluster event: %v", clusterEvent)
|
||||
}
|
||||
c.logger.WithField("worker", workerID).Infof("%s of the '%s' cluster has been queued", eventType, clusterName)
|
||||
c.logger.WithField("worker", workerID).Infof("%q of the %q cluster has been queued", eventType, clusterName)
|
||||
}
|
||||
|
||||
func (c *Controller) postgresqlAdd(obj interface{}) {
|
||||
@@ -282,7 +305,7 @@ func (c *Controller) postgresqlUpdate(prev, cur interface{}) {
|
||||
if !ok {
|
||||
c.logger.Errorf("could not cast to postgresql spec")
|
||||
}
|
||||
if pgOld.Metadata.ResourceVersion == pgNew.Metadata.ResourceVersion {
|
||||
if pgOld.ResourceVersion == pgNew.ResourceVersion {
|
||||
return
|
||||
}
|
||||
if reflect.DeepEqual(pgOld.Spec, pgNew.Spec) {
|
||||
|
||||
+14
-17
@@ -4,6 +4,7 @@ import (
|
||||
"fmt"
|
||||
"hash/crc32"
|
||||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
"k8s.io/client-go/pkg/api/v1"
|
||||
extv1beta "k8s.io/client-go/pkg/apis/extensions/v1beta1"
|
||||
|
||||
@@ -16,15 +17,12 @@ import (
|
||||
|
||||
func (c *Controller) makeClusterConfig() cluster.Config {
|
||||
infrastructureRoles := make(map[string]spec.PgUser)
|
||||
for k, v := range c.InfrastructureRoles {
|
||||
for k, v := range c.config.InfrastructureRoles {
|
||||
infrastructureRoles[k] = v
|
||||
}
|
||||
|
||||
return cluster.Config{
|
||||
KubeClient: c.KubeClient,
|
||||
RestClient: c.RestClient,
|
||||
RestConfig: c.RestConfig,
|
||||
TeamsAPIClient: c.TeamsAPIClient,
|
||||
RestConfig: c.config.RestConfig,
|
||||
OpConfig: config.Copy(c.opConfig),
|
||||
InfrastructureRoles: infrastructureRoles,
|
||||
}
|
||||
@@ -32,7 +30,7 @@ func (c *Controller) makeClusterConfig() cluster.Config {
|
||||
|
||||
func thirdPartyResource(TPRName string) *extv1beta.ThirdPartyResource {
|
||||
return &extv1beta.ThirdPartyResource{
|
||||
ObjectMeta: v1.ObjectMeta{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
//ThirdPartyResources are cluster-wide
|
||||
Name: TPRName,
|
||||
},
|
||||
@@ -48,33 +46,32 @@ func (c *Controller) clusterWorkerID(clusterName spec.NamespacedName) uint32 {
|
||||
}
|
||||
|
||||
func (c *Controller) createTPR() error {
|
||||
TPRName := fmt.Sprintf("%s.%s", constants.TPRName, constants.TPRVendor)
|
||||
tpr := thirdPartyResource(TPRName)
|
||||
tpr := thirdPartyResource(constants.TPRName)
|
||||
|
||||
_, err := c.KubeClient.ExtensionsV1beta1().ThirdPartyResources().Create(tpr)
|
||||
_, err := c.KubeClient.ThirdPartyResources().Create(tpr)
|
||||
if err != nil {
|
||||
if !k8sutil.ResourceAlreadyExists(err) {
|
||||
return err
|
||||
}
|
||||
c.logger.Infof("ThirdPartyResource '%s' is already registered", TPRName)
|
||||
c.logger.Infof("ThirdPartyResource %q is already registered", constants.TPRName)
|
||||
} else {
|
||||
c.logger.Infof("ThirdPartyResource '%s' has been registered", TPRName)
|
||||
c.logger.Infof("ThirdPartyResource %q' has been registered", constants.TPRName)
|
||||
}
|
||||
|
||||
return k8sutil.WaitTPRReady(c.RestClient, c.opConfig.TPR.ReadyWaitInterval, c.opConfig.TPR.ReadyWaitTimeout, c.opConfig.Namespace)
|
||||
}
|
||||
|
||||
func (c *Controller) getInfrastructureRoles() (result map[string]spec.PgUser, err error) {
|
||||
if c.opConfig.InfrastructureRolesSecretName == (spec.NamespacedName{}) {
|
||||
func (c *Controller) getInfrastructureRoles(rolesSecret *spec.NamespacedName) (result map[string]spec.PgUser, err error) {
|
||||
if *rolesSecret == (spec.NamespacedName{}) {
|
||||
// we don't have infrastructure roles defined, bail out
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
infraRolesSecret, err := c.KubeClient.
|
||||
Secrets(c.opConfig.InfrastructureRolesSecretName.Namespace).
|
||||
Get(c.opConfig.InfrastructureRolesSecretName.Name)
|
||||
Secrets(rolesSecret.Namespace).
|
||||
Get(rolesSecret.Name, metav1.GetOptions{})
|
||||
if err != nil {
|
||||
c.logger.Debugf("Infrastructure roles secret name: %s", c.opConfig.InfrastructureRolesSecretName)
|
||||
c.logger.Debugf("Infrastructure roles secret name: %q", *rolesSecret)
|
||||
return nil, fmt.Errorf("could not get infrastructure roles secret: %v", err)
|
||||
}
|
||||
|
||||
@@ -102,7 +99,7 @@ Users:
|
||||
case "inrole":
|
||||
t.MemberOf = append(t.MemberOf, s)
|
||||
default:
|
||||
c.logger.Warnf("Unknown key %s", p)
|
||||
c.logger.Warnf("Unknown key %q", p)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,154 @@
|
||||
package controller
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"reflect"
|
||||
"testing"
|
||||
|
||||
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
|
||||
v1core "k8s.io/client-go/kubernetes/typed/core/v1"
|
||||
"k8s.io/client-go/pkg/api/v1"
|
||||
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/spec"
|
||||
"github.com/zalando-incubator/postgres-operator/pkg/util/k8sutil"
|
||||
)
|
||||
|
||||
const (
|
||||
testInfrastructureRolesSecretName = "infrastructureroles-test"
|
||||
)
|
||||
|
||||
type mockSecret struct {
|
||||
v1core.SecretInterface
|
||||
}
|
||||
|
||||
func (c *mockSecret) Get(name string, options metav1.GetOptions) (*v1.Secret, error) {
|
||||
if name != testInfrastructureRolesSecretName {
|
||||
return nil, fmt.Errorf("NotFound")
|
||||
}
|
||||
secret := &v1.Secret{}
|
||||
secret.Name = mockController.opConfig.ClusterNameLabel
|
||||
secret.Data = map[string][]byte{
|
||||
"user1": []byte("testrole"),
|
||||
"password1": []byte("testpassword"),
|
||||
"inrole1": []byte("testinrole"),
|
||||
}
|
||||
return secret, nil
|
||||
|
||||
}
|
||||
|
||||
type MockSecretGetter struct {
|
||||
}
|
||||
|
||||
func (c *MockSecretGetter) Secrets(namespace string) v1core.SecretInterface {
|
||||
return &mockSecret{}
|
||||
}
|
||||
|
||||
func newMockKubernetesClient() k8sutil.KubernetesClient {
|
||||
return k8sutil.KubernetesClient{SecretsGetter: &MockSecretGetter{}}
|
||||
}
|
||||
|
||||
func newMockController() *Controller {
|
||||
controller := NewController(&Config{})
|
||||
controller.opConfig.ClusterNameLabel = "cluster-name"
|
||||
controller.opConfig.InfrastructureRolesSecretName =
|
||||
spec.NamespacedName{v1.NamespaceDefault, testInfrastructureRolesSecretName}
|
||||
controller.opConfig.Workers = 4
|
||||
controller.KubeClient = newMockKubernetesClient()
|
||||
return controller
|
||||
}
|
||||
|
||||
var mockController = newMockController()
|
||||
|
||||
func TestPodClusterName(t *testing.T) {
|
||||
var testTable = []struct {
|
||||
in *v1.Pod
|
||||
expected spec.NamespacedName
|
||||
}{
|
||||
{
|
||||
&v1.Pod{},
|
||||
spec.NamespacedName{},
|
||||
},
|
||||
{
|
||||
&v1.Pod{
|
||||
ObjectMeta: metav1.ObjectMeta{
|
||||
Namespace: v1.NamespaceDefault,
|
||||
Labels: map[string]string{
|
||||
mockController.opConfig.ClusterNameLabel: "testcluster",
|
||||
},
|
||||
},
|
||||
},
|
||||
spec.NamespacedName{v1.NamespaceDefault, "testcluster"},
|
||||
},
|
||||
}
|
||||
for _, test := range testTable {
|
||||
resp := mockController.podClusterName(test.in)
|
||||
if resp != test.expected {
|
||||
t.Errorf("expected response %v does not match the actual %v", test.expected, resp)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestClusterWorkerID(t *testing.T) {
|
||||
var testTable = []struct {
|
||||
in spec.NamespacedName
|
||||
expected uint32
|
||||
}{
|
||||
{
|
||||
in: spec.NamespacedName{"foo", "bar"},
|
||||
expected: 2,
|
||||
},
|
||||
{
|
||||
in: spec.NamespacedName{"default", "testcluster"},
|
||||
expected: 3,
|
||||
},
|
||||
}
|
||||
for _, test := range testTable {
|
||||
resp := mockController.clusterWorkerID(test.in)
|
||||
if resp != test.expected {
|
||||
t.Errorf("expected response %v does not match the actual %v", test.expected, resp)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGetInfrastructureRoles(t *testing.T) {
|
||||
var testTable = []struct {
|
||||
secretName spec.NamespacedName
|
||||
expectedRoles map[string]spec.PgUser
|
||||
expectedError error
|
||||
}{
|
||||
{
|
||||
spec.NamespacedName{},
|
||||
nil,
|
||||
nil,
|
||||
},
|
||||
{
|
||||
spec.NamespacedName{v1.NamespaceDefault, "null"},
|
||||
nil,
|
||||
fmt.Errorf(`could not get infrastructure roles secret: NotFound`),
|
||||
},
|
||||
{
|
||||
spec.NamespacedName{v1.NamespaceDefault, testInfrastructureRolesSecretName},
|
||||
map[string]spec.PgUser{
|
||||
"testrole": {
|
||||
"testrole",
|
||||
"testpassword",
|
||||
nil,
|
||||
[]string{"testinrole"},
|
||||
},
|
||||
},
|
||||
nil,
|
||||
},
|
||||
}
|
||||
for _, test := range testTable {
|
||||
roles, err := mockController.getInfrastructureRoles(&test.secretName)
|
||||
if err != test.expectedError {
|
||||
if err != nil && test.expectedError != nil && err.Error() == test.expectedError.Error() {
|
||||
continue
|
||||
}
|
||||
t.Errorf("expected error '%v' does not match the actual error '%v'", test.expectedError, err)
|
||||
}
|
||||
if !reflect.DeepEqual(roles, test.expectedRoles) {
|
||||
t.Errorf("expected roles output %v does not match the actual %v", test.expectedRoles, roles)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user