service-cache: add support for service mutators

Extend the service cache to specify additional mutators to be executed
when a service is received from k8s and converted to the internal
representation.

Signed-off-by: Marco Iorio <marco.iorio@isovalent.com>

pkg/k8s/service_cache.go

@@ -90,6 +90,8 @@ type ServiceCache struct {
 	nodeAddressing types.NodeAddressing
 
 	selfNodeZoneLabel string
+
+	ServiceMutators []func(svc *slim_corev1.Service, svcInfo *Service)
 }
 
 // NewServiceCache returns a new ServiceCache
@@ -192,6 +194,10 @@ func (s *ServiceCache) UpdateService(k8sSvc *slim_corev1.Service, swg *lock.Stop
 		return svcID
 	}
 
+	for _, mutator := range s.ServiceMutators {
+		mutator(k8sSvc, newService)
+	}
+
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 

pkg/k8s/service_cache_test.go

@@ -328,6 +328,29 @@ func (s *K8sSuite) TestCacheActionString(c *check.C) {
 	c.Assert(DeleteService.String(), check.Equals, "service-deleted")
 }
 
+func (s *K8sSuite) TestServiceMutators(c *check.C) {
+	var m1, m2 int
+
+	svcCache := NewServiceCache(fakeDatapath.NewNodeAddressing())
+	svcCache.ServiceMutators = append(svcCache.ServiceMutators,
+		func(svc *slim_corev1.Service, svcInfo *Service) { m1++ },
+		func(svc *slim_corev1.Service, svcInfo *Service) { m2++ },
+	)
+	swg := lock.NewStoppableWaitGroup()
+	svcCache.UpdateService(&slim_corev1.Service{
+		ObjectMeta: slim_metav1.ObjectMeta{Name: "foo", Namespace: "bar"},
+		Spec: slim_corev1.ServiceSpec{
+			ClusterIP: "127.0.0.1",
+			Selector:  map[string]string{"foo": "bar"},
+			Type:      slim_corev1.ServiceTypeClusterIP,
+		},
+	}, swg)
+
+	// Assert that the service mutators configured have been executed.
+	c.Assert(m1, check.Equals, 1)
+	c.Assert(m2, check.Equals, 1)
+}
+
 func (s *K8sSuite) TestExternalServiceMerging(c *check.C) {
 	svcCache := NewServiceCache(fakeDatapath.NewNodeAddressing())