Add ctx param to GetLocalNode

main.go

@@ -132,7 +132,7 @@ func main() {
 		return
 	}
 
-	gw, err := gateway.New(&gateway.Config{
+	gw, err := gateway.New(ctx, &gateway.Config{
 		LeaderElectionConfig: gateway.LeaderElectionConfig{
 			LeaseDuration: time.Duration(gwLeadershipConfig.LeaseDuration) * time.Second,
 			RenewDeadline: time.Duration(gwLeadershipConfig.RenewDeadline) * time.Second,

pkg/endpoint/local_endpoint.go

@@ -133,13 +133,13 @@ func (l *Local) Create(ctx context.Context) error {
 	return err
 }
 
-func GetLocalSpec(submSpec *types.SubmarinerSpecification, k8sClient kubernetes.Interface,
+func GetLocalSpec(ctx context.Context, submSpec *types.SubmarinerSpecification, k8sClient kubernetes.Interface,
 	airGappedDeployment bool,
 ) (*submv1.EndpointSpec, error) {
 	// We'll panic if submSpec is nil, this is intentional
 	privateIP := GetLocalIP()
 
-	gwNode, err := node.GetLocalNode(k8sClient)
+	gwNode, err := node.GetLocalNode(ctx, k8sClient)
 	if err != nil {
 		return nil, errors.Wrap(err, "getting information on the local node")
 	}

pkg/endpoint/local_endpoint_test.go

@@ -96,7 +96,7 @@ var _ = Describe("GetLocalSpec", func() {
 	})
 
 	It("should return a valid EndpointSpec object", func() {
-		spec, err := endpoint.GetLocalSpec(submSpec, client, false)
+		spec, err := endpoint.GetLocalSpec(context.TODO(), submSpec, client, false)
 
 		Expect(err).ToNot(HaveOccurred())
 		Expect(spec.ClusterID).To(Equal("east"))
@@ -116,7 +116,7 @@ var _ = Describe("GetLocalSpec", func() {
 		})
 
 		It("should return the udp-port backend config of the cluster", func() {
-			spec, err := endpoint.GetLocalSpec(submSpec, client, false)
+			spec, err := endpoint.GetLocalSpec(context.TODO(), submSpec, client, false)
 			Expect(err).ToNot(HaveOccurred())
 			Expect(spec.BackendConfig[testUDPPortLabel]).To(Equal(testClusterUDPPort))
 		})
@@ -128,14 +128,14 @@ var _ = Describe("GetLocalSpec", func() {
 		})
 
 		It("should return a valid EndpointSpec object", func() {
-			_, err := endpoint.GetLocalSpec(submSpec, client, false)
+			_, err := endpoint.GetLocalSpec(context.TODO(), submSpec, client, false)
 			Expect(err).ToNot(HaveOccurred())
 		})
 	})
 
 	When("the gateway node is not annotated with public-ip", func() {
 		It("should use empty public-ip in the endpoint object for air-gapped deployments", func() {
-			spec, err := endpoint.GetLocalSpec(submSpec, client, true)
+			spec, err := endpoint.GetLocalSpec(context.TODO(), submSpec, client, true)
 
 			Expect(err).ToNot(HaveOccurred())
 			Expect(spec.ClusterID).To(Equal("east"))
@@ -150,7 +150,7 @@ var _ = Describe("GetLocalSpec", func() {
 		})
 
 		It("should use the annotated public-ip for air-gapped deployments", func() {
-			spec, err := endpoint.GetLocalSpec(submSpec, client, true)
+			spec, err := endpoint.GetLocalSpec(context.TODO(), submSpec, client, true)
 
 			Expect(err).ToNot(HaveOccurred())
 			Expect(spec.PrivateIP).To(Equal(testPrivateIP))
@@ -164,7 +164,7 @@ var _ = Describe("GetLocalSpec", func() {
 		})
 
 		It("should set the HealthCheckIP", func() {
-			spec, err := endpoint.GetLocalSpec(submSpec, client, true)
+			spec, err := endpoint.GetLocalSpec(context.TODO(), submSpec, client, true)
 			Expect(err).ToNot(HaveOccurred())
 			Expect(spec.HealthCheckIP).To(Equal(cniInterfaceIP))
 		})
@@ -175,7 +175,7 @@ var _ = Describe("GetLocalSpec", func() {
 			})
 
 			It("should not set the HealthCheckIP", func() {
-				spec, err := endpoint.GetLocalSpec(submSpec, client, true)
+				spec, err := endpoint.GetLocalSpec(context.TODO(), submSpec, client, true)
 				Expect(err).ToNot(HaveOccurred())
 				Expect(spec.HealthCheckIP).To(BeEmpty())
 			})

pkg/event/handler.go

@@ -19,6 +19,7 @@ limitations under the License.
 package event
 
 import (
+	"context"
 	"sync/atomic"
 
 	submV1 "github.com/submariner-io/submariner/pkg/apis/submariner.io/v1"
@@ -81,7 +82,7 @@ type NodeHandler interface {
 
 type Handler interface {
 	// Init is called once on startup to let the handler initialize any state it needs.
-	Init() error
+	Init(ctx context.Context) error
 
 	// SetHandlerState is called once on startup after Init with the HandlerState that can be used to access global data from event callbacks.
 	SetState(handlerCtx HandlerState)
@@ -106,7 +107,7 @@ type HandlerBase struct {
 	handlerState atomic.Value
 }
 
-func (ev *HandlerBase) Init() error {
+func (ev *HandlerBase) Init(_ context.Context) error {
 	return nil
 }
 

pkg/event/registry.go

@@ -19,6 +19,7 @@ limitations under the License.
 package event
 
 import (
+	"context"
 	"strings"
 
 	"github.com/pkg/errors"
@@ -39,15 +40,15 @@ var logger = log.Logger{Logger: logf.Log.WithName("EventRegistry")}
 // NewRegistry creates a new registry with the given name, typically referencing the owner, to manage event
 // Handlers that match the given networkPlugin name. The given event Handlers whose associated network plugin matches the given
 // networkPlugin name are added. Non-matching Handlers are ignored. Handlers will be called in registration order.
-func NewRegistry(name, networkPlugin string, eventHandlers ...Handler) (*Registry, error) {
+func NewRegistry(ctx context.Context, name, networkPlugin string, eventHandlers ...Handler) (*Registry, error) {
 	r := &Registry{
 		name:          name,
 		networkPlugin: strings.ToLower(networkPlugin),
 		eventHandlers: []Handler{},
 	}
 
 	for _, eventHandler := range eventHandlers {
-		err := r.addHandler(eventHandler)
+		err := r.addHandler(ctx, eventHandler)
 		if err != nil {
 			return nil, err
 		}
@@ -61,15 +62,15 @@ func (er *Registry) GetName() string {
 	return er.name
 }
 
-func (er *Registry) addHandler(eventHandler Handler) error {
+func (er *Registry) addHandler(ctx context.Context, eventHandler Handler) error {
 	evNetworkPlugins := set.New[string]()
 
 	for _, np := range eventHandler.GetNetworkPlugins() {
 		evNetworkPlugins.Insert(strings.ToLower(np))
 	}
 
 	if evNetworkPlugins.Has(AnyNetworkPlugin) || evNetworkPlugins.Has(er.networkPlugin) {
-		if err := eventHandler.Init(); err != nil {
+		if err := eventHandler.Init(ctx); err != nil {
 			return errors.Wrapf(err, "Event handler %q failed to initialize", eventHandler.GetName())
 		}
 

pkg/event/registry_test.go

@@ -19,6 +19,8 @@ limitations under the License.
 package event_test
 
 import (
+	"context"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	submv1 "github.com/submariner-io/submariner/pkg/apis/submariner.io/v1"
@@ -55,8 +57,8 @@ var _ = Describe("Event Registry", func() {
 
 			var err error
 
-			registry, err = event.NewRegistry("test-registry", npGenericKubeproxyIptables, logger.NewHandler(), matchingHandlers[0],
-				nonMatchingHandlers[0], matchingHandlers[1], matchingHandlers[2])
+			registry, err = event.NewRegistry(context.TODO(), "test-registry", npGenericKubeproxyIptables, logger.NewHandler(),
+				matchingHandlers[0], nonMatchingHandlers[0], matchingHandlers[1], matchingHandlers[2])
 			Expect(err).NotTo(HaveOccurred())
 			Expect(registry.GetName()).To(Equal("test-registry"))
 		})
@@ -115,7 +117,7 @@ var _ = Describe("Event Registry", func() {
 			h := testing.NewTestHandler("test-handler", event.AnyNetworkPlugin, make(chan testing.TestEvent, 10))
 			h.FailOnEvent(testing.EvInit)
 
-			_, err := event.NewRegistry("test-registry", event.AnyNetworkPlugin, h)
+			_, err := event.NewRegistry(context.TODO(), "test-registry", event.AnyNetworkPlugin, h)
 			Expect(err).To(HaveOccurred())
 		})
 	})

pkg/event/testing/controller_support.go

@@ -67,7 +67,7 @@ func (c *ControllerSupport) Start(handlers ...event.Handler) {
 		networkPlugin = event.AnyNetworkPlugin
 	}
 
-	registry, err := event.NewRegistry("test-registry", networkPlugin, handlers...)
+	registry, err := event.NewRegistry(context.TODO(), "test-registry", networkPlugin, handlers...)
 	Expect(err).To(Succeed())
 
 	config := controller.Config{

pkg/event/testing/testing.go

@@ -19,6 +19,7 @@ limitations under the License.
 package testing
 
 import (
+	"context"
 	"fmt"
 	"sync"
 
@@ -110,7 +111,7 @@ func (t *TestHandlerBase) addEvent(eventName string, param interface{}) error {
 	return nil
 }
 
-func (t *TestHandlerBase) Init() error {
+func (t *TestHandlerBase) Init(_ context.Context) error {
 	t.Initialized = true
 	return t.checkFailOnEvent(EvInit)
 }

pkg/gateway/gateway.go

@@ -104,7 +104,7 @@ type gatewayType struct {
 
 var logger = log.Logger{Logger: logf.Log.WithName("Gateway")}
 
-func New(config *Config) (Interface, error) {
+func New(ctx context.Context, config *Config) (Interface, error) {
 	logger.Info("Initializing the gateway engine")
 
 	g := &gatewayType{
@@ -144,7 +144,7 @@ func New(config *Config) (Interface, error) {
 	g.airGapped = os.Getenv("AIR_GAPPED_DEPLOYMENT") == "true"
 	logger.Infof("AIR_GAPPED_DEPLOYMENT is set to %t", g.airGapped)
 
-	localEndpointSpec, err := endpoint.GetLocalSpec(&g.Spec, g.KubeClient, g.airGapped)
+	localEndpointSpec, err := endpoint.GetLocalSpec(ctx, &g.Spec, g.KubeClient, g.airGapped)
 	if err != nil {
 		return nil, errors.Wrap(err, "error creating local endpoint object")
 	}

pkg/gateway/gateway_test.go

@@ -321,7 +321,7 @@ func newTestDriver() *testDriver {
 	})
 
 	JustBeforeEach(func() {
-		gw, err := gateway.New(&t.config)
+		gw, err := gateway.New(context.TODO(), &t.config)
 		Expect(err).To(Succeed())
 
 		ctx, stop := context.WithCancel(context.Background())

pkg/globalnet/controllers/gateway_monitor.go

@@ -53,7 +53,7 @@ import (
 	"k8s.io/utils/set"
 )
 
-func NewGatewayMonitor(config *GatewayMonitorConfig) (Interface, error) {
+func NewGatewayMonitor(ctx context.Context, config *GatewayMonitorConfig) (Interface, error) {
 	// We'll panic if config is nil, this is intentional
 	gatewayMonitor := &gatewayMonitor{
 		baseController:          newBaseController(),
@@ -130,7 +130,12 @@ func NewGatewayMonitor(config *GatewayMonitorConfig) (Interface, error) {
 
 	gatewayMonitor.gatewaySharedInformerStopCh = make(chan struct{})
 
-	return &gatewayMonitorInterface{monitor: gatewayMonitor}, nil
+	registry, err := event.NewRegistry(ctx, "globalnet-registry", event.AnyNetworkPlugin, gatewayMonitor)
+	if err != nil {
+		return nil, errors.Wrap(err, "error creating event registry")
+	}
+
+	return &gatewayMonitorInterface{monitor: gatewayMonitor, registry: registry}, nil
 }
 
 func (g *gatewayMonitorInterface) Start() error {
@@ -145,13 +150,8 @@ func (g *gatewayMonitorInterface) Start() error {
 		g.monitor.gatewaySharedInformer.Run(g.monitor.gatewaySharedInformerStopCh)
 	}()
 
-	registry, err := event.NewRegistry("globalnet-registry", event.AnyNetworkPlugin, g.monitor)
-	if err != nil {
-		return errors.Wrap(err, "error creating event registry")
-	}
-
 	eventController, err := controller.New(&controller.Config{
-		Registry:   registry,
+		Registry:   g.registry,
 		RestMapper: g.monitor.RestMapper,
 		Client:     g.monitor.Client,
 		Scheme:     g.monitor.Scheme,
@@ -175,7 +175,7 @@ func (g *gatewayMonitor) GetNetworkPlugins() []string {
 	return []string{event.AnyNetworkPlugin}
 }
 
-func (g *gatewayMonitor) Init() error {
+func (g *gatewayMonitor) Init(_ context.Context) error {
 	return g.createNATChain(constants.SmGlobalnetMarkChain)
 }
 

pkg/globalnet/controllers/gateway_monitor_test.go

@@ -416,7 +416,7 @@ func (t *gatewayMonitorTestDriver) start() {
 
 	localSubnets := []string{localCIDR}
 
-	t.controller, err = controllers.NewGatewayMonitor(&controllers.GatewayMonitorConfig{
+	t.controller, err = controllers.NewGatewayMonitor(context.TODO(), &controllers.GatewayMonitorConfig{
 		RestMapper: t.restMapper,
 		Client:     t.dynClient,
 		Scheme:     t.scheme,

pkg/globalnet/controllers/types.go

@@ -120,7 +120,8 @@ type LeaderElectionInfo struct {
 }
 
 type gatewayMonitorInterface struct {
-	monitor *gatewayMonitor
+	monitor  *gatewayMonitor
+	registry *event.Registry
 }
 
 type gatewayMonitor struct {

pkg/globalnet/main.go

@@ -101,7 +101,7 @@ func main() {
 	logger.Info("Starting submariner-globalnet", spec)
 
 	// set up signals so we handle the first shutdown signal gracefully
-	stopCh := signals.SetupSignalHandler().Done()
+	ctx := signals.SetupSignalHandler()
 
 	defer http.StartServer(http.Metrics|http.Profile, spec.MetricsPort)()
 
@@ -139,7 +139,7 @@ func main() {
 
 	clusterCIDRs := cidr.ExtractIPv4Subnets(localCluster.Spec.ClusterCIDR)
 
-	gatewayMonitor, err := controllers.NewGatewayMonitor(&controllers.GatewayMonitorConfig{
+	gatewayMonitor, err := controllers.NewGatewayMonitor(ctx, &controllers.GatewayMonitorConfig{
 		Client:            dynClient,
 		RestMapper:        restMapper,
 		Scheme:            scheme.Scheme,
@@ -154,7 +154,7 @@ func main() {
 	err = gatewayMonitor.Start()
 	logger.FatalOnError(err, "Error starting the gatewayMonitor")
 
-	<-stopCh
+	<-ctx.Done()
 	gatewayMonitor.Stop()
 
 	logger.Infof("All controllers stopped or exited. Stopping main loop")