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

Use ConfigMap to store operator's config

Browse Source
This commit is contained in:
Murat Kabilov
2017-05-12 11:41:34 +02:00
parent 47e3e29a56
commit da438aab3a
16 changed files with 323 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/cluster/pod.go
+1 -1
View File
@@ -183,7 +183,7 @@ func (c *Cluster) recreatePods() error {
var masterPod v1.Pod
for _, pod := range pods.Items {
role := util.PodSpiloRole(&pod)
role := c.PodSpiloRole(&pod)
if role == constants.PodRoleMaster {
masterPod = pod
pkg/cluster/util.go
+16 -8
View File
@@ -103,7 +103,7 @@ func (c *Cluster) waitForPodLabel(podEvents chan spec.PodEvent) error {
for {
select {
case podEvent := <-podEvents:
role := util.PodSpiloRole(podEvent.CurPod)
role := c.PodSpiloRole(podEvent.CurPod)
// We cannot assume any role of the newly created pod. Normally, for a multi-pod cluster
// we should observe the 'replica' value, but it could be that some pods are not allowed
// to promote, therefore, the new pod could be a master as well.
@@ -156,10 +156,14 @@ func (c *Cluster) waitPodLabelsReady() error {
LabelSelector: ls.String(),
}
masterListOption := v1.ListOptions{
LabelSelector: labels.Merge(ls, labels.Set{constants.SpiloRoleLabel: constants.PodRoleMaster}).String(),
LabelSelector: labels.Merge(ls, labels.Set{
c.OpConfig.PodRoleLabel: constants.PodRoleMaster,
}).String(),
}
replicaListOption := v1.ListOptions{
LabelSelector: labels.Merge(ls, labels.Set{constants.SpiloRoleLabel: constants.PodRoleReplica}).String(),
LabelSelector: labels.Merge(ls, labels.Set{
c.OpConfig.PodRoleLabel: constants.PodRoleReplica,
}).String(),
}
pods, err := c.KubeClient.Pods(namespace).List(listOptions)
if err != nil {
@@ -203,10 +207,10 @@ func (c *Cluster) waitStatefulsetPodsReady() error {
}
func (c *Cluster) labelsSet() labels.Set {
return labels.Set{
constants.ApplicationNameLabel: constants.ApplicationNameLabelValue,
constants.ClusterNameLabel: c.Metadata.Name,
}
lbls := c.OpConfig.ClusterLabels
lbls[c.OpConfig.ClusterNameLabel] = c.Metadata.Name
return labels.Set(lbls)
}
func (c *Cluster) dnsName() string {
@@ -228,7 +232,7 @@ func (c *Cluster) credentialSecretName(username string) string {
}
func (c *Cluster) deleteEtcdKey() error {
etcdKey := fmt.Sprintf("/service/%s", c.Metadata.Name)
etcdKey := fmt.Sprintf("/%s/%s", c.OpConfig.EtcdScope, c.Metadata.Name)
//TODO: retry multiple times
resp, err := c.EtcdClient.Delete(context.Background(),
@@ -245,3 +249,7 @@ func (c *Cluster) deleteEtcdKey() error {
return nil
}
func (c *Cluster) PodSpiloRole(pod *v1.Pod) string {
return pod.Labels[c.OpConfig.PodRoleLabel]
}
pkg/controller/controller.go
-2
View File
@@ -17,7 +17,6 @@ import (
)
type Config struct {
PodNamespace string
KubeClient *kubernetes.Clientset
RestClient *rest.RESTClient
EtcdClient etcdclient.KeysAPI
@@ -63,7 +62,6 @@ func (c *Controller) Run(stopCh <-chan struct{}, wg *sync.WaitGroup) {
c.initController()
c.logger.Infof("'%s' namespace will be watched", c.PodNamespace)
go c.runInformers(stopCh)
c.logger.Info("Started working in background")
pkg/controller/pod.go
+5 -5
View File
@@ -29,7 +29,7 @@ func (c *Controller) podListFunc(options api.ListOptions) (runtime.Object, error
TimeoutSeconds: options.TimeoutSeconds,
}
return c.KubeClient.CoreV1().Pods(c.PodNamespace).List(opts)
return c.KubeClient.CoreV1().Pods(c.opConfig.Namespace).List(opts)
}
func (c *Controller) podWatchFunc(options api.ListOptions) (watch.Interface, error) {
@@ -52,7 +52,7 @@ func (c *Controller) podWatchFunc(options api.ListOptions) (watch.Interface, err
TimeoutSeconds: options.TimeoutSeconds,
}
return c.KubeClient.CoreV1Client.Pods(c.PodNamespace).Watch(opts)
return c.KubeClient.CoreV1Client.Pods(c.opConfig.Namespace).Watch(opts)
}
func (c *Controller) podAdd(obj interface{}) {
@@ -62,7 +62,7 @@ func (c *Controller) podAdd(obj interface{}) {
}
podEvent := spec.PodEvent{
ClusterName: util.PodClusterName(pod),
ClusterName: c.PodClusterName(pod),
PodName: util.NameFromMeta(pod.ObjectMeta),
CurPod: pod,
EventType: spec.PodEventAdd,
@@ -83,7 +83,7 @@ func (c *Controller) podUpdate(prev, cur interface{}) {
}
podEvent := spec.PodEvent{
ClusterName: util.PodClusterName(curPod),
ClusterName: c.PodClusterName(curPod),
PodName: util.NameFromMeta(curPod.ObjectMeta),
PrevPod: prevPod,
CurPod: curPod,
@@ -100,7 +100,7 @@ func (c *Controller) podDelete(obj interface{}) {
}
podEvent := spec.PodEvent{
ClusterName: util.PodClusterName(pod),
ClusterName: c.PodClusterName(pod),
PodName: util.NameFromMeta(pod.ObjectMeta),
CurPod: pod,
EventType: spec.PodEventDelete,
pkg/controller/postgresql.go
+2 -2
View File
@@ -19,7 +19,7 @@ import (
func (c *Controller) clusterListFunc(options api.ListOptions) (runtime.Object, error) {
c.logger.Info("Getting list of currently running clusters")
object, err := c.RestClient.Get().
Namespace(c.PodNamespace).
Namespace(c.opConfig.Namespace).
Resource(constants.ResourceName).
VersionedParams(&options, api.ParameterCodec).
FieldsSelectorParam(fields.Everything()).
@@ -65,7 +65,7 @@ func (c *Controller) clusterListFunc(options api.ListOptions) (runtime.Object, e
func (c *Controller) clusterWatchFunc(options api.ListOptions) (watch.Interface, error) {
return c.RestClient.Get().
Prefix("watch").
Namespace(c.PodNamespace).
Namespace(c.opConfig.Namespace).
Resource(constants.ResourceName).
VersionedParams(&options, api.ParameterCodec).
FieldsSelectorParam(fields.Everything()).
pkg/controller/util.go
+12 -1
View File
@@ -72,7 +72,7 @@ func (c *Controller) createTPR() error {
restClient := c.RestClient
return k8sutil.WaitTPRReady(restClient, c.opConfig.TPR.ReadyWaitInterval, c.opConfig.TPR.ReadyWaitTimeout, c.PodNamespace)
return k8sutil.WaitTPRReady(restClient, c.opConfig.TPR.ReadyWaitInterval, c.opConfig.TPR.ReadyWaitTimeout, c.opConfig.Namespace)
}
func (c *Controller) getInfrastructureRoles() (result map[string]spec.PgUser, err error) {
@@ -125,3 +125,14 @@ Users:
return result, nil
}
func (c *Controller) PodClusterName(pod *v1.Pod) spec.NamespacedName {
if name, ok := pod.Labels[c.opConfig.ClusterNameLabel]; ok {
return spec.NamespacedName{
Namespace: pod.Namespace,
Name: name,
}
}
return spec.NamespacedName{}
}
pkg/util/config/config.go
+48 -39
View File
@@ -2,64 +2,53 @@ package config
import (
"encoding/json"
"fmt"
"strings"
"time"
"github.bus.zalan.do/acid/postgres-operator/pkg/spec"
"github.com/kelseyhightower/envconfig"
)
type TPR struct {
ReadyWaitInterval time.Duration `split_words:"true" default:"3s"`
ReadyWaitTimeout time.Duration `split_words:"true" default:"30s"`
ResyncPeriod time.Duration `split_words:"true" default:"5m"`
ReadyWaitInterval time.Duration `name:"ready_wait_interval" default:"4s"`
ReadyWaitTimeout time.Duration `name:"ready_wait_timeout" default:"30s"`
ResyncPeriod time.Duration `name:"resync_period" default:"5m"`
}
type Resources struct {
ResyncPeriodPod time.Duration `split_words:"true" default:"5m"`
ResourceCheckInterval time.Duration `split_words:"true" default:"3s"`
ResourceCheckTimeout time.Duration `split_words:"true" default:"10m"`
PodLabelWaitTimeout time.Duration `split_words:"true" default:"10m"`
PodDeletionWaitTimeout time.Duration `split_words:"true" default:"10m"`
ResyncPeriodPod time.Duration `name:"resync_period_pod" default:"5m"`
ResourceCheckInterval time.Duration `name:"resource_check_interval" default:"3s"`
ResourceCheckTimeout time.Duration `name:"resource_check_timeout" default:"10m"`
PodLabelWaitTimeout time.Duration `name:"pod_label_wait_timeout" default:"10m"`
PodDeletionWaitTimeout time.Duration `name:"pod_deletion_wait_timeout" default:"10m"`
ClusterLabels map[string]string `name:"cluster_labels" default:"application:spilo"`
ClusterNameLabel string `name:"cluster_name_label" default:"cluster-name"`
PodRoleLabel string `name:"pod_role_label" default:"spilo-role"`
}
type Auth struct {
PamRoleName string `split_words:"true" default:"zalandos"`
PamConfiguration string `split_words:"true" default:"https://info.example.com/oauth2/tokeninfo?access_token= uid realm=/employees"`
TeamsAPIUrl string `envconfig:"teams_api_url" default:"https://teams.example.com/api/"`
OAuthTokenSecretName spec.NamespacedName `envconfig:"oauth_token_secret_name" default:"postgresql-operator"`
InfrastructureRolesSecretName spec.NamespacedName `split_words:"true"`
SuperUsername string `split_words:"true" default:"postgres"`
ReplicationUsername string `split_words:"true" default:"replication"`
PamRoleName string `name:"pam_rol_name" default:"zalandos"`
PamConfiguration string `name:"pam_configuration" default:"https://info.example.com/oauth2/tokeninfo?access_token= uid realm=/employees"`
TeamsAPIUrl string `name:"teams_api_url" default:"https://teams.example.com/api/"`
OAuthTokenSecretName spec.NamespacedName `name:"oauth_token_secret_name" default:"postgresql-operator"`
InfrastructureRolesSecretName spec.NamespacedName `name:"infrastructure_roles_secret_name"`
SuperUsername string `name:"super_username" default:"postgres"`
ReplicationUsername string `name:"replication_username" default:"replication"`
}
type Config struct {
TPR
Resources
Auth
EtcdHost string `split_words:"true" default:"etcd-client.default.svc.cluster.local:2379"`
DockerImage string `split_words:"true" default:"registry.opensource.zalan.do/acid/spilo-9.6:1.2-p12"`
ServiceAccountName string `split_words:"true" default:"operator"`
DbHostedZone string `split_words:"true" default:"db.example.com"`
EtcdScope string `split_words:"true" default:"service"`
WALES3Bucket string `envconfig:"wal_s3_bucket"`
KubeIAMRole string `envconfig:"kube_iam_role"`
DebugLogging bool `split_words:"true" default:"false"`
DNSNameFormat string `envconfig:"dns_name_format" default:"%s.%s.%s"`
}
func LoadFromEnv() *Config {
//TODO: maybe we should use ConfigMaps( https://kubernetes.io/docs/tasks/configure-pod-container/configmap/ ) instead?
var cfg Config
err := envconfig.Process("PGOP", &cfg)
if err != nil {
panic(fmt.Errorf("Can't read config: %v", err))
}
cfg.EtcdScope = strings.Trim(cfg.EtcdScope, "/")
return &cfg
Namespace string `name:"namespace"`
EtcdHost string `name:"etcd_host" default:"etcd-client.default.svc.cluster.local:2379"`
DockerImage string `name:"docker_image" default:"registry.opensource.zalan.do/acid/spilo-9.6:1.2-p12"`
ServiceAccountName string `name:"service_account_name" default:"operator"`
DbHostedZone string `name:"db_hosted_zone" default:"db.example.com"`
EtcdScope string `name:"etcd_scope" default:"service"`
WALES3Bucket string `name:"wal_s3_bucket"`
KubeIAMRole string `name:"kube_iam_role"`
DebugLogging bool `name:"debug_logging" default:"false"`
DNSNameFormat string `name:"dns_name_format" default:"%s.%s.%s"`
}
func (c Config) MustMarshal() string {
@@ -70,3 +59,23 @@ func (c Config) MustMarshal() string {
return string(b)
}
func NewFromMap(m map[string]string) *Config {
cfg := Config{}
fields, _ := structFields(&cfg)
for _, structField := range fields {
key := strings.ToLower(structField.Name)
value, ok := m[key]
if !ok && structField.Default != "" {
value = structField.Default
}
err := processField(value, structField.Field)
if err != nil {
panic(err)
}
}
return &cfg
}
pkg/util/config/util.go
+164
View File
@@ -0,0 +1,164 @@
package config
import (
"fmt"
"reflect"
"strconv"
"strings"
"time"
)
type Decoder interface {
Decode(value string) error
}
type fieldInfo struct {
Name string
Default string
Field reflect.Value
}
func decoderFrom(field reflect.Value) (d Decoder) {
// it may be impossible for a struct field to fail this check
if !field.CanInterface() {
return
}
d, ok := field.Interface().(Decoder)
if !ok && field.CanAddr() {
d, ok = field.Addr().Interface().(Decoder)
}
return d
}
// taken from github.com/kelseyhightower/envconfig
func structFields(spec interface{}) ([]fieldInfo, error) {
s := reflect.ValueOf(spec).Elem()
// over allocate an info array, we will extend if needed later
infos := make([]fieldInfo, 0, s.NumField())
for i := 0; i < s.NumField(); i++ {
f := s.Field(i)
ftype := s.Type().Field(i)
fieldName := ftype.Tag.Get("name")
if fieldName == "" {
fieldName = strings.ToLower(ftype.Name)
}
// Capture information about the config variable
info := fieldInfo{
Name: fieldName,
Field: f,
Default: ftype.Tag.Get("default"),
}
infos = append(infos, info)
if f.Kind() == reflect.Struct {
// honor Decode if present
if decoderFrom(f) == nil {
embeddedPtr := f.Addr().Interface()
embeddedInfos, err := structFields(embeddedPtr)
if err != nil {
return nil, err
}
infos = append(infos[:len(infos)-1], embeddedInfos...)
continue
}
}
}
return infos, nil
}
func processField(value string, field reflect.Value) error {
typ := field.Type()
decoder := decoderFrom(field)
if decoder != nil {
return decoder.Decode(value)
}
if typ.Kind() == reflect.Ptr {
typ = typ.Elem()
if field.IsNil() {
field.Set(reflect.New(typ))
}
field = field.Elem()
}
switch typ.Kind() {
case reflect.String:
field.SetString(value)
case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
var (
val int64
err error
)
if field.Kind() == reflect.Int64 && typ.PkgPath() == "time" && typ.Name() == "Duration" {
var d time.Duration
d, err = time.ParseDuration(value)
val = int64(d)
} else {
val, err = strconv.ParseInt(value, 0, typ.Bits())
}
if err != nil {
return err
}
field.SetInt(val)
case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64:
val, err := strconv.ParseUint(value, 0, typ.Bits())
if err != nil {
return err
}
field.SetUint(val)
case reflect.Bool:
val, err := strconv.ParseBool(value)
if err != nil {
return err
}
field.SetBool(val)
case reflect.Float32, reflect.Float64:
val, err := strconv.ParseFloat(value, typ.Bits())
if err != nil {
return err
}
field.SetFloat(val)
case reflect.Slice:
vals := strings.Split(value, ",")
sl := reflect.MakeSlice(typ, len(vals), len(vals))
for i, val := range vals {
err := processField(val, sl.Index(i))
if err != nil {
return err
}
}
field.Set(sl)
case reflect.Map:
pairs := strings.Split(value, ",")
mp := reflect.MakeMap(typ)
for _, pair := range pairs {
kvpair := strings.Split(pair, ":")
if len(kvpair) != 2 {
return fmt.Errorf("invalid map item: %q", pair)
}
k := reflect.New(typ.Key()).Elem()
err := processField(kvpair[0], k)
if err != nil {
return err
}
v := reflect.New(typ.Elem()).Elem()
err = processField(kvpair[1], v)
if err != nil {
return err
}
mp.SetMapIndex(k, v)
}
field.Set(mp)
}
return nil
}
pkg/util/constants/constants.go
+12 -16
View File
@@ -2,20 +2,16 @@ package constants
const (
//Constants
TPRName = "postgresql"
TPRVendor = "acid.zalan.do"
TPRDescription = "Managed PostgreSQL clusters"
TPRApiVersion = "v1"
DataVolumeName = "pgdata"
PasswordLength = 64
UserSecretTemplate = "%s.%s.credentials.%s.%s" // Username, ClusterName, TPRName, TPRVendor
ZalandoDnsNameAnnotation = "zalando.org/dnsname"
KubeIAmAnnotation = "iam.amazonaws.com/role"
ResourceName = TPRName + "s"
PodRoleMaster = "master"
PodRoleReplica = "replica"
ApplicationNameLabel = "application"
ApplicationNameLabelValue = "spilo"
SpiloRoleLabel = "spilo-role"
ClusterNameLabel = "version"
TPRName = "postgresql"
TPRVendor = "acid.zalan.do"
TPRDescription = "Managed PostgreSQL clusters"
TPRApiVersion = "v1"
DataVolumeName = "pgdata"
PasswordLength = 64
UserSecretTemplate = "%s.%s.credentials.%s.%s" // Username, ClusterName, TPRName, TPRVendor
ZalandoDnsNameAnnotation = "zalando.org/dnsname"
KubeIAmAnnotation = "iam.amazonaws.com/role"
ResourceName = TPRName + "s"
PodRoleMaster = "master"
PodRoleReplica = "replica"
)
pkg/util/util.go
-16
View File
@@ -12,7 +12,6 @@ import (
"k8s.io/client-go/pkg/api/v1"
"github.bus.zalan.do/acid/postgres-operator/pkg/spec"
"github.bus.zalan.do/acid/postgres-operator/pkg/util/constants"
)
var passwordChars = []byte("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789")
@@ -37,21 +36,6 @@ func NameFromMeta(meta v1.ObjectMeta) spec.NamespacedName {
}
}
func PodClusterName(pod *v1.Pod) spec.NamespacedName {
if name, ok := pod.Labels[constants.ClusterNameLabel]; ok {
return spec.NamespacedName{
Namespace: pod.Namespace,
Name: name,
}
}
return spec.NamespacedName{}
}
func PodSpiloRole(pod *v1.Pod) string {
return pod.Labels[constants.SpiloRoleLabel]
}
func PGUserPassword(user spec.PgUser) string {
s := md5.Sum([]byte(user.Password + user.Name))