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

Merge branch 'master' into default-db-roles

Browse Source
This commit is contained in:
Felix Kunde
2019-10-04 18:14:23 +02:00
parent 18da6e552d baae1887b3
commit 7d9cfe9f54
31 changed files with 633 additions and 451 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/apis/acid.zalan.do/v1/zz_generated.deepcopy.go
+2 -2
View File
@@ -249,7 +249,7 @@ func (in *OperatorConfigurationData) DeepCopy() *OperatorConfigurationData {
func (in *OperatorConfigurationList) DeepCopyInto(out *OperatorConfigurationList) {
*out = *in
out.TypeMeta = in.TypeMeta
out.ListMeta = in.ListMeta
in.ListMeta.DeepCopyInto(&out.ListMeta)
if in.Items != nil {
in, out := &in.Items, &out.Items
*out = make([]OperatorConfiguration, len(*in))
@@ -604,7 +604,7 @@ func (in *Postgresql) DeepCopyObject() runtime.Object {
func (in *PostgresqlList) DeepCopyInto(out *PostgresqlList) {
*out = *in
out.TypeMeta = in.TypeMeta
out.ListMeta = in.ListMeta
in.ListMeta.DeepCopyInto(&out.ListMeta)
if in.Items != nil {
in, out := &in.Items, &out.Items
*out = make([]Postgresql, len(*in))
pkg/cluster/cluster.go
+4 -4
View File
@@ -13,7 +13,7 @@ import (
"time"
"github.com/sirupsen/logrus"
"k8s.io/api/apps/v1beta1"
appsv1 "k8s.io/api/apps/v1"
v1 "k8s.io/api/core/v1"
policybeta1 "k8s.io/api/policy/v1beta1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -53,7 +53,7 @@ type kubeResources struct {
Services map[PostgresRole]*v1.Service
Endpoints map[PostgresRole]*v1.Endpoints
Secrets map[types.UID]*v1.Secret
Statefulset *v1beta1.StatefulSet
Statefulset *appsv1.StatefulSet
PodDisruptionBudget *policybeta1.PodDisruptionBudget
//Pods are treated separately
//PVCs are treated separately
@@ -219,7 +219,7 @@ func (c *Cluster) Create() error {
service *v1.Service
ep *v1.Endpoints
ss *v1beta1.StatefulSet
ss *appsv1.StatefulSet
)
defer func() {
@@ -323,7 +323,7 @@ func (c *Cluster) Create() error {
return nil
}
func (c *Cluster) compareStatefulSetWith(statefulSet *v1beta1.StatefulSet) *compareStatefulsetResult {
func (c *Cluster) compareStatefulSetWith(statefulSet *appsv1.StatefulSet) *compareStatefulsetResult {
reasons := make([]string, 0)
var match, needsRollUpdate, needsReplace bool
pkg/cluster/k8sres.go
+8 -8
View File
@@ -7,7 +7,7 @@ import (
"github.com/sirupsen/logrus"
"k8s.io/api/apps/v1beta1"
appsv1 "k8s.io/api/apps/v1"
v1 "k8s.io/api/core/v1"
policybeta1 "k8s.io/api/policy/v1beta1"
"k8s.io/apimachinery/pkg/api/resource"
@@ -711,7 +711,7 @@ func makeResources(cpuRequest, memoryRequest, cpuLimit, memoryLimit string) acid
}
}
func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*v1beta1.StatefulSet, error) {
func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*appsv1.StatefulSet, error) {
var (
err error
@@ -917,25 +917,25 @@ func (c *Cluster) generateStatefulSet(spec *acidv1.PostgresSpec) (*v1beta1.State
// the operator has domain-specific logic on how to do rolling updates of PG clusters
// so we do not use default rolling updates implemented by stateful sets
// that leaves the legacy "OnDelete" update strategy as the only option
updateStrategy := v1beta1.StatefulSetUpdateStrategy{Type: v1beta1.OnDeleteStatefulSetStrategyType}
updateStrategy := appsv1.StatefulSetUpdateStrategy{Type: appsv1.OnDeleteStatefulSetStrategyType}
var podManagementPolicy v1beta1.PodManagementPolicyType
var podManagementPolicy appsv1.PodManagementPolicyType
if c.OpConfig.PodManagementPolicy == "ordered_ready" {
podManagementPolicy = v1beta1.OrderedReadyPodManagement
podManagementPolicy = appsv1.OrderedReadyPodManagement
} else if c.OpConfig.PodManagementPolicy == "parallel" {
podManagementPolicy = v1beta1.ParallelPodManagement
podManagementPolicy = appsv1.ParallelPodManagement
} else {
return nil, fmt.Errorf("could not set the pod management policy to the unknown value: %v", c.OpConfig.PodManagementPolicy)
}
statefulSet := &v1beta1.StatefulSet{
statefulSet := &appsv1.StatefulSet{
ObjectMeta: metav1.ObjectMeta{
Name: c.statefulSetName(),
Namespace: c.Namespace,
Labels: c.labelsSet(true),
Annotations: map[string]string{rollingUpdateStatefulsetAnnotationKey: "false"},
},
Spec: v1beta1.StatefulSetSpec{
Spec: appsv1.StatefulSetSpec{
Replicas: &numberOfInstances,
Selector: c.labelsSelector(),
ServiceName: c.serviceName(Master),
pkg/cluster/pod.go
+2 -2
View File
@@ -4,12 +4,12 @@ import (
"fmt"
"math/rand"
appsv1 "k8s.io/api/apps/v1"
v1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"github.com/zalando/postgres-operator/pkg/spec"
"github.com/zalando/postgres-operator/pkg/util"
"k8s.io/api/apps/v1beta1"
)
func (c *Cluster) listPods() ([]v1.Pod, error) {
@@ -205,7 +205,7 @@ 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 {
var sset *v1beta1.StatefulSet
var sset *appsv1.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)
pkg/cluster/resources.go
+10 -10
View File
@@ -5,7 +5,7 @@ import (
"strconv"
"strings"
"k8s.io/api/apps/v1beta1"
appsv1 "k8s.io/api/apps/v1"
batchv1beta1 "k8s.io/api/batch/v1beta1"
v1 "k8s.io/api/core/v1"
policybeta1 "k8s.io/api/policy/v1beta1"
@@ -64,7 +64,7 @@ func (c *Cluster) listResources() error {
return nil
}
func (c *Cluster) createStatefulSet() (*v1beta1.StatefulSet, error) {
func (c *Cluster) createStatefulSet() (*appsv1.StatefulSet, error) {
c.setProcessName("creating statefulset")
statefulSetSpec, err := c.generateStatefulSet(&c.Spec)
if err != nil {
@@ -95,7 +95,7 @@ func getPodIndex(podName string) (int32, error) {
return int32(res), nil
}
func (c *Cluster) preScaleDown(newStatefulSet *v1beta1.StatefulSet) error {
func (c *Cluster) preScaleDown(newStatefulSet *appsv1.StatefulSet) error {
masterPod, err := c.getRolePods(Master)
if err != nil {
return fmt.Errorf("could not get master pod: %v", err)
@@ -135,7 +135,7 @@ func (c *Cluster) preScaleDown(newStatefulSet *v1beta1.StatefulSet) error {
// setRollingUpdateFlagForStatefulSet sets the indicator or the rolling update requirement
// in the StatefulSet annotation.
func (c *Cluster) setRollingUpdateFlagForStatefulSet(sset *v1beta1.StatefulSet, val bool) {
func (c *Cluster) setRollingUpdateFlagForStatefulSet(sset *appsv1.StatefulSet, val bool) {
anno := sset.GetAnnotations()
if anno == nil {
anno = make(map[string]string)
@@ -160,7 +160,7 @@ func (c *Cluster) applyRollingUpdateFlagforStatefulSet(val bool) error {
// getRollingUpdateFlagFromStatefulSet returns the value of the rollingUpdate flag from the passed
// StatefulSet, reverting to the default value in case of errors
func (c *Cluster) getRollingUpdateFlagFromStatefulSet(sset *v1beta1.StatefulSet, defaultValue bool) (flag bool) {
func (c *Cluster) getRollingUpdateFlagFromStatefulSet(sset *appsv1.StatefulSet, defaultValue bool) (flag bool) {
anno := sset.GetAnnotations()
flag = defaultValue
@@ -181,7 +181,7 @@ func (c *Cluster) getRollingUpdateFlagFromStatefulSet(sset *v1beta1.StatefulSet,
// mergeRollingUpdateFlagUsingCache returns the value of the rollingUpdate flag from the passed
// statefulset, however, the value can be cleared if there is a cached flag in the cluster that
// is set to false (the discrepancy could be a result of a failed StatefulSet update)
func (c *Cluster) mergeRollingUpdateFlagUsingCache(runningStatefulSet *v1beta1.StatefulSet) bool {
func (c *Cluster) mergeRollingUpdateFlagUsingCache(runningStatefulSet *appsv1.StatefulSet) bool {
var (
cachedStatefulsetExists, clearRollingUpdateFromCache, podsRollingUpdateRequired bool
)
@@ -207,7 +207,7 @@ func (c *Cluster) mergeRollingUpdateFlagUsingCache(runningStatefulSet *v1beta1.S
return podsRollingUpdateRequired
}
func (c *Cluster) updateStatefulSetAnnotations(annotations map[string]string) (*v1beta1.StatefulSet, error) {
func (c *Cluster) updateStatefulSetAnnotations(annotations map[string]string) (*appsv1.StatefulSet, error) {
c.logger.Debugf("updating statefulset annotations")
patchData, err := metaAnnotationsPatch(annotations)
if err != nil {
@@ -223,7 +223,7 @@ func (c *Cluster) updateStatefulSetAnnotations(annotations map[string]string) (*
return result, nil
}
func (c *Cluster) updateStatefulSet(newStatefulSet *v1beta1.StatefulSet) error {
func (c *Cluster) updateStatefulSet(newStatefulSet *appsv1.StatefulSet) error {
c.setProcessName("updating statefulset")
if c.Statefulset == nil {
return fmt.Errorf("there is no statefulset in the cluster")
@@ -264,7 +264,7 @@ func (c *Cluster) updateStatefulSet(newStatefulSet *v1beta1.StatefulSet) error {
}
// replaceStatefulSet deletes an old StatefulSet and creates the new using spec in the PostgreSQL CRD.
func (c *Cluster) replaceStatefulSet(newStatefulSet *v1beta1.StatefulSet) error {
func (c *Cluster) replaceStatefulSet(newStatefulSet *appsv1.StatefulSet) error {
c.setProcessName("replacing statefulset")
if c.Statefulset == nil {
return fmt.Errorf("there is no statefulset in the cluster")
@@ -676,7 +676,7 @@ func (c *Cluster) GetEndpointReplica() *v1.Endpoints {
}
// GetStatefulSet returns cluster's kubernetes StatefulSet
func (c *Cluster) GetStatefulSet() *v1beta1.StatefulSet {
func (c *Cluster) GetStatefulSet() *appsv1.StatefulSet {
return c.Statefulset
}
pkg/cluster/types.go
+2 -2
View File
@@ -4,7 +4,7 @@ import (
"time"
acidv1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
"k8s.io/api/apps/v1beta1"
appsv1 "k8s.io/api/apps/v1"
"k8s.io/api/core/v1"
policybeta1 "k8s.io/api/policy/v1beta1"
"k8s.io/apimachinery/pkg/types"
@@ -60,7 +60,7 @@ type ClusterStatus struct {
ReplicaService *v1.Service
MasterEndpoint *v1.Endpoints
ReplicaEndpoint *v1.Endpoints
StatefulSet *v1beta1.StatefulSet
StatefulSet *appsv1.StatefulSet
PodDisruptionBudget *policybeta1.PodDisruptionBudget
CurrentProcess Process
pkg/cluster/util.go
+2 -2
View File
@@ -11,7 +11,7 @@ import (
"strings"
"time"
"k8s.io/api/apps/v1beta1"
appsv1 "k8s.io/api/apps/v1"
v1 "k8s.io/api/core/v1"
policybeta1 "k8s.io/api/policy/v1beta1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -168,7 +168,7 @@ func (c *Cluster) logPDBChanges(old, new *policybeta1.PodDisruptionBudget, isUpd
c.logger.Debugf("diff\n%s\n", util.PrettyDiff(old.Spec, new.Spec))
}
func (c *Cluster) logStatefulSetChanges(old, new *v1beta1.StatefulSet, isUpdate bool, reasons []string) {
func (c *Cluster) logStatefulSetChanges(old, new *appsv1.StatefulSet, isUpdate bool, reasons []string) {
if isUpdate {
c.logger.Infof("statefulset %q has been changed", util.NameFromMeta(old.ObjectMeta))
} else {
pkg/controller/postgresql.go
+1 -1
View File
@@ -312,7 +312,7 @@ func (c *Controller) processClusterEventsQueue(idx int, stopCh <-chan struct{},
for {
obj, err := c.clusterEventQueues[idx].Pop(cache.PopProcessFunc(func(interface{}) error { return nil }))
if err != nil {
if err == cache.FIFOClosedError {
if err == cache.ErrFIFOClosed {
return
}
c.logger.Errorf("error when processing cluster events queue: %v", err)
pkg/generated/clientset/versioned/clientset.go
+7 -8
View File
@@ -25,6 +25,8 @@ SOFTWARE.
package versioned
import (
"fmt"
acidv1 "github.com/zalando/postgres-operator/pkg/generated/clientset/versioned/typed/acid.zalan.do/v1"
discovery "k8s.io/client-go/discovery"
rest "k8s.io/client-go/rest"
@@ -34,8 +36,6 @@ import (
type Interface interface {
Discovery() discovery.DiscoveryInterface
AcidV1() acidv1.AcidV1Interface
// Deprecated: please explicitly pick a version if possible.
Acid() acidv1.AcidV1Interface
}
// Clientset contains the clients for groups. Each group has exactly one
@@ -50,12 +50,6 @@ func (c *Clientset) AcidV1() acidv1.AcidV1Interface {
return c.acidV1
}
// Deprecated: Acid retrieves the default version of AcidClient.
// Please explicitly pick a version.
func (c *Clientset) Acid() acidv1.AcidV1Interface {
return c.acidV1
}
// Discovery retrieves the DiscoveryClient
func (c *Clientset) Discovery() discovery.DiscoveryInterface {
if c == nil {
@@ -65,9 +59,14 @@ func (c *Clientset) Discovery() discovery.DiscoveryInterface {
}
// NewForConfig creates a new Clientset for the given config.
// If config's RateLimiter is not set and QPS and Burst are acceptable,
// NewForConfig will generate a rate-limiter in configShallowCopy.
func NewForConfig(c *rest.Config) (*Clientset, error) {
configShallowCopy := *c
if configShallowCopy.RateLimiter == nil && configShallowCopy.QPS > 0 {
if configShallowCopy.Burst <= 0 {
return nil, fmt.Errorf("Burst is required to be greater than 0 when RateLimiter is not set and QPS is set to greater than 0")
}
configShallowCopy.RateLimiter = flowcontrol.NewTokenBucketRateLimiter(configShallowCopy.QPS, configShallowCopy.Burst)
}
var cs Clientset
pkg/generated/clientset/versioned/fake/clientset_generated.go
+6 -6
View File
@@ -47,7 +47,7 @@ func NewSimpleClientset(objects ...runtime.Object) *Clientset {
}
}
cs := &Clientset{}
cs := &Clientset{tracker: o}
cs.discovery = &fakediscovery.FakeDiscovery{Fake: &cs.Fake}
cs.AddReactor("*", "*", testing.ObjectReaction(o))
cs.AddWatchReactor("*", func(action testing.Action) (handled bool, ret watch.Interface, err error) {
@@ -69,20 +69,20 @@ func NewSimpleClientset(objects ...runtime.Object) *Clientset {
type Clientset struct {
testing.Fake
discovery *fakediscovery.FakeDiscovery
tracker testing.ObjectTracker
}
func (c *Clientset) Discovery() discovery.DiscoveryInterface {
return c.discovery
}
func (c *Clientset) Tracker() testing.ObjectTracker {
return c.tracker
}
var _ clientset.Interface = &Clientset{}
// AcidV1 retrieves the AcidV1Client
func (c *Clientset) AcidV1() acidv1.AcidV1Interface {
return &fakeacidv1.FakeAcidV1{Fake: &c.Fake}
}
// Acid retrieves the AcidV1Client
func (c *Clientset) Acid() acidv1.AcidV1Interface {
return &fakeacidv1.FakeAcidV1{Fake: &c.Fake}
}
pkg/generated/clientset/versioned/fake/register.go
+9 -7
View File
@@ -30,15 +30,14 @@ import (
runtime "k8s.io/apimachinery/pkg/runtime"
schema "k8s.io/apimachinery/pkg/runtime/schema"
serializer "k8s.io/apimachinery/pkg/runtime/serializer"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
)
var scheme = runtime.NewScheme()
var codecs = serializer.NewCodecFactory(scheme)
var parameterCodec = runtime.NewParameterCodec(scheme)
func init() {
v1.AddToGroupVersion(scheme, schema.GroupVersion{Version: "v1"})
AddToScheme(scheme)
var localSchemeBuilder = runtime.SchemeBuilder{
acidv1.AddToScheme,
}
// AddToScheme adds all types of this clientset into the given scheme. This allows composition
@@ -51,10 +50,13 @@ func init() {
// )
//
// kclientset, _ := kubernetes.NewForConfig(c)
// aggregatorclientsetscheme.AddToScheme(clientsetscheme.Scheme)
// _ = aggregatorclientsetscheme.AddToScheme(clientsetscheme.Scheme)
//
// After this, RawExtensions in Kubernetes types will serialize kube-aggregator types
// correctly.
func AddToScheme(scheme *runtime.Scheme) {
acidv1.AddToScheme(scheme)
var AddToScheme = localSchemeBuilder.AddToScheme
func init() {
v1.AddToGroupVersion(scheme, schema.GroupVersion{Version: "v1"})
utilruntime.Must(AddToScheme(scheme))
}
pkg/generated/clientset/versioned/scheme/register.go
+9 -7
View File
@@ -30,15 +30,14 @@ import (
runtime "k8s.io/apimachinery/pkg/runtime"
schema "k8s.io/apimachinery/pkg/runtime/schema"
serializer "k8s.io/apimachinery/pkg/runtime/serializer"
utilruntime "k8s.io/apimachinery/pkg/util/runtime"
)
var Scheme = runtime.NewScheme()
var Codecs = serializer.NewCodecFactory(Scheme)
var ParameterCodec = runtime.NewParameterCodec(Scheme)
func init() {
v1.AddToGroupVersion(Scheme, schema.GroupVersion{Version: "v1"})
AddToScheme(Scheme)
var localSchemeBuilder = runtime.SchemeBuilder{
acidv1.AddToScheme,
}
// AddToScheme adds all types of this clientset into the given scheme. This allows composition
@@ -51,10 +50,13 @@ func init() {
// )
//
// kclientset, _ := kubernetes.NewForConfig(c)
// aggregatorclientsetscheme.AddToScheme(clientsetscheme.Scheme)
// _ = aggregatorclientsetscheme.AddToScheme(clientsetscheme.Scheme)
//
// After this, RawExtensions in Kubernetes types will serialize kube-aggregator types
// correctly.
func AddToScheme(scheme *runtime.Scheme) {
acidv1.AddToScheme(scheme)
var AddToScheme = localSchemeBuilder.AddToScheme
func init() {
v1.AddToGroupVersion(Scheme, schema.GroupVersion{Version: "v1"})
utilruntime.Must(AddToScheme(Scheme))
}
pkg/generated/clientset/versioned/typed/acid.zalan.do/v1/acid.zalan.do_client.go
+1 -2
View File
@@ -27,7 +27,6 @@ package v1
import (
v1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
"github.com/zalando/postgres-operator/pkg/generated/clientset/versioned/scheme"
serializer "k8s.io/apimachinery/pkg/runtime/serializer"
rest "k8s.io/client-go/rest"
)
@@ -82,7 +81,7 @@ func setConfigDefaults(config *rest.Config) error {
gv := v1.SchemeGroupVersion
config.GroupVersion = &gv
config.APIPath = "/apis"
config.NegotiatedSerializer = serializer.DirectCodecFactory{CodecFactory: scheme.Codecs}
config.NegotiatedSerializer = scheme.Codecs.WithoutConversion()
if config.UserAgent == "" {
config.UserAgent = rest.DefaultKubernetesUserAgent()
pkg/generated/clientset/versioned/typed/acid.zalan.do/v1/fake/fake_postgresql.go
+1 -1
View File
@@ -137,7 +137,7 @@ func (c *FakePostgresqls) DeleteCollection(options *v1.DeleteOptions, listOption
// Patch applies the patch and returns the patched postgresql.
func (c *FakePostgresqls) Patch(name string, pt types.PatchType, data []byte, subresources ...string) (result *acidzalandov1.Postgresql, err error) {
obj, err := c.Fake.
Invokes(testing.NewPatchSubresourceAction(postgresqlsResource, c.ns, name, data, subresources...), &acidzalandov1.Postgresql{})
Invokes(testing.NewPatchSubresourceAction(postgresqlsResource, c.ns, name, pt, data, subresources...), &acidzalandov1.Postgresql{})
if obj == nil {
return nil, err
pkg/generated/clientset/versioned/typed/acid.zalan.do/v1/postgresql.go
+17
View File
@@ -25,6 +25,8 @@ SOFTWARE.
package v1
import (
"time"
v1 "github.com/zalando/postgres-operator/pkg/apis/acid.zalan.do/v1"
scheme "github.com/zalando/postgres-operator/pkg/generated/clientset/versioned/scheme"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -82,11 +84,16 @@ func (c *postgresqls) Get(name string, options metav1.GetOptions) (result *v1.Po
// List takes label and field selectors, and returns the list of Postgresqls that match those selectors.
func (c *postgresqls) List(opts metav1.ListOptions) (result *v1.PostgresqlList, err error) {
var timeout time.Duration
if opts.TimeoutSeconds != nil {
timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
}
result = &v1.PostgresqlList{}
err = c.client.Get().
Namespace(c.ns).
Resource("postgresqls").
VersionedParams(&opts, scheme.ParameterCodec).
Timeout(timeout).
Do().
Into(result)
return
@@ -94,11 +101,16 @@ func (c *postgresqls) List(opts metav1.ListOptions) (result *v1.PostgresqlList,
// Watch returns a watch.Interface that watches the requested postgresqls.
func (c *postgresqls) Watch(opts metav1.ListOptions) (watch.Interface, error) {
var timeout time.Duration
if opts.TimeoutSeconds != nil {
timeout = time.Duration(*opts.TimeoutSeconds) * time.Second
}
opts.Watch = true
return c.client.Get().
Namespace(c.ns).
Resource("postgresqls").
VersionedParams(&opts, scheme.ParameterCodec).
Timeout(timeout).
Watch()
}
@@ -156,10 +168,15 @@ func (c *postgresqls) Delete(name string, options *metav1.DeleteOptions) error {
// DeleteCollection deletes a collection of objects.
func (c *postgresqls) DeleteCollection(options *metav1.DeleteOptions, listOptions metav1.ListOptions) error {
var timeout time.Duration
if listOptions.TimeoutSeconds != nil {
timeout = time.Duration(*listOptions.TimeoutSeconds) * time.Second
}
return c.client.Delete().
Namespace(c.ns).
Resource("postgresqls").
VersionedParams(&listOptions, scheme.ParameterCodec).
Timeout(timeout).
Body(options).
Do().
Error()
pkg/generated/informers/externalversions/internalinterfaces/factory_interfaces.go
+2
View File
@@ -33,6 +33,7 @@ import (
cache "k8s.io/client-go/tools/cache"
)
// NewInformerFunc takes versioned.Interface and time.Duration to return a SharedIndexInformer.
type NewInformerFunc func(versioned.Interface, time.Duration) cache.SharedIndexInformer
// SharedInformerFactory a small interface to allow for adding an informer without an import cycle
@@ -41,4 +42,5 @@ type SharedInformerFactory interface {
InformerFor(obj runtime.Object, newFunc NewInformerFunc) cache.SharedIndexInformer
}
// TweakListOptionsFunc is a function that transforms a v1.ListOptions.
type TweakListOptionsFunc func(*v1.ListOptions)
pkg/util/k8sutil/k8sutil.go
+18 -18
View File
@@ -16,8 +16,8 @@ import (
apiextbeta1 "k8s.io/apiextensions-apiserver/pkg/client/clientset/clientset/typed/apiextensions/v1beta1"
apierrors "k8s.io/apimachinery/pkg/api/errors"
"k8s.io/client-go/kubernetes"
"k8s.io/client-go/kubernetes/typed/apps/v1beta1"
v1core "k8s.io/client-go/kubernetes/typed/core/v1"
appsv1 "k8s.io/client-go/kubernetes/typed/apps/v1"
corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
policyv1beta1 "k8s.io/client-go/kubernetes/typed/policy/v1beta1"
rbacv1beta1 "k8s.io/client-go/kubernetes/typed/rbac/v1beta1"
"k8s.io/client-go/rest"
@@ -29,17 +29,17 @@ import (
// KubernetesClient describes getters for Kubernetes objects
type KubernetesClient struct {
v1core.SecretsGetter
v1core.ServicesGetter
v1core.EndpointsGetter
v1core.PodsGetter
v1core.PersistentVolumesGetter
v1core.PersistentVolumeClaimsGetter
v1core.ConfigMapsGetter
v1core.NodesGetter
v1core.NamespacesGetter
v1core.ServiceAccountsGetter
v1beta1.StatefulSetsGetter
corev1.SecretsGetter
corev1.ServicesGetter
corev1.EndpointsGetter
corev1.PodsGetter
corev1.PersistentVolumesGetter
corev1.PersistentVolumeClaimsGetter
corev1.ConfigMapsGetter
corev1.NodesGetter
corev1.NamespacesGetter
corev1.ServiceAccountsGetter
appsv1.StatefulSetsGetter
rbacv1beta1.RoleBindingsGetter
policyv1beta1.PodDisruptionBudgetsGetter
apiextbeta1.CustomResourceDefinitionsGetter
@@ -50,14 +50,14 @@ type KubernetesClient struct {
}
type mockSecret struct {
v1core.SecretInterface
corev1.SecretInterface
}
type MockSecretGetter struct {
}
type mockConfigMap struct {
v1core.ConfigMapInterface
corev1.ConfigMapInterface
}
type MockConfigMapsGetter struct {
@@ -101,7 +101,7 @@ func NewFromConfig(cfg *rest.Config) (KubernetesClient, error) {
kubeClient.PersistentVolumesGetter = client.CoreV1()
kubeClient.NodesGetter = client.CoreV1()
kubeClient.NamespacesGetter = client.CoreV1()
kubeClient.StatefulSetsGetter = client.AppsV1beta1()
kubeClient.StatefulSetsGetter = client.AppsV1()
kubeClient.PodDisruptionBudgetsGetter = client.PolicyV1beta1()
kubeClient.RESTClient = client.CoreV1().RESTClient()
kubeClient.RoleBindingsGetter = client.RbacV1beta1()
@@ -215,12 +215,12 @@ func (c *mockConfigMap) Get(name string, options metav1.GetOptions) (*v1.ConfigM
}
// Secrets to be mocked
func (c *MockSecretGetter) Secrets(namespace string) v1core.SecretInterface {
func (c *MockSecretGetter) Secrets(namespace string) corev1.SecretInterface {
return &mockSecret{}
}
// ConfigMaps to be mocked
func (c *MockConfigMapsGetter) ConfigMaps(namespace string) v1core.ConfigMapInterface {
func (c *MockConfigMapsGetter) ConfigMaps(namespace string) corev1.ConfigMapInterface {
return &mockConfigMap{}
}
pkg/util/patroni/patroni.go
+25 -4
View File
@@ -5,7 +5,9 @@ import (
"encoding/json"
"fmt"
"io/ioutil"
"net"
"net/http"
"strconv"
"time"
"github.com/sirupsen/logrus"
@@ -43,8 +45,19 @@ func New(logger *logrus.Entry) *Patroni {
}
}
func apiURL(masterPod *v1.Pod) string {
return fmt.Sprintf("http://%s:%d", masterPod.Status.PodIP, apiPort)
func apiURL(masterPod *v1.Pod) (string, error) {
ip := net.ParseIP(masterPod.Status.PodIP)
if ip == nil {
return "", fmt.Errorf("%s is not a valid IP", masterPod.Status.PodIP)
}
// Sanity check PodIP
if ip.To4() == nil {
if ip.To16() == nil {
// Shouldn't ever get here, but library states it's possible.
return "", fmt.Errorf("%s is not a valid IPv4/IPv6 address", masterPod.Status.PodIP)
}
}
return fmt.Sprintf("http://%s", net.JoinHostPort(ip.String(), strconv.Itoa(apiPort))), nil
}
func (p *Patroni) httpPostOrPatch(method string, url string, body *bytes.Buffer) (err error) {
@@ -88,7 +101,11 @@ func (p *Patroni) Switchover(master *v1.Pod, candidate string) error {
if err != nil {
return fmt.Errorf("could not encode json: %v", err)
}
return p.httpPostOrPatch(http.MethodPost, apiURL(master)+failoverPath, buf)
apiURLString, err := apiURL(master)
if err != nil {
return err
}
return p.httpPostOrPatch(http.MethodPost, apiURLString+failoverPath, buf)
}
//TODO: add an option call /patroni to check if it is necessary to restart the server
@@ -100,5 +117,9 @@ func (p *Patroni) SetPostgresParameters(server *v1.Pod, parameters map[string]st
if err != nil {
return fmt.Errorf("could not encode json: %v", err)
}
return p.httpPostOrPatch(http.MethodPatch, apiURL(server)+configPath, buf)
apiURLString, err := apiURL(server)
if err != nil {
return err
}
return p.httpPostOrPatch(http.MethodPatch, apiURLString+configPath, buf)
}
pkg/util/patroni/patroni_test.go
+74
View File
@@ -0,0 +1,74 @@
package patroni
import (
"errors"
"fmt"
"k8s.io/api/core/v1"
"testing"
)
func newMockPod(ip string) *v1.Pod {
return &v1.Pod{
Status: v1.PodStatus{
PodIP: ip,
},
}
}
func TestApiURL(t *testing.T) {
var testTable = []struct {
podIP string
expectedResponse string
expectedError error
}{
{
"127.0.0.1",
fmt.Sprintf("http://127.0.0.1:%d", apiPort),
nil,
},
{
"0000:0000:0000:0000:0000:0000:0000:0001",
fmt.Sprintf("http://[::1]:%d", apiPort),
nil,
},
{
"::1",
fmt.Sprintf("http://[::1]:%d", apiPort),
nil,
},
{
"",
"",
errors.New(" is not a valid IP"),
},
{
"foobar",
"",
errors.New("foobar is not a valid IP"),
},
{
"127.0.1",
"",
errors.New("127.0.1 is not a valid IP"),
},
{
":::",
"",
errors.New("::: is not a valid IP"),
},
}
for _, test := range testTable {
resp, err := apiURL(newMockPod(test.podIP))
if resp != test.expectedResponse {
t.Errorf("expected response %v does not match the actual %v", test.expectedResponse, resp)
}
if err != test.expectedError {
if err == nil || test.expectedError == nil {
t.Errorf("expected error '%v' does not match the actual error '%v'", test.expectedError, err)
}
if err != nil && test.expectedError != nil && err.Error() != test.expectedError.Error() {
t.Errorf("expected error '%v' does not match the actual error '%v'", test.expectedError, err)
}
}
}
}