Specify NamespaceInformer for SI and EPS resource syncers

...so resource creation is retried if the service namespace doesn't initially exist and is subsequently created. Signed-off-by: Tom Pantelis <[email protected]>
submariner-io · May 17, 2024 · d64c5a0 · d64c5a0
1 parent a7ea394
commit d64c5a0
Show file tree

Hide file tree

Showing 7 changed files with 75 additions and 16 deletions.
diff --git a/go.mod b/go.mod
@@ -8,7 +8,7 @@ require (
 	github.com/onsi/gomega v1.31.1
 	github.com/pkg/errors v0.9.1
 	github.com/prometheus/client_golang v1.18.0
-	github.com/submariner-io/admiral v0.17.1
+	github.com/submariner-io/admiral v0.17.2-0.20240516172424-ba3d8efe4148
 	github.com/submariner-io/shipyard v0.17.1
 	k8s.io/api v0.29.3
 	k8s.io/apimachinery v0.29.3

diff --git a/go.sum b/go.sum
@@ -388,8 +388,8 @@ github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/stretchr/testify v1.8.4 h1:CcVxjf3Q8PM0mHUKJCdn+eZZtm5yQwehR5yeSVQQcUk=
 github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
-github.com/submariner-io/admiral v0.17.1 h1:p7XkFfbKaQArlzwzZrU4tW5etGjzjlslLhG1p9b/vgs=
-github.com/submariner-io/admiral v0.17.1/go.mod h1:38k64mD4hRBQ2UR1VeaJsrIJmmxa3vo5TdevdEUM93k=
+github.com/submariner-io/admiral v0.17.2-0.20240516172424-ba3d8efe4148 h1:geJLegFPzA2YUkpPpewCgjnSsECQuWuRRnBQ4MWOA7A=
+github.com/submariner-io/admiral v0.17.2-0.20240516172424-ba3d8efe4148/go.mod h1:38k64mD4hRBQ2UR1VeaJsrIJmmxa3vo5TdevdEUM93k=
 github.com/submariner-io/shipyard v0.17.1 h1:CGKBOwl9cU9aVG7jS+PV7FjMZOmFKY0YuQJa8y2Yo1c=
 github.com/submariner-io/shipyard v0.17.1/go.mod h1:hhlOvzkProcAJe/3cm52FYUEknAzJlBmIkRfjl/SMSw=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=

diff --git a/pkg/agent/controller/agent.go b/pkg/agent/controller/agent.go
@@ -108,6 +108,17 @@ func New(spec *AgentSpecification, syncerConf broker.SyncerConfig, syncerMetricN
 		return nil, errors.Wrap(err, "error creating Service syncer")
 	}
 
+	syncerConf.NamespaceInformer, err = syncer.NewSharedInformer(&syncer.ResourceSyncerConfig{
+		SourceClient: syncerConf.LocalClient,
+		RestMapper:   syncerConf.RestMapper,
+		ResourceType: &corev1.Namespace{},
+	})
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating namespace informer")
+	}
+
+	agentController.namespaceInformer = syncerConf.NamespaceInformer
+
 	agentController.serviceExportClient = &ServiceExportClient{
 		NamespaceableResourceInterface: syncerConf.LocalClient.Resource(*gvr),
 		converter:                      converter{scheme: syncerConf.Scheme},
@@ -139,6 +150,10 @@ func (a *Controller) Start(stopCh <-chan struct{}) error {
 	// Start the informer factories to begin populating the informer caches
 	logger.Info("Starting Agent controller")
 
+	go func() {
+		a.namespaceInformer.Run(stopCh)
+	}()
+
 	if err := a.serviceExportSyncer.Start(stopCh); err != nil {
 		return errors.Wrap(err, "error starting ServiceExport syncer")
 	}

diff --git a/pkg/agent/controller/clusterip_service_test.go b/pkg/agent/controller/clusterip_service_test.go
@@ -23,6 +23,7 @@ import (
 	"strconv"
 
 	. "github.com/onsi/ginkgo/v2"
+	"github.com/submariner-io/admiral/pkg/fake"
 	"github.com/submariner-io/admiral/pkg/resource"
 	"github.com/submariner-io/admiral/pkg/syncer/test"
 	testutil "github.com/submariner-io/admiral/pkg/test"
@@ -288,6 +289,48 @@ func testClusterIPServiceInOneCluster() {
 			test.RemoteNamespace)), "other-eps")
 	})
 
+	When("the namespace of an exported service does not initially exist on an importing cluster", func() {
+		BeforeEach(func() {
+			fake.AddVerifyNamespaceReactor(&t.cluster2.localDynClient.Fake, "serviceimports", "endpointslices")
+		})
+
+		JustBeforeEach(func() {
+			t.cluster1.createService()
+			t.cluster1.createServiceExport()
+		})
+
+		It("should eventually import the service when the namespace is created", func() {
+			expServiceImport := &mcsv1a1.ServiceImport{
+				ObjectMeta: metav1.ObjectMeta{
+					Name:      t.cluster1.service.Name,
+					Namespace: t.cluster1.service.Namespace,
+				},
+				Spec: mcsv1a1.ServiceImportSpec{
+					Type:  mcsv1a1.ClusterSetIP,
+					Ports: t.aggregatedServicePorts,
+				},
+				Status: mcsv1a1.ServiceImportStatus{
+					Clusters: []mcsv1a1.ClusterStatus{{Cluster: t.cluster1.clusterID}},
+				},
+			}
+
+			awaitServiceImport(t.cluster1.localServiceImportClient, expServiceImport)
+
+			testutil.EnsureNoResource(resource.ForDynamic(t.cluster2.localServiceImportClient.Namespace(
+				t.cluster1.service.Namespace)), t.cluster1.service.Name)
+			t.cluster2.ensureNoEndpointSlice()
+
+			By("Creating namespace on importing cluster")
+
+			test.CreateResource(t.cluster2.localDynClient.Resource(corev1.SchemeGroupVersion.WithResource("namespaces")),
+				&corev1.Namespace{
+					ObjectMeta: metav1.ObjectMeta{Name: t.cluster1.service.Namespace},
+				})
+
+			t.awaitNonHeadlessServiceExported(&t.cluster1)
+		})
+	})
+
 	When("an existing ServiceExport has the legacy Synced status condition", func() {
 		BeforeEach(func() {
 			t.cluster1.serviceExport.Status.Conditions = []mcsv1a1.ServiceExportCondition{

diff --git a/pkg/agent/controller/controller_suite_test.go b/pkg/agent/controller/controller_suite_test.go
@@ -109,8 +109,6 @@ func init() {
 	if err != nil {
 		panic(err)
 	}
-
-	controller.BrokerResyncPeriod = time.Millisecond * 100
 }
 
 func TestController(t *testing.T) {
@@ -293,7 +291,7 @@ func newTestDiver() *testDriver {
 		syncerConfig: &broker.SyncerConfig{
 			BrokerNamespace: test.RemoteNamespace,
 			RestMapper: test.GetRESTMapperFor(&mcsv1a1.ServiceExport{}, &mcsv1a1.ServiceImport{}, &corev1.Service{},
-				&corev1.Endpoints{}, &discovery.EndpointSlice{}, controller.GetGlobalIngressIPObj()),
+				&corev1.Endpoints{}, &corev1.Namespace{}, &discovery.EndpointSlice{}, controller.GetGlobalIngressIPObj()),
 			BrokerClient: brokerClient,
 			Scheme:       syncerScheme,
 		},

diff --git a/pkg/agent/controller/service_import.go b/pkg/agent/controller/service_import.go
@@ -90,16 +90,17 @@ func newServiceImportController(spec *AgentSpecification, syncerMetricNames Agen
 	}
 
 	controller.remoteSyncer, err = syncer.NewResourceSyncer(&syncer.ResourceSyncerConfig{
-		Name:             "Remote ServiceImport",
-		SourceClient:     brokerClient,
-		SourceNamespace:  brokerNamespace,
-		RestMapper:       syncerConfig.RestMapper,
-		Federator:        federate.NewCreateOrUpdateFederator(syncerConfig.LocalClient, syncerConfig.RestMapper, corev1.NamespaceAll, ""),
-		ResourceType:     &mcsv1a1.ServiceImport{},
-		Transform:        controller.onRemoteServiceImport,
-		OnSuccessfulSync: controller.serviceImportMigrator.onSuccessfulSyncFromBroker,
-		Scheme:           syncerConfig.Scheme,
-		ResyncPeriod:     BrokerResyncPeriod,
+		Name:              "Remote ServiceImport",
+		SourceClient:      brokerClient,
+		SourceNamespace:   brokerNamespace,
+		RestMapper:        syncerConfig.RestMapper,
+		Federator:         federate.NewCreateOrUpdateFederator(syncerConfig.LocalClient, syncerConfig.RestMapper, corev1.NamespaceAll, ""),
+		ResourceType:      &mcsv1a1.ServiceImport{},
+		Transform:         controller.onRemoteServiceImport,
+		OnSuccessfulSync:  controller.serviceImportMigrator.onSuccessfulSyncFromBroker,
+		Scheme:            syncerConfig.Scheme,
+		ResyncPeriod:      BrokerResyncPeriod,
+		NamespaceInformer: syncerConfig.NamespaceInformer,
 		SyncCounterOpts: &prometheus.GaugeOpts{
 			Name: syncerMetricNames.ServiceImportCounterName,
 			Help: "Count of imported services",

diff --git a/pkg/agent/controller/types.go b/pkg/agent/controller/types.go
@@ -32,6 +32,7 @@ import (
 	k8slabels "k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/tools/cache"
 	mcsv1a1 "sigs.k8s.io/mcs-api/pkg/apis/v1alpha1"
 )
 
@@ -59,6 +60,7 @@ type Controller struct {
 	serviceSyncer               syncer.Interface
 	serviceImportController     *ServiceImportController
 	localServiceImportFederator federate.Federator
+	namespaceInformer           cache.SharedInformer
 }
 
 type AgentSpecification struct {