Adjust datastore syncer to watch the local Gateway

...instead of the local Node for global IP updates.

Signed-off-by: Tom Pantelis <[email protected]>

Author: tpantelis

pkg/controllers/datastoresyncer/datastore_endpoint_sync_test.go

@@ -32,7 +32,6 @@ import (
 	testutil "github.com/submariner-io/admiral/pkg/test"
 	submarinerv1 "github.com/submariner-io/submariner/pkg/apis/submariner.io/v1"
 	"github.com/submariner-io/submariner/pkg/globalnet/constants"
-	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/runtime/schema"
@@ -123,30 +122,30 @@ func testEndpointSyncing() {
 		})
 	})
 
-	When("the local Node's global IP is updated", func() {
-		var node *corev1.Node
+	When("the local Gateway's global IP is updated", func() {
+		var gateway *submarinerv1.Gateway
 
 		BeforeEach(func() {
-			node = &corev1.Node{
+			gateway = &submarinerv1.Gateway{
 				ObjectMeta: metav1.ObjectMeta{
-					Name:        nodeName,
+					Name:        t.localEndpoint.Hostname,
 					Annotations: map[string]string{constants.SmGlobalIP: "200.0.0.40"},
 				},
 			}
 
-			test.CreateResource(t.localNodes, node)
+			test.CreateResource(t.localGateways, gateway)
 		})
 
 		JustBeforeEach(func() {
-			t.localEndpoint.HealthCheckIP = node.Annotations[constants.SmGlobalIP]
+			t.localEndpoint.HealthCheckIP = gateway.Annotations[constants.SmGlobalIP]
 			awaitEndpoint(t.localEndpoints, t.localEndpoint)
 		})
 
 		It("should update the local Endpoint's HealthCheckIP", func() {
-			node.Annotations[constants.SmGlobalIP] = "200.0.0.100"
-			t.localEndpoint.HealthCheckIP = node.Annotations[constants.SmGlobalIP]
+			gateway.Annotations[constants.SmGlobalIP] = "200.0.0.100"
+			t.localEndpoint.HealthCheckIP = gateway.Annotations[constants.SmGlobalIP]
 
-			test.UpdateResource(t.localNodes, node)
+			test.UpdateResource(t.localGateways, gateway)
 			awaitEndpoint(t.localEndpoints, t.localEndpoint)
 		})
 
@@ -155,8 +154,8 @@ func testEndpointSyncing() {
 				Expect(t.localEndpoints.Delete(context.Background(), getEndpointName(t.localEndpoint), metav1.DeleteOptions{})).
 					To(Succeed())
 
-				node.Annotations[constants.SmGlobalIP] = "200.0.0.100"
-				test.UpdateResource(t.localNodes, node)
+				gateway.Annotations[constants.SmGlobalIP] = "200.0.0.100"
+				test.UpdateResource(t.localGateways, gateway)
 
 				testutil.EnsureNoResource(resource.ForDynamic(t.localEndpoints), getEndpointName(t.localEndpoint))
 			})

pkg/controllers/datastoresyncer/datastoresyncer.go

@@ -20,7 +20,6 @@ package datastoresyncer
 
 import (
 	"context"
-	"os"
 
 	"github.com/pkg/errors"
 	"github.com/submariner-io/admiral/pkg/federate"
@@ -33,10 +32,10 @@ import (
 	"github.com/submariner-io/submariner/pkg/cidr"
 	"github.com/submariner-io/submariner/pkg/endpoint"
 	"github.com/submariner-io/submariner/pkg/types"
-	k8sv1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/dynamic"
@@ -47,7 +46,6 @@ import (
 type DatastoreSyncer struct {
 	localCluster  types.SubmarinerCluster
 	localEndpoint *endpoint.Local
-	localNodeName string
 	syncerConfig  broker.SyncerConfig
 }
 
@@ -94,8 +92,8 @@ func (d *DatastoreSyncer) Start(ctx context.Context) error {
 	}
 
 	if len(d.localCluster.Spec.GlobalCIDR) > 0 {
-		if err := d.startNodeWatcher(ctx.Done()); err != nil {
-			return errors.WithMessage(err, "startNodeWatcher returned error")
+		if err := d.startGatewayWatcher(ctx.Done()); err != nil {
+			return errors.WithMessage(err, "startGatewayWatcher returned error")
 		}
 	}
 
@@ -231,45 +229,34 @@ func (d *DatastoreSyncer) ensureExclusiveEndpoint(ctx context.Context, syncer *b
 	return nil
 }
 
-func (d *DatastoreSyncer) startNodeWatcher(stopCh <-chan struct{}) error {
-	nodeName, ok := os.LookupEnv("NODE_NAME")
-	if !ok {
-		// Healthcheck in globalnet deployments will not work because of missing NODE_NAME.
-		logger.Error(nil, "Error reading the NODE_NAME from the env, healthChecker functionality will not work.")
-	} else {
-		d.localNodeName = nodeName
-		return d.createNodeWatcher(stopCh)
-	}
-
-	return nil
-}
-
-func (d *DatastoreSyncer) createNodeWatcher(stopCh <-chan struct{}) error {
+func (d *DatastoreSyncer) startGatewayWatcher(stopCh <-chan struct{}) error {
 	resourceWatcher, err := watcher.New(&watcher.Config{
 		Scheme:     scheme.Scheme,
 		RestConfig: d.syncerConfig.LocalRestConfig,
 		RestMapper: d.syncerConfig.RestMapper,
 		Client:     d.syncerConfig.LocalClient,
 		ResourceConfigs: []watcher.ResourceConfig{
 			{
-				Name:                "Node watcher for datastoresyncer",
-				ResourceType:        &k8sv1.Node{},
-				ResourcesEquivalent: d.areNodesEquivalent,
+				Name:                "Gateway watcher for datastoresyncer",
+				ResourceType:        &submarinerv1.Gateway{},
+				SourceNamespace:     d.syncerConfig.LocalNamespace,
+				ResourcesEquivalent: d.areGatewaysEquivalent,
+				SourceFieldSelector: fields.Set(map[string]string{"metadata.name": d.localEndpoint.Spec().Hostname}).AsSelector().String(),
 				Handler: watcher.EventHandlerFuncs{
-					OnCreateFunc: d.handleCreateOrUpdateNode,
-					OnUpdateFunc: d.handleCreateOrUpdateNode,
+					OnCreateFunc: d.handleCreateOrUpdateGateway,
+					OnUpdateFunc: d.handleCreateOrUpdateGateway,
 					OnDeleteFunc: nil,
 				},
 			},
 		},
 	})
 	if err != nil {
-		return errors.Wrap(err, "error creating resource watcher for Nodes")
+		return errors.Wrap(err, "error creating Gateway resource watcher")
 	}
 
 	err = resourceWatcher.Start(stopCh)
 	if err != nil {
-		return errors.Wrap(err, "error starting the resource watcher")
+		return errors.Wrap(err, "error starting the Gateway resource watcher")
 	}
 
 	return nil

pkg/controllers/datastoresyncer/datastoresyncer_suite_test.go

@@ -21,7 +21,6 @@ package datastoresyncer_test
 import (
 	"context"
 	"fmt"
-	"os"
 	"reflect"
 	"testing"
 	"time"
@@ -53,7 +52,6 @@ const (
 	otherClusterID  = "west"
 	localNamespace  = "submariner"
 	brokerNamespace = "submariner-broker"
-	nodeName        = "raiders"
 )
 
 func TestDatastoresyncer(t *testing.T) {
@@ -83,7 +81,7 @@ type testDriver struct {
 	localClusters    dynamic.ResourceInterface
 	brokerClusters   dynamic.ResourceInterface
 	localEndpoints   dynamic.ResourceInterface
-	localNodes       dynamic.ResourceInterface
+	localGateways    dynamic.ResourceInterface
 	brokerEndpoints  dynamic.ResourceInterface
 	syncerScheme     *runtime.Scheme
 	restMapper       meta.RESTMapper
@@ -128,7 +126,7 @@ func newTestDriver() *testDriver {
 		t.brokerClient = dynamicfake.NewSimpleDynamicClient(t.syncerScheme)
 		fake.AddBasicReactors(&t.brokerClient.Fake)
 
-		t.restMapper = test.GetRESTMapperFor(&submarinerv1.Cluster{}, &submarinerv1.Endpoint{}, &corev1.Node{})
+		t.restMapper = test.GetRESTMapperFor(&submarinerv1.Cluster{}, &submarinerv1.Endpoint{}, &submarinerv1.Gateway{}, &corev1.Node{})
 
 		clusterGVR := test.GetGroupVersionResourceFor(t.restMapper, &submarinerv1.Cluster{})
 		t.localClusters = t.localClient.Resource(*clusterGVR).Namespace(localNamespace)
@@ -138,7 +136,8 @@ func newTestDriver() *testDriver {
 		t.localEndpoints = t.localClient.Resource(*endpointGVR).Namespace(localNamespace)
 		t.brokerEndpoints = t.brokerClient.Resource(*endpointGVR).Namespace(brokerNamespace)
 
-		t.localNodes = t.localClient.Resource(*test.GetGroupVersionResourceFor(t.restMapper, &corev1.Node{})).Namespace("")
+		t.localGateways = t.localClient.Resource(*test.GetGroupVersionResourceFor(t.restMapper, &submarinerv1.Gateway{})).
+			Namespace(localNamespace)
 	})
 
 	JustBeforeEach(func() {
@@ -153,8 +152,6 @@ func newTestDriver() *testDriver {
 }
 
 func (t *testDriver) run() {
-	os.Setenv("NODE_NAME", nodeName)
-
 	t.syncer = datastoresyncer.New(&broker.SyncerConfig{
 		LocalClient:     t.localClient,
 		LocalNamespace:  localNamespace,

pkg/controllers/datastoresyncer/node_handler.go renamed to pkg/controllers/datastoresyncer/gateway_handler.go

@@ -22,44 +22,34 @@ import (
 	"context"
 	"net"
 
+	"github.com/submariner-io/admiral/pkg/resource"
 	submarinerv1 "github.com/submariner-io/submariner/pkg/apis/submariner.io/v1"
 	"github.com/submariner-io/submariner/pkg/globalnet/constants"
-	k8sv1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
 )
 
-func (d *DatastoreSyncer) handleCreateOrUpdateNode(obj runtime.Object, _ int) bool {
-	node := obj.(*k8sv1.Node)
-	if node.Name != d.localNodeName {
-		return false
-	}
-
-	globalIPOfNode := node.GetAnnotations()[constants.SmGlobalIP]
+func (d *DatastoreSyncer) handleCreateOrUpdateGateway(obj runtime.Object, _ int) bool {
+	globalIP := resource.MustToMeta(obj).GetAnnotations()[constants.SmGlobalIP]
 
-	// Validate that globalIPOfNode falls in the globalCIDR allocated to the cluster.
-	if globalIPOfNode != "" {
+	// Validate that the global IP falls in the global CIDR allocated to the cluster.
+	if globalIP != "" {
 		_, ipnet, err := net.ParseCIDR(d.localCluster.Spec.GlobalCIDR[0])
 		if err != nil {
 			// Ideally this will not happen as globalCIDR is expected to be a valid CIDR.
 			logger.Errorf(err, "Error parsing the GlobalCIDR %q", d.localCluster.Spec.GlobalCIDR)
 			return false
 		}
 
-		if ipnet.Contains(net.ParseIP(globalIPOfNode)) {
-			return d.updateLocalEndpointIfNecessary(globalIPOfNode)
+		if ipnet.Contains(net.ParseIP(globalIP)) {
+			return d.updateLocalEndpointIfNecessary(globalIP)
 		}
 	}
 
 	return false
 }
 
-func (d *DatastoreSyncer) areNodesEquivalent(obj1, obj2 *unstructured.Unstructured) bool {
-	if obj1.GetName() != d.localNodeName {
-		// Ignore this event. We are only interested in active GatewayNode events.
-		return true
-	}
-
+func (d *DatastoreSyncer) areGatewaysEquivalent(obj1, obj2 *unstructured.Unstructured) bool {
 	existingGlobalIP := obj1.GetAnnotations()[constants.SmGlobalIP]
 	newGlobalIP := obj2.GetAnnotations()[constants.SmGlobalIP]
 
@@ -70,16 +60,16 @@ func (d *DatastoreSyncer) areNodesEquivalent(obj1, obj2 *unstructured.Unstructur
 	return existingGlobalIP == newGlobalIP
 }
 
-func (d *DatastoreSyncer) updateLocalEndpointIfNecessary(globalIPOfNode string) bool {
+func (d *DatastoreSyncer) updateLocalEndpointIfNecessary(globalIP string) bool {
 	spec := d.localEndpoint.Spec()
-	if spec.HealthCheckIP != globalIPOfNode {
-		logger.Infof("Updating the endpoint HealthCheckIP to globalIP %q", globalIPOfNode)
+	if spec.HealthCheckIP != globalIP {
+		logger.Infof("Updating the endpoint HealthCheckIP to globalIP %q", globalIP)
 
 		err := d.localEndpoint.Update(context.TODO(), func(existing *submarinerv1.EndpointSpec) {
-			existing.HealthCheckIP = globalIPOfNode
+			existing.HealthCheckIP = globalIP
 		})
 		if err != nil {
-			logger.Warningf("Error updating the local submariner Endpoint with HealthcheckIP %s: %v", globalIPOfNode, err)
+			logger.Warningf("Error updating the local submariner Endpoint with HealthcheckIP %s: %v", globalIP, err)
 			return true
 		}
 	}