[scheduler, mtsource] Initial HA support for MT Kafka source

config/source/multi/deployments/adapter.yaml

@@ -80,3 +80,18 @@ spec:
           containerPort: 8008
 
       terminationGracePeriodSeconds: 10
+      affinity:
+        podAntiAffinity:
+          preferredDuringSchedulingIgnoredDuringExecution:
+          - podAffinityTerm:
+              labelSelector:
+                matchLabels:
+                  control-plane: kafkasource-mt-adapter
+              topologyKey: kubernetes.io/hostname
+            weight: 50
+          - podAffinityTerm:
+              labelSelector:
+                matchLabels:
+                  control-plane: kafkasource-mt-adapter
+              topologyKey: topology.kubernetes.io/zone
+            weight: 50

config/source/multi/deployments/controller.yaml

@@ -47,12 +47,16 @@ spec:
 
         # How often (in seconds) the autoscaler tries to scale down the statefulset.
         - name: AUTOSCALER_REFRESH_PERIOD
-          value: '10'
+          value: '100'
 
         # The number of virtual replicas this pod can handle.
         - name: POD_CAPACITY
           value: '100'
 
+        # The scheduling policy type for placing vreplicas on pods (see type SchedulerPolicyType for enum list)
+        - name: SCHEDULER_POLICY_TYPE
+          value: 'MAXFILLUP'
+
         resources:
           requests:
             cpu: 20m

config/source/multi/roles/clusterrole.yaml

@@ -93,6 +93,7 @@ rules:
     - events
     - configmaps
     - secrets
+    - nodes
   verbs: *everything
 
 # let the webhook label the appropriate namespace

pkg/apis/duck/v1alpha1/placement_types.go

@@ -50,6 +50,9 @@ type Placement struct {
 	// PodName is the name of the pod where the resource is placed
 	PodName string `json:"podName,omitempty"`
 
+	// ZoneName is the name of the zone where the pod is located
+	ZoneName string `json:"zoneName,omitempty"`
+
 	// VReplicas is the number of virtual replicas assigned to in the pod
 	VReplicas int32 `json:"vreplicas,omitempty"`
 }

pkg/common/scheduler/statefulset/autoscaler.go

@@ -124,8 +124,10 @@ func (a *autoscaler) doautoscale(ctx context.Context, attemptScaleDown bool, pen
 		// The number of replicas may be lower than the last ordinal, for instance
 		// when the statefulset is manually scaled down. In that case, replicas above
 		// scale.Spec.Replicas have not been considered when scheduling vreplicas.
-		// Adjust accordingly
-		pending -= state.freeCapacity()
+		// Adjust accordingly (applicable only for MAXFILLUP scheduling policy and not for HA)
+		if state.schedulerPolicy != EVENSPREAD {
+			pending -= state.freeCapacity()
+		}
 
 		// Still need more?
 		if pending > 0 {

pkg/common/scheduler/statefulset/autoscaler_test.go

@@ -24,9 +24,9 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	gtesting "k8s.io/client-go/testing"
 
+	listers "knative.dev/eventing/pkg/reconciler/testing/v1"
 	kubeclient "knative.dev/pkg/client/injection/kube/client/fake"
 	_ "knative.dev/pkg/client/injection/kube/informers/apps/v1/statefulset/fake"
-	"knative.dev/pkg/logging"
 
 	duckv1alpha1 "knative.dev/eventing-kafka/pkg/apis/duck/v1alpha1"
 	"knative.dev/eventing-kafka/pkg/common/scheduler"
@@ -39,12 +39,13 @@ const (
 
 func TestAutoscaler(t *testing.T) {
 	testCases := []struct {
-		name         string
-		replicas     int32
-		vpods        []scheduler.VPod
-		pendings     int32
-		scaleDown    bool
-		wantReplicas int32
+		name            string
+		replicas        int32
+		vpods           []scheduler.VPod
+		pendings        int32
+		scaleDown       bool
+		wantReplicas    int32
+		schedulerPolicy SchedulerPolicyType
 	}{
 		{
 			name:     "no replicas, no placements, no pending",
@@ -181,17 +182,42 @@ func TestAutoscaler(t *testing.T) {
 			pendings:     int32(8),
 			wantReplicas: int32(3),
 		},
+		{
+			name:     "no replicas, with placements, with pending, enough capacity",
+			replicas: int32(0),
+			vpods: []scheduler.VPod{
+				tscheduler.NewVPod(testNs, "vpod-1", 15, []duckv1alpha1.Placement{
+					{PodName: "pod-0", VReplicas: int32(8)},
+					{PodName: "pod-1", VReplicas: int32(7)}}),
+			},
+			pendings:        int32(3),
+			wantReplicas:    int32(3),
+			schedulerPolicy: EVENSPREAD,
+		},
+		{
+			name:     "with replicas, with placements, with pending, enough capacity",
+			replicas: int32(2),
+			vpods: []scheduler.VPod{
+				tscheduler.NewVPod(testNs, "vpod-1", 15, []duckv1alpha1.Placement{
+					{PodName: "pod-0", VReplicas: int32(8)},
+					{PodName: "pod-1", VReplicas: int32(7)}}),
+			},
+			pendings:        int32(3),
+			wantReplicas:    int32(3),
+			schedulerPolicy: EVENSPREAD,
+		},
 	}
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
 			ctx, _ := setupFakeContext(t)
 
 			vpodClient := tscheduler.NewVPodClient()
-			stateAccessor := newStateBuilder(logging.FromContext(ctx), vpodClient.List, 10)
+			ls := listers.NewListers(nil)
+			stateAccessor := newStateBuilder(ctx, vpodClient.List, 10, tc.schedulerPolicy, ls.GetNodeLister())
 
 			sfsClient := kubeclient.Get(ctx).AppsV1().StatefulSets(testNs)
-			_, err := sfsClient.Create(ctx, makeStatefulset(ctx, testNs, sfsName, tc.replicas), metav1.CreateOptions{})
+			_, err := sfsClient.Create(ctx, makeStatefulset(testNs, sfsName, tc.replicas), metav1.CreateOptions{})
 			if err != nil {
 				t.Fatal("unexpected error", err)
 			}
@@ -231,10 +257,11 @@ func TestAutoscalerScaleDownToZero(t *testing.T) {
 	})
 
 	vpodClient := tscheduler.NewVPodClient()
-	stateAccessor := newStateBuilder(logging.FromContext(ctx), vpodClient.List, 10)
+	ls := listers.NewListers(nil)
+	stateAccessor := newStateBuilder(ctx, vpodClient.List, 10, MAXFILLUP, ls.GetNodeLister())
 
 	sfsClient := kubeclient.Get(ctx).AppsV1().StatefulSets(testNs)
-	_, err := sfsClient.Create(ctx, makeStatefulset(ctx, testNs, sfsName, 10), metav1.CreateOptions{})
+	_, err := sfsClient.Create(ctx, makeStatefulset(testNs, sfsName, 10), metav1.CreateOptions{})
 	if err != nil {
 		t.Fatal("unexpected error", err)
 	}