Merge remote-tracking branch 'origin' into lb-kill-switch

Author: eberlep

api/v1/postgres_types.go

@@ -700,10 +700,11 @@ func (p *Postgres) ToUnstructuredZalandoPostgresql(z *zalando.Postgresql, c *cor
 	// finally, overwrite the (special to us) shared buffer parameter
 	setSharedBufferSize(z.Spec.PostgresqlParam.Parameters, p.Spec.Size.SharedBuffer)
 
-	z.Spec.Resources.ResourceRequests.CPU = p.Spec.Size.CPU
-	z.Spec.Resources.ResourceRequests.Memory = p.Spec.Size.Memory
-	z.Spec.Resources.ResourceLimits.CPU = p.Spec.Size.CPU
-	z.Spec.Resources.ResourceLimits.Memory = p.Spec.Size.Memory
+	z.Spec.Resources = &zalando.Resources{}
+	z.Spec.Resources.ResourceRequests.CPU = pointer.String(p.Spec.Size.CPU)
+	z.Spec.Resources.ResourceRequests.Memory = pointer.String(p.Spec.Size.Memory)
+	z.Spec.Resources.ResourceLimits.CPU = pointer.String(p.Spec.Size.CPU)
+	z.Spec.Resources.ResourceLimits.Memory = pointer.String(p.Spec.Size.Memory)
 	z.Spec.TeamID = p.generateTeamID()
 	z.Spec.Volume.Size = p.Spec.Size.StorageSize
 	z.Spec.Volume.StorageClass = sc
@@ -790,12 +791,9 @@ func (p *Postgres) ToUnstructuredZalandoPostgresql(z *zalando.Postgresql, c *cor
 	} else {
 		// overwrite connection info
 		z.Spec.StandbyCluster = &zalando.StandbyDescription{
-			StandbyMethod:          StandbyMethod,
-			StandbyHost:            p.Spec.PostgresConnection.ConnectionIP,
-			StandbyPort:            strconv.FormatInt(int64(p.Spec.PostgresConnection.ConnectionPort), 10),
-			StandbySecretName:      "standby." + p.ToPeripheralResourceName() + ".credentials",
-			S3WalPath:              "",
-			StandbyApplicationName: p.ObjectMeta.Name,
+			StandbyHost: p.Spec.PostgresConnection.ConnectionIP,
+			StandbyPort: strconv.FormatInt(int64(p.Spec.PostgresConnection.ConnectionPort), 10),
+			// S3WalPath:              "",
 		}
 	}
 

controllers/postgres_controller.go

@@ -56,12 +56,19 @@ import (
 const (
 	postgresExporterServiceName                    string = "postgres-exporter"
 	postgresExporterServicePortName                string = "metrics"
+	postgresExporterServicePortKeyName             string = "postgres-exporter-service-port"
+	postgresExporterServiceTargetPortKeyName       string = "postgres-exporter-service-target-port"
+	walGExporterServicePortName                    string = "backup-metrics"
+	walGExporterServicePortKeyName                 string = "wal-g-exporter-service-port"
+	walGExporterServiceTargetPortKeyName           string = "wal-g-exporter-service-target-port"
 	postgresExporterServiceTenantAnnotationName    string = pg.TenantLabelName
 	postgresExporterServiceProjectIDAnnotationName string = pg.ProjectIDLabelName
 	storageEncryptionKeyName                       string = "storage-encryption-key"
 	storageEncryptionKeyFinalizerName              string = "postgres.database.fits.cloud/secret-finalizer"
 	walGEncryptionSecretNamePostfix                string = "-walg-encryption"
 	walGEncryptionSecretKeyName                    string = "key"
+	podMonitorName                                 string = "patroni"
+	podMonitorPort                                 string = "8008"
 	initDBName                                     string = "postgres-initdb"
 	initDBSQLDummy                                 string = `SELECT 'NOOP';`
 	debugLogLevel                                  int    = 1
@@ -102,6 +109,7 @@ type PostgresReconciler struct {
 	EnableCustomTLSCert                 bool
 	TLSClusterIssuer                    string
 	TLSSubDomain                        string
+	EnableWalGExporter                  bool
 }
 
 type PatroniStandbyCluster struct {
@@ -300,6 +308,12 @@ func (r *PostgresReconciler) Reconcile(ctx context.Context, req ctrl.Request) (c
 		return ctrl.Result{}, fmt.Errorf("error while creating sidecars servicemonitor %v: %w", namespace, err)
 	}
 
+	// Add pod monitor
+	if err := r.createOrUpdatePatroniPodMonitor(ctx, namespace, instance); err != nil {
+		r.recorder.Eventf(instance, "Warning", "Error", "failed to create podmonitor: %v", err)
+		return ctrl.Result{}, fmt.Errorf("error while creating podmonitor %v: %w", namespace, err)
+	}
+
 	// Make sure the storage secret exist, if necessary
 	if err := r.ensureStorageEncryptionSecret(log, ctx, instance); err != nil {
 		r.recorder.Eventf(instance, "Warning", "Error", "failed to create storage secret: %v", err)
@@ -1473,16 +1487,28 @@ func (r *PostgresReconciler) createOrUpdateNetPol(ctx context.Context, instance
 
 // createOrUpdateExporterSidecarServices ensures the necessary services to access the sidecars exist
 func (r *PostgresReconciler) createOrUpdateExporterSidecarServices(log logr.Logger, ctx context.Context, namespace string, c *corev1.ConfigMap, in *pg.Postgres) error {
-	exporterServicePort, error := strconv.ParseInt(c.Data["postgres-exporter-service-port"], 10, 32)
+	pesPort, error := strconv.ParseInt(c.Data[postgresExporterServicePortKeyName], 10, 32)
+	if error != nil {
+		log.Error(error, "postgres-exporter-service-port could not be parsed to int32, falling back to default value")
+		pesPort = 9187
+	}
+
+	pesTargetPort, error := strconv.ParseInt(c.Data[postgresExporterServiceTargetPortKeyName], 10, 32)
+	if error != nil {
+		log.Error(error, "postgres-exporter-service-target-port could not be parsed to int32, falling back to default value")
+		pesTargetPort = pesPort
+	}
+
+	wgesPort, error := strconv.ParseInt(c.Data[walGExporterServicePortKeyName], 10, 32)
 	if error != nil {
-		// todo log error
-		exporterServicePort = 9187
+		log.Error(error, "wal-g-exporter-service-port could not be parsed to int32, falling back to default value")
+		pesPort = 9351
 	}
 
-	exporterServiceTargetPort, error := strconv.ParseInt(c.Data["postgres-exporter-service-target-port"], 10, 32)
+	wgesTargetPort, error := strconv.ParseInt(c.Data[walGExporterServiceTargetPortKeyName], 10, 32)
 	if error != nil {
-		// todo log error
-		exporterServiceTargetPort = exporterServicePort
+		log.Error(error, "wal-g-exporter-service-target-port could not be parsed to int32, falling back to default value")
+		pesTargetPort = pesPort
 	}
 
 	labels := map[string]string{
@@ -1506,11 +1532,20 @@ func (r *PostgresReconciler) createOrUpdateExporterSidecarServices(log logr.Logg
 	pes.Spec.Ports = []corev1.ServicePort{
 		{
 			Name:       postgresExporterServicePortName,
-			Port:       int32(exporterServicePort), //nolint
+			Port:       int32(pesPort), //nolint
 			Protocol:   corev1.ProtocolTCP,
-			TargetPort: intstr.FromInt(int(exporterServiceTargetPort)),
+			TargetPort: intstr.FromInt(int(pesTargetPort)),
 		},
 	}
+	if r.EnableWalGExporter {
+		wgesp := corev1.ServicePort{
+			Name:       walGExporterServicePortName,
+			Port:       int32(wgesPort), //nolint
+			Protocol:   corev1.ProtocolTCP,
+			TargetPort: intstr.FromInt(int(wgesTargetPort)),
+		}
+		pes.Spec.Ports = append(pes.Spec.Ports, wgesp)
+	}
 	selector := map[string]string{
 		pg.ApplicationLabelName: pg.ApplicationLabelValue,
 	}
@@ -1586,6 +1621,12 @@ func (r *PostgresReconciler) createOrUpdateExporterSidecarServiceMonitor(log log
 			Port: postgresExporterServicePortName,
 		},
 	}
+	if r.EnableWalGExporter {
+		wgesme := coreosv1.Endpoint{
+			Port: walGExporterServicePortName,
+		}
+		pesm.Spec.Endpoints = append(pesm.Spec.Endpoints, wgesme)
+	}
 	pesm.Spec.NamespaceSelector = coreosv1.NamespaceSelector{
 		MatchNames: []string{namespace},
 	}
@@ -1596,6 +1637,10 @@ func (r *PostgresReconciler) createOrUpdateExporterSidecarServiceMonitor(log log
 	pesm.Spec.Selector = metav1.LabelSelector{
 		MatchLabels: matchLabels,
 	}
+	pesm.Spec.TargetLabels = []string{
+		"postgres_partition_id=" + in.Spec.PartitionID,
+		"is_primary=" + strconv.FormatBool(in.IsReplicationPrimaryOrStandalone()),
+	}
 
 	// try to fetch any existing postgres-exporter service
 	ns := types.NamespacedName{
@@ -1623,6 +1668,69 @@ func (r *PostgresReconciler) createOrUpdateExporterSidecarServiceMonitor(log log
 	return nil
 }
 
+// createOrUpdatePatroniPodMonitor ensures the servicemonitors for the sidecars exist
+func (r *PostgresReconciler) createOrUpdatePatroniPodMonitor(ctx context.Context, namespace string, in *pg.Postgres) error {
+	log := r.Log.WithValues("namespace", namespace)
+
+	labels := map[string]string{
+		"app":     "postgres-exporter",
+		"release": "prometheus",
+	}
+
+	annotations := map[string]string{
+		postgresExporterServiceTenantAnnotationName:    in.Spec.Tenant,
+		postgresExporterServiceProjectIDAnnotationName: in.Spec.ProjectID,
+	}
+
+	pm := &coreosv1.PodMonitor{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        podMonitorName,
+			Namespace:   namespace,
+			Labels:      labels,
+			Annotations: annotations,
+		},
+	}
+
+	pm.Spec.PodMetricsEndpoints = []coreosv1.PodMetricsEndpoint{
+		{
+			Port: podMonitorPort,
+		},
+	}
+	pm.Spec.NamespaceSelector = coreosv1.NamespaceSelector{
+		MatchNames: []string{namespace},
+	}
+	matchLabels := map[string]string{
+		"application": "spilo",
+	}
+	pm.Spec.Selector = metav1.LabelSelector{
+		MatchLabels: matchLabels,
+	}
+
+	// try to fetch any existing podmonitor
+	ns := types.NamespacedName{
+		Namespace: namespace,
+		Name:      podMonitorName,
+	}
+	old := &coreosv1.PodMonitor{}
+	if err := r.SvcClient.Get(ctx, ns, old); err == nil {
+		// Copy the resource version
+		pm.ObjectMeta.ResourceVersion = old.ObjectMeta.ResourceVersion
+		if err := r.SvcClient.Update(ctx, pm); err != nil {
+			return fmt.Errorf("error while updating the podmonitor: %w", err)
+		}
+		log.Info("pod monitor updated")
+		return nil
+	}
+
+	// local podmonitor does not exist, creating it
+	if err := r.SvcClient.Create(ctx, pm); err != nil {
+		return fmt.Errorf("error while creating the podmonitor: %w", err)
+	}
+	log.Info("podmonitor created")
+
+	return nil
+}
+
 func (r *PostgresReconciler) deleteExporterSidecarService(log logr.Logger, ctx context.Context, namespace string) error {
 	s := &corev1.Service{
 		ObjectMeta: metav1.ObjectMeta{

go.mod

@@ -12,7 +12,7 @@ require (
 	github.com/onsi/gomega v1.27.10
 	github.com/prometheus-operator/prometheus-operator/pkg/apis/monitoring v0.56.2
 	github.com/spf13/viper v1.12.0
-	github.com/zalando/postgres-operator v1.7.0
+	github.com/zalando/postgres-operator v1.11.0
 	k8s.io/api v0.28.9
 	k8s.io/apiextensions-apiserver v0.28.9
 	k8s.io/apimachinery v0.28.9
@@ -73,15 +73,15 @@ require (
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.25.0 // indirect
 	golang.org/x/crypto v0.21.0 // indirect
-	golang.org/x/exp v0.0.0-20220722155223-a9213eeb770e // indirect
+	golang.org/x/exp v0.0.0-20240112132812-db7319d0e0e3 // indirect
 	golang.org/x/mod v0.14.0 // indirect
 	golang.org/x/net v0.23.0 // indirect
 	golang.org/x/oauth2 v0.8.0 // indirect
 	golang.org/x/sys v0.18.0 // indirect
 	golang.org/x/term v0.18.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	golang.org/x/time v0.3.0 // indirect
-	golang.org/x/tools v0.16.1 // indirect
+	golang.org/x/tools v0.17.0 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/protobuf v1.33.0 // indirect
@@ -98,5 +98,3 @@ require (
 	sigs.k8s.io/structured-merge-diff/v4 v4.2.3 // indirect
 	sigs.k8s.io/yaml v1.3.0 // indirect
 )
-
-replace github.com/zalando/postgres-operator v1.7.0 => github.com/ermajn/postgres-operator v1.0.1-0.20211123085256-711648b7fdde