Merge pull request #1366 from eggfoobar/arbiter-node-support

OCPEDGE-1195: Arbiter node support

pkg/etcdcli/etcdcli.go

@@ -516,6 +516,11 @@ func endpoints(nodeLister corev1listers.NodeLister,
 		if err != nil {
 			return nil, fmt.Errorf("failed to list control plane nodes: %w", err)
 		}
+		nodesArbiter, err := nodeLister.List(labels.Set{"node-role.kubernetes.io/arbiter": ""}.AsSelector())
+		if err != nil {
+			return nil, fmt.Errorf("failed to list arbiter nodes: %w", err)
+		}
+		nodes = append(nodes, nodesArbiter...)
 		for _, node := range nodes {
 			internalIP, _, err := dnshelpers.GetPreferredInternalIPAddressForNodeName(network, node)
 			if err != nil {

pkg/operator/ceohelpers/control_plane_topology.go

@@ -1,10 +1,13 @@
 package ceohelpers
 
 import (
+	"context"
 	"fmt"
 
 	configv1 "github.com/openshift/api/config/v1"
+	configv1client "github.com/openshift/client-go/config/clientset/versioned/typed/config/v1"
 	configv1listers "github.com/openshift/client-go/config/listers/config/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/klog/v2"
 )
 
@@ -31,3 +34,15 @@ func IsSingleNodeTopology(infraLister configv1listers.InfrastructureLister) (boo
 
 	return topology == configv1.SingleReplicaTopologyMode, nil
 }
+
+// IsArbiterNodeTopology returns if the cluster infrastructure ControlPlaneTopology is set to HighlyAvailableArbiterMode
+// We use the infra interface in this situation instead of the lister because typically you are looking to find out this information
+// in order to configure controllers before the informers are running.
+func IsArbiterNodeTopology(ctx context.Context, infraClient configv1client.InfrastructureInterface) (bool, error) {
+	infra, err := infraClient.Get(ctx, InfrastructureClusterName, metav1.GetOptions{})
+	if err != nil {
+		return false, err
+	}
+	return infra.Status.ControlPlaneTopology == configv1.HighlyAvailableArbiterMode, nil
+
+}

pkg/operator/ceohelpers/informer.go

@@ -0,0 +1,120 @@
+package ceohelpers
+
+import (
+	"context"
+	time "time"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	runtime "k8s.io/apimachinery/pkg/runtime"
+	watch "k8s.io/apimachinery/pkg/watch"
+	kubernetes "k8s.io/client-go/kubernetes"
+	corev1listers "k8s.io/client-go/listers/core/v1"
+	cache "k8s.io/client-go/tools/cache"
+)
+
+// NewMultiSelectorNodeLister create a node lister that allows for querying based on multiple selectors for ORing on keys.
+// This is done because of a performance design choice in selector queries that does not allow for ORing on keys.
+// see: https://github.com/kubernetes/kubernetes/issues/90549#issuecomment-620625847
+//
+// NOTE: If you are selecting based on ORing values, that is already supported, please use a format like `key in (value1,value2)`
+// see: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#set-based-requirement
+func NewMultiSelectorNodeLister(indexer cache.Indexer, extraSelectors ...labels.Selector) corev1listers.NodeLister {
+	return &mergedNodeLister{indexer: indexer, extraSelectors: extraSelectors}
+}
+
+// NewMultiSelectorNodeInformer create a node informer with multiple selector types for ORing on the keys of node types.
+// This is done because of a performance design choice in selector queries that does not allow for ORing on keys.
+// see: https://github.com/kubernetes/kubernetes/issues/90549#issuecomment-620625847
+//
+// NOTE: If you are selecting based on ORing values, that is already supported, please use a format like `key in (value1,value2)`
+// see: https://kubernetes.io/docs/concepts/overview/working-with-objects/labels/#set-based-requirement
+func NewMultiSelectorNodeInformer(client kubernetes.Interface, resyncPeriod time.Duration, indexers cache.Indexers, selectors ...string) cache.SharedIndexInformer {
+	return cache.NewSharedIndexInformer(
+		&cache.ListWatch{
+			ListFunc: func(options metav1.ListOptions) (runtime.Object, error) {
+				var filteredNodes *corev1.NodeList
+				for _, selector := range selectors {
+					options.LabelSelector = selector
+					nodes, err := client.CoreV1().Nodes().List(context.TODO(), options)
+					if err != nil {
+						return nil, err
+					}
+					if filteredNodes == nil {
+						filteredNodes = nodes
+					} else {
+						filteredNodes.Items = append(filteredNodes.Items, nodes.Items...)
+					}
+				}
+				return filteredNodes, nil
+			},
+			WatchFunc: func(options metav1.ListOptions) (watch.Interface, error) {
+				var filteredNodeWatchers mergedWatchFunc
+				for _, selector := range selectors {
+					options.LabelSelector = selector
+					nodeWatcher, err := client.CoreV1().Nodes().Watch(context.TODO(), options)
+					if err != nil {
+						return nil, err
+					}
+					filteredNodeWatchers = append(filteredNodeWatchers, nodeWatcher)
+				}
+				return filteredNodeWatchers, nil
+			},
+		},
+		&corev1.Node{},
+		resyncPeriod,
+		indexers,
+	)
+}
+
+type mergedNodeLister struct {
+	indexer        cache.Indexer
+	extraSelectors []labels.Selector
+}
+
+// List lists all Nodes in the indexer.
+func (s *mergedNodeLister) List(selector labels.Selector) (ret []*corev1.Node, err error) {
+	err = cache.ListAll(s.indexer, selector, func(m interface{}) {
+		ret = append(ret, m.(*corev1.Node))
+	})
+	for _, selector := range s.extraSelectors {
+		err = cache.ListAll(s.indexer, selector, func(m interface{}) {
+			ret = append(ret, m.(*corev1.Node))
+		})
+	}
+	return ret, err
+}
+
+// Get retrieves the Node from the index for a given name.
+func (s *mergedNodeLister) Get(name string) (*corev1.Node, error) {
+	obj, exists, err := s.indexer.GetByKey(name)
+	if err != nil {
+		return nil, err
+	}
+	if !exists {
+		return nil, errors.NewNotFound(corev1.Resource("node"), name)
+	}
+	return obj.(*corev1.Node), nil
+}
+
+type mergedWatchFunc []watch.Interface
+
+func (funcs mergedWatchFunc) Stop() {
+	for _, fun := range funcs {
+		fun.Stop()
+	}
+}
+
+func (funcs mergedWatchFunc) ResultChan() <-chan watch.Event {
+	out := make(chan watch.Event)
+	for _, fun := range funcs {
+		go func(eventChannel <-chan watch.Event) {
+			for v := range eventChannel {
+				out <- v
+			}
+		}(fun.ResultChan())
+	}
+	return out
+}

pkg/operator/ceohelpers/informer_test.go

@@ -0,0 +1,177 @@
+package ceohelpers
+
+import (
+	"context"
+	"testing"
+	time "time"
+
+	"github.com/stretchr/testify/require"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
+	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/client-go/kubernetes/fake"
+	"k8s.io/client-go/tools/cache"
+)
+
+const (
+	masterNodeLabelSelectorString  = "node-role.kubernetes.io/master"
+	arbiterNodeLabelSelectorString = "node-role.kubernetes.io/arbiter"
+)
+
+func TestMultiSelectors(t *testing.T) {
+	testCases := []struct {
+		name                string
+		nodes               []corev1.Node
+		expectedNodes       []corev1.Node
+		expectedErrorReason string
+	}{
+		{
+			name: "control plane nodes should be returned",
+			nodes: []corev1.Node{
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "master0",
+						Labels: map[string]string{
+							masterNodeLabelSelectorString: "",
+						},
+					},
+				},
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "master1",
+						Labels: map[string]string{
+							masterNodeLabelSelectorString: "",
+						},
+					},
+				},
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "master2",
+						Labels: map[string]string{
+							masterNodeLabelSelectorString: "",
+						},
+					},
+				},
+			},
+			expectedNodes: []corev1.Node{
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "master0",
+					},
+				},
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "master1",
+					},
+				},
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "master2",
+					},
+				},
+			},
+		},
+		{
+			name: "control plane nodes should be returned with arbiter",
+			nodes: []corev1.Node{
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "master0",
+						Labels: map[string]string{
+							masterNodeLabelSelectorString: "",
+						},
+					},
+				},
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "master1",
+						Labels: map[string]string{
+							masterNodeLabelSelectorString: "",
+						},
+					},
+				},
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "arbiter0",
+						Labels: map[string]string{
+							arbiterNodeLabelSelectorString: "",
+						},
+					},
+				},
+			},
+			expectedNodes: []corev1.Node{
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "master0",
+					},
+				},
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "master1",
+					},
+				},
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "arbiter0",
+					},
+				},
+			},
+		},
+		{
+			name: "control plane nodes should not include non labeled arbiter nodes",
+			nodes: []corev1.Node{
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "arbiter0",
+					},
+				},
+			},
+			expectedNodes: []corev1.Node{
+				{
+					ObjectMeta: v1.ObjectMeta{
+						Name: "arbiter0",
+					},
+				},
+			},
+			expectedErrorReason: "NotFound",
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+
+			kubeClient := fake.NewClientset()
+			for _, n := range tc.nodes {
+				kubeClient.Tracker().Add(&n)
+			}
+
+			arbiterNodeLabelSelector, err := labels.Parse(arbiterNodeLabelSelectorString)
+			require.Nil(t, err)
+
+			informer := NewMultiSelectorNodeInformer(
+				kubeClient,
+				1*time.Hour,
+				cache.Indexers{cache.NamespaceIndex: cache.MetaNamespaceIndexFunc},
+				masterNodeLabelSelectorString, arbiterNodeLabelSelectorString)
+
+			lister := NewMultiSelectorNodeLister(informer.GetIndexer(), arbiterNodeLabelSelector)
+
+			ctx, cancel := context.WithCancel(context.TODO())
+			defer cancel()
+
+			go informer.Run(ctx.Done())
+			cache.WaitForCacheSync(ctx.Done(), informer.HasSynced)
+
+			for _, n := range tc.expectedNodes {
+				node, err := lister.Get(n.Name)
+				if tc.expectedErrorReason != "" {
+					require.Equal(t, tc.expectedErrorReason, string(errors.ReasonForError(err)))
+				} else {
+					require.Nil(t, err)
+					require.NotEmpty(t, node)
+				}
+			}
+
+		})
+	}
+}

pkg/operator/configobservation/controlplanereplicascount/observe_control_plane_replicas_count.go

@@ -67,5 +67,13 @@ func readDesiredControlPlaneReplicas(configMapListerForKubeSystemNamespace corev
 		return 0, fmt.Errorf("required field: %s.controlPlane.replicas doesn't exist in cm: %s/kube-system", installConfigKeyName, clusterConfigConfigMapName)
 	}
 
+	desiredArbiterReplicas, exists, err := unstructured.NestedFloat64(unstructuredInstallConfig, "arbiterNode", "replicas")
+	if err != nil {
+		return 0, fmt.Errorf("failed to extract field: %s.arbiterNode.replicas from cm: %s/kube-system, err: %v", installConfigKeyName, clusterConfigConfigMapName, err)
+	}
+	if exists {
+		desiredReplicas += desiredArbiterReplicas
+	}
+
 	return int(desiredReplicas), nil
 }