Merge pull request #275 from cybozu-go/add-pdb

Support PDB for egress

docs/usage.md

@@ -173,6 +173,7 @@ Step 5 and 6 are automatically done by Coil.
 It defines an egress portal of the cluster for some destinations.
 
 Coil creates a `Deployment` and `Service` for each `Egress`.
+It also creates a `PodDisruptionBudget` when `spec.podDisruptionBudget` is specified.
 
 Here is an example `Egress` resource for the Internet:
 
@@ -231,6 +232,8 @@ spec:
   sessionAffinityConfig:
     clientIP:
       timeoutSeconds: 43200
+  podDisruptionBudget:
+    maxUnavailable: 1
 ```
 
 Only `destinations` are mandatory.  Other fields in `spec` are optional.
@@ -244,6 +247,7 @@ You may customize the container of egress Pods as shown in the above example.
 | `template`              | [PodTemplateSpec][]       | Copied to Deployment's `spec.template`.                              |
 | `sessionAffinity`       | `ClusterIP` or `None`     | Copied to Service's `spec.sessionAffinity`.  Default is `ClusterIP`. |
 | `sessionAffinityConfig` | [SessionAffinityConfig][] | Copied to Service's `spec.sessionAffinityConfig`.                    |
+| `podDisruptionBudget`   | `EgressPDBSpec`           | `minAvailable` and `maxUnavailable` are copied to PDB's spec.        |
 
 ### Client Pods
 

v2/api/v2/egress_types.go

@@ -2,11 +2,15 @@ package v2
 
 import (
 	"net"
+	"strconv"
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	apivalidation "k8s.io/apimachinery/pkg/api/validation"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/validation"
+	"k8s.io/apimachinery/pkg/util/intstr"
+	utilvalidation "k8s.io/apimachinery/pkg/util/validation"
 	"k8s.io/apimachinery/pkg/util/validation/field"
 )
 
@@ -59,6 +63,10 @@ type EgressSpec struct {
 	// The default is false.
 	// +optional
 	FouSourcePortAuto bool `json:"fouSourcePortAuto,omitempty"`
+
+	// PodDisruptionBudget is an optional PodDisruptionBudget for Egress NAT pods.
+	// +optional
+	PodDisruptionBudget *EgressPDBSpec `json:"podDisruptionBudget,omitempty"`
 }
 
 // EgressPodTemplate defines pod template for Egress
@@ -75,6 +83,17 @@ type EgressPodTemplate struct {
 	Spec corev1.PodSpec `json:"spec,omitempty"`
 }
 
+// EgressPDB defines PDB for Egress
+type EgressPDBSpec struct {
+	// MinAvailable is the minimum number of pods that must be available at any given time.
+	// +optional
+	MinAvailable *intstr.IntOrString `json:"minAvailable,omitempty"`
+
+	// MaxUnavailable is the maximum number of pods that can be unavailable at any given time.
+	// +optional
+	MaxUnavailable *intstr.IntOrString `json:"maxUnavailable,omitempty"`
+}
+
 // Metadata defines a simplified version of ObjectMeta.
 type Metadata struct {
 	// Annotations are optional annotations
@@ -119,13 +138,76 @@ func (es EgressSpec) validate() field.ErrorList {
 		}
 	}
 
+	if es.PodDisruptionBudget != nil {
+		pp := p.Child("podDisruptionBudget")
+		allErrs = append(allErrs, validatePodDisruptionBudget(*es.PodDisruptionBudget, pp)...)
+	}
+
 	return allErrs
 }
 
 func (es EgressSpec) validateUpdate() field.ErrorList {
 	return es.validate()
 }
 
+// For validation of PodDisruptionBudget
+// Ref. https://github.com/kubernetes/kubernetes/blob/master/pkg/apis/policy/validation/validation.go
+func validatePodDisruptionBudget(pdb EgressPDBSpec, fldPath *field.Path) field.ErrorList {
+	allErrs := field.ErrorList{}
+
+	if pdb.MinAvailable != nil && pdb.MaxUnavailable != nil {
+		allErrs = append(allErrs, field.Invalid(fldPath, pdb, "minAvailable and maxUnavailable cannot be both set"))
+	}
+
+	if pdb.MinAvailable != nil {
+		allErrs = append(allErrs, validatePositiveIntOrPercent(*pdb.MinAvailable, fldPath.Child("minAvailable"))...)
+		allErrs = append(allErrs, isNotMoreThan100Percent(*pdb.MinAvailable, fldPath.Child("minAvailable"))...)
+	}
+
+	if pdb.MaxUnavailable != nil {
+		allErrs = append(allErrs, validatePositiveIntOrPercent(*pdb.MaxUnavailable, fldPath.Child("maxUnavailable"))...)
+		allErrs = append(allErrs, isNotMoreThan100Percent(*pdb.MaxUnavailable, fldPath.Child("maxUnavailable"))...)
+	}
+
+	return allErrs
+}
+
+func validatePositiveIntOrPercent(intOrPercent intstr.IntOrString, fldPath *field.Path) field.ErrorList {
+	allErrs := field.ErrorList{}
+	switch intOrPercent.Type {
+	case intstr.String:
+		for _, msg := range utilvalidation.IsValidPercent(intOrPercent.StrVal) {
+			allErrs = append(allErrs, field.Invalid(fldPath, intOrPercent, msg))
+		}
+	case intstr.Int:
+		allErrs = append(allErrs, apivalidation.ValidateNonnegativeField(int64(intOrPercent.IntValue()), fldPath)...)
+	default:
+		allErrs = append(allErrs, field.Invalid(fldPath, intOrPercent, "must be an integer or percentage (e.g '5%%')"))
+	}
+	return allErrs
+}
+
+func isNotMoreThan100Percent(intOrStringValue intstr.IntOrString, fldPath *field.Path) field.ErrorList {
+	allErrs := field.ErrorList{}
+	value, isPercent := getPercentValue(intOrStringValue)
+	if !isPercent || value <= 100 {
+		return nil
+	}
+	allErrs = append(allErrs, field.Invalid(fldPath, intOrStringValue, "must not be greater than 100%"))
+	return allErrs
+}
+
+func getPercentValue(intOrStringValue intstr.IntOrString) (int, bool) {
+	if intOrStringValue.Type != intstr.String {
+		return 0, false
+	}
+	if len(utilvalidation.IsValidPercent(intOrStringValue.StrVal)) != 0 {
+		return 0, false
+	}
+	value, _ := strconv.Atoi(intOrStringValue.StrVal[:len(intOrStringValue.StrVal)-1])
+	return value, true
+}
+
 // EgressStatus defines the observed state of Egress
 type EgressStatus struct {
 	// INSERT ADDITIONAL STATUS FIELD - define observed state of cluster

v2/api/v2/egress_webhook_test.go

@@ -9,6 +9,7 @@ import (
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/util/intstr"
 )
 
 func makeEgress() *Egress {
@@ -175,4 +176,27 @@ var _ = Describe("Egress Webhook", func() {
 		err = k8sClient.Update(ctx, r)
 		Expect(err).NotTo(HaveOccurred())
 	})
+
+	It("should allow valid PDB", func() {
+		r := makeEgress()
+		maxUnavailable := intstr.FromInt(1)
+		r.Spec.PodDisruptionBudget = &EgressPDBSpec{MaxUnavailable: &maxUnavailable}
+		err := k8sClient.Create(ctx, r)
+		Expect(err).NotTo(HaveOccurred())
+	})
+
+	It("should deny invalid PDB", func() {
+		r := makeEgress()
+		maxUnavailable := intstr.FromInt(1)
+		minAvailable := intstr.FromInt(1)
+		r.Spec.PodDisruptionBudget = &EgressPDBSpec{MaxUnavailable: &maxUnavailable, MinAvailable: &minAvailable}
+		err := k8sClient.Create(ctx, r)
+		Expect(err).To(HaveOccurred())
+
+		r = makeEgress()
+		maxUnavailable = intstr.FromString("120%")
+		r.Spec.PodDisruptionBudget = &EgressPDBSpec{MaxUnavailable: &maxUnavailable}
+		err = k8sClient.Create(ctx, r)
+		Expect(err).To(HaveOccurred())
+	})
 })

v2/config/crd/bases/coil.cybozu.com_egresses.yaml

@@ -52,6 +52,25 @@ spec:
                   If set to true, the kernel picks a flow based on the flow hash of the encapsulated packet.
                   The default is false.
                 type: boolean
+              podDisruptionBudget:
+                description: PodDisruptionBudget is an optional PodDisruptionBudget
+                  for Egress NAT pods.
+                properties:
+                  maxUnavailable:
+                    anyOf:
+                    - type: integer
+                    - type: string
+                    description: MaxUnavailable is the maximum number of pods that
+                      can be unavailable at any given time.
+                    x-kubernetes-int-or-string: true
+                  minAvailable:
+                    anyOf:
+                    - type: integer
+                    - type: string
+                    description: MinAvailable is the minimum number of pods that must
+                      be available at any given time.
+                    x-kubernetes-int-or-string: true
+                type: object
               replicas:
                 default: 1
                 description: |-

v2/config/rbac/coil-controller_role.yaml

@@ -94,6 +94,18 @@ rules:
   - get
   - patch
   - update
+- apiGroups:
+  - policy
+  resources:
+  - poddisruptionbudgets
+  verbs:
+  - create
+  - delete
+  - get
+  - list
+  - patch
+  - update
+  - watch
 - apiGroups:
   - rbac.authorization.k8s.io
   resources:

v2/controllers/egress_controller.go

@@ -8,6 +8,7 @@ import (
 	"github.com/go-logr/logr"
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	policyv1 "k8s.io/api/policy/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -32,6 +33,7 @@ type EgressReconciler struct {
 // +kubebuilder:rbac:groups=coil.cybozu.com,resources=egresses/status,verbs=get;update;patch
 // +kubebuilder:rbac:groups="",resources=services;serviceaccounts,verbs=get;list;watch;create;update;patch
 // +kubebuilder:rbac:groups=apps,resources=deployments,verbs=get;list;watch;create;update;patch
+// +kubebuilder:rbac:groups=policy,resources=poddisruptionbudgets,verbs=get;list;watch;create;update;patch;delete
 
 // coil-controller needs to have access to Pods to grant egress service accounts the same privilege.
 // +kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch
@@ -81,6 +83,11 @@ func (r *EgressReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctr
 		return ctrl.Result{}, err
 	}
 
+	if err := r.reconcilePDB(ctx, logger, eg); err != nil {
+		logger.Error(err, "failed to reconcile pod disruption budget")
+		return ctrl.Result{}, err
+	}
+
 	if err := r.updateStatus(ctx, logger, eg); err != nil {
 		logger.Error(err, "failed to update status")
 		return ctrl.Result{}, err
@@ -366,6 +373,56 @@ func (r *EgressReconciler) reconcileService(ctx context.Context, log logr.Logger
 	return nil
 }
 
+func (r *EgressReconciler) reconcilePDB(ctx context.Context, log logr.Logger, eg *coilv2.Egress) error {
+	if eg.Spec.PodDisruptionBudget == nil {
+		return nil
+	}
+
+	pdb := &policyv1.PodDisruptionBudget{}
+	pdb.Namespace = eg.Namespace
+	pdb.Name = eg.Name
+
+	result, err := ctrl.CreateOrUpdate(ctx, r.Client, pdb, func() error {
+		if pdb.DeletionTimestamp != nil {
+			return nil
+		}
+
+		if pdb.Labels == nil {
+			pdb.Labels = make(map[string]string)
+		}
+		for k, v := range selectorLabels(eg.Name) {
+			pdb.Labels[k] = v
+		}
+
+		// set immutable fields only for a new object
+		if pdb.CreationTimestamp.IsZero() {
+			if err := ctrl.SetControllerReference(eg, pdb, r.Scheme); err != nil {
+				return err
+			}
+		}
+
+		if eg.Spec.PodDisruptionBudget.MinAvailable != nil {
+			pdb.Spec.MinAvailable = eg.Spec.PodDisruptionBudget.MinAvailable
+		}
+		if eg.Spec.PodDisruptionBudget.MaxUnavailable != nil {
+			pdb.Spec.MaxUnavailable = eg.Spec.PodDisruptionBudget.MaxUnavailable
+		}
+		pdb.Spec.Selector = &metav1.LabelSelector{
+			MatchLabels: selectorLabels(eg.Name),
+		}
+
+		return nil
+	})
+	if err != nil {
+		return err
+	}
+
+	if result != controllerutil.OperationResultNone {
+		log.Info(string(result) + " pod disruption budget")
+	}
+	return nil
+}
+
 func (r *EgressReconciler) updateStatus(ctx context.Context, log logr.Logger, eg *coilv2.Egress) error {
 	depl := &appsv1.Deployment{}
 	if err := r.Get(ctx, client.ObjectKey{Namespace: eg.Namespace, Name: eg.Name}, depl); err != nil {
@@ -404,6 +461,7 @@ func (r *EgressReconciler) SetupWithManager(mgr ctrl.Manager) error {
 		For(&coilv2.Egress{}).
 		Owns(&appsv1.Deployment{}).
 		Owns(&corev1.Service{}).
+		Owns(&policyv1.PodDisruptionBudget{}).
 		Complete(r)
 }