Pod Anti Affinity Webhook

main.go

@@ -97,6 +97,10 @@ const (
 	walGExporterMemoryLimitFlg                  = "walg-exporter-memory-limit"
 	podAntiaffinityPreferredDuringSchedulingFlg = "pod-antiaffinity-preferred-during-scheduling"
 	podAntiaffinityTopologyKeyFlg               = "pod-antiaffinity-topology-key"
+	enablePodTopologySpreadConstraintWebhookFlg = "enable-pod-topology-spread-constraint-webhook"
+	podTopologySpreadConstraintTopologyKeyFlg   = "pod-topology-spread-constraint-topology-key"
+	podTopologySpreadConstraintMaxSkewFlg       = "pod-topology-spread-constraint-max-skew"
+	podTopologySpreadConstraintMinDomainsFlg    = "pod-topology-spread-constraint-min-domains"
 )
 
 var (
@@ -119,33 +123,34 @@ func init() {
 func main() {
 
 	var (
-		metricsAddrCtrlMgr         string
-		metricsAddrSvcMgr          string
-		partitionID                string
-		tenant                     string
-		ctrlClusterKubeconfig      string
-		pspName                    string
-		lbIP                       string
-		storageClass               string
-		postgresImage              string
-		etcdHost                   string
-		operatorImage              string
-		majorVersionUpgradeMode    string
-		postgresletNamespace       string
-		sidecarsCMName             string
-		controlPlaneNamespace      string
-		etcdImage                  string
-		etcdBackupSidecarImage     string
-		etcdBackupSecretName       string
-		etcdPSPName                string
-		postgresletFullname        string
-		initDBJobCMName            string
-		tlsClusterIssuer           string
-		tlsSubDomain               string
-		walGExporterImage          string
-		walGExporterCPULimit       string
-		walGExporterMemoryLimit    string
-		podAntiaffinityTopologyKey string
+		metricsAddrCtrlMgr                     string
+		metricsAddrSvcMgr                      string
+		partitionID                            string
+		tenant                                 string
+		ctrlClusterKubeconfig                  string
+		pspName                                string
+		lbIP                                   string
+		storageClass                           string
+		postgresImage                          string
+		etcdHost                               string
+		operatorImage                          string
+		majorVersionUpgradeMode                string
+		postgresletNamespace                   string
+		sidecarsCMName                         string
+		controlPlaneNamespace                  string
+		etcdImage                              string
+		etcdBackupSidecarImage                 string
+		etcdBackupSecretName                   string
+		etcdPSPName                            string
+		postgresletFullname                    string
+		initDBJobCMName                        string
+		tlsClusterIssuer                       string
+		tlsSubDomain                           string
+		walGExporterImage                      string
+		walGExporterCPULimit                   string
+		walGExporterMemoryLimit                string
+		podAntiaffinityTopologyKey             string
+		podTopologySpreadConstraintTopologyKey string
 
 		enableLeaderElection                     bool
 		enableCRDRegistration                    bool
@@ -164,10 +169,13 @@ func main() {
 		enableFsGroupChangePolicyWebhook         bool
 		enableWalGExporter                       bool
 		podAntiaffinityPreferredDuringScheduling bool
+		enablePodTopologySpreadConstraintWebhook bool
 
 		portRangeStart                        int32
 		portRangeSize                         int32
 		replicationChangeRequeueTimeInSeconds int
+		podTopologySpreadConstraintMaxSkew    int32
+		podTopologySpreadConstraintMinDomains int32
 
 		patroniTTL          uint32
 		patroniLoopWait     uint32
@@ -355,6 +363,14 @@ func main() {
 	walGExporterMemoryLimit = viper.GetString(walGExporterMemoryLimitFlg)
 	resource.MustParse(walGExporterMemoryLimit)
 
+	viper.SetDefault(enablePodTopologySpreadConstraintWebhookFlg, false)
+	enablePodTopologySpreadConstraintWebhook = viper.GetBool(enablePodTopologySpreadConstraintWebhookFlg)
+	viper.SetDefault(podTopologySpreadConstraintTopologyKeyFlg, "machine.metal-stack.io/rack")
+	podTopologySpreadConstraintTopologyKey = viper.GetString(podTopologySpreadConstraintTopologyKeyFlg)
+	viper.SetDefault(podTopologySpreadConstraintMaxSkewFlg, 1)
+	podTopologySpreadConstraintMaxSkew = viper.GetInt32(podTopologySpreadConstraintMaxSkewFlg)
+	podTopologySpreadConstraintMinDomains = viper.GetInt32(podTopologySpreadConstraintMinDomainsFlg)
+
 	ctrl.Log.Info("flag",
 		metricsAddrSvcMgrFlg, metricsAddrSvcMgr,
 		metricsAddrCtrlMgrFlg, metricsAddrCtrlMgr,
@@ -406,6 +422,10 @@ func main() {
 		walGExporterImageFlg, walGExporterImage,
 		walGExporterCPULimitFlg, walGExporterCPULimit,
 		walGExporterMemoryLimitFlg, walGExporterMemoryLimit,
+		enablePodTopologySpreadConstraintWebhookFlg, enablePodTopologySpreadConstraintWebhook,
+		podTopologySpreadConstraintTopologyKeyFlg, podTopologySpreadConstraintTopologyKey,
+		podTopologySpreadConstraintMaxSkewFlg, podTopologySpreadConstraintMaxSkew,
+		podTopologySpreadConstraintMinDomainsFlg, podTopologySpreadConstraintMinDomains,
 	)
 
 	svcClusterConf := ctrl.GetConfigOrDie()
@@ -549,18 +569,21 @@ func main() {
 	}
 	// +kubebuilder:scaffold:builder
 
-	if enableFsGroupChangePolicyWebhook {
-		svcClusterMgr.GetWebhookServer().Register(
-			"/mutate-v1-pod",
-			&webhook.Admission{
-				Handler: &webhooks.FsGroupChangePolicySetter{
-					SvcClient: svcClusterMgr.GetClient(),
-					Decoder:   admission.NewDecoder(svcClusterMgr.GetScheme()),
-					Log:       ctrl.Log.WithName("webhooks").WithName("FsGroupChangePolicySetter"),
-				},
+	svcClusterMgr.GetWebhookServer().Register(
+		"/mutate-v1-pod",
+		&webhook.Admission{
+			Handler: &webhooks.SpiloPodMutator{
+				SvcClient:                                svcClusterMgr.GetClient(),
+				Decoder:                                  admission.NewDecoder(svcClusterMgr.GetScheme()),
+				Log:                                      ctrl.Log.WithName("webhooks").WithName("SpiloPodMutator"),
+				EnableFsGroupChangePolicyWebhook:         enableFsGroupChangePolicyWebhook,
+				EnablePodTopologySpreadConstraintWebhook: enablePodTopologySpreadConstraintWebhook,
+				PodTopologySpreadConstraintTopologyKey:   podTopologySpreadConstraintTopologyKey,
+				PodTopologySpreadConstraintMaxSkew:       podTopologySpreadConstraintMaxSkew,
+				PodTopologySpreadConstraintMinDomains:    podTopologySpreadConstraintMinDomains,
 			},
-		)
-	}
+		},
+	)
 
 	ctx := context.Background()
 

pkg/webhooks/spiloPodMutator.go

@@ -0,0 +1,96 @@
+package webhooks
+
+import (
+	"context"
+	"encoding/json"
+	"net/http"
+
+	"github.com/go-logr/logr"
+	v1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"sigs.k8s.io/controller-runtime/pkg/webhook/admission"
+
+	pg "github.com/fi-ts/postgreslet/api/v1"
+)
+
+// +kubebuilder:webhook:path=/mutate-v1-pod,mutating=true,failurePolicy=ignore,groups=core,resources=pods,verbs=create;update,versions=v1,name=fsgroupchangepolicy.postgres.fits.cloud
+
+// SpiloPodMutator Adds securityContext.fsGroupChangePolicy=OnRootMismatch when the securityContext.fsGroup field is set
+type SpiloPodMutator struct {
+	SvcClient                                client.Client
+	Decoder                                  admission.Decoder
+	Log                                      logr.Logger
+	EnableFsGroupChangePolicyWebhook         bool
+	EnablePodTopologySpreadConstraintWebhook bool
+	PodTopologySpreadConstraintTopologyKey   string
+	PodTopologySpreadConstraintMaxSkew       int32
+	PodTopologySpreadConstraintMinDomains    int32
+}
+
+func (a *SpiloPodMutator) Handle(ctx context.Context, req admission.Request) admission.Response {
+	log := a.Log.WithValues("name", req.Name, "ns", req.Namespace)
+	log.V(1).Info("handling admission request")
+
+	pod := &v1.Pod{}
+	err := a.Decoder.Decode(req, pod)
+	if err != nil {
+		log.Error(err, "failed to decode request")
+		return admission.Errored(http.StatusBadRequest, err)
+	}
+
+	//
+	// FSGroupChangePolicy
+	//
+	if a.EnableFsGroupChangePolicyWebhook {
+		// when the fsGroup field is set, also set the fsGroupChangePolicy to OnRootMismatch
+		if pod.Spec.SecurityContext != nil && pod.Spec.SecurityContext.FSGroup != nil {
+			p := v1.FSGroupChangeOnRootMismatch
+			pod.Spec.SecurityContext.FSGroupChangePolicy = &p
+			log.V(1).Info("Mutating Pod securityContext", "pod", pod)
+		}
+	}
+
+	//
+	// TopologySpreadConstraint
+	//
+	if a.EnablePodTopologySpreadConstraintWebhook {
+		tsc := v1.TopologySpreadConstraint{
+			MaxSkew:           a.PodTopologySpreadConstraintMaxSkew,
+			WhenUnsatisfiable: v1.ScheduleAnyway,
+			LabelSelector: &metav1.LabelSelector{
+				MatchLabels: map[string]string{
+					"application":           "spilo",
+					"cluster-name":          pod.ObjectMeta.Labels["cluster-name"],
+					pg.NameLabelName:        pod.ObjectMeta.Labels[pg.NameLabelName],
+					pg.PartitionIDLabelName: pod.ObjectMeta.Labels[pg.PartitionIDLabelName],
+					pg.ProjectIDLabelName:   pod.ObjectMeta.Labels[pg.ProjectIDLabelName],
+					pg.TenantLabelName:      pod.ObjectMeta.Labels[pg.TenantLabelName],
+					pg.UIDLabelName:         pod.ObjectMeta.Labels[pg.UIDLabelName],
+					"team":                  pod.ObjectMeta.Labels["team"],
+				},
+			},
+			TopologyKey: a.PodTopologySpreadConstraintTopologyKey,
+		}
+
+		// if defined, set the minDomains (and corresponding whenUnsatisfied) field as well
+		if a.PodTopologySpreadConstraintMinDomains > 0 {
+			tsc.MinDomains = &a.PodTopologySpreadConstraintMinDomains
+			tsc.WhenUnsatisfiable = v1.DoNotSchedule
+		}
+
+		// override topology spread constraints
+		pod.Spec.TopologySpreadConstraints = []v1.TopologySpreadConstraint{tsc}
+		log.V(1).Info("Mutating Pod topologySpreadConstraints", "topologySpreadConstraints", pod.Spec.TopologySpreadConstraints)
+	}
+
+	marshaledPod, err := json.Marshal(pod)
+	if err != nil {
+		log.Error(err, "failed to marshal response")
+		return admission.Errored(http.StatusInternalServerError, err)
+	}
+
+	log.V(1).Info("done")
+	return admission.PatchResponseFromRaw(req.Object.Raw, marshaledPod)
+}