Virtual kubelet: working queue-based service reflection

Author: giorio94

cmd/virtual-kubelet/root/flag.go

@@ -27,6 +27,9 @@ func InstallFlags(flags *pflag.FlagSet, c *Opts) {
 	flags.StringVar(&c.MetricsAddr, "metrics-addr", c.MetricsAddr, "address to listen for metrics/stats requests")
 
 	flags.UintVar(&c.PodSyncWorkers, "pod-sync-workers", c.PodSyncWorkers, "the number of pod synchronization workers")
+	flags.UintVar(&c.ServiceWorkers, "service-reflection-workers", c.ServiceWorkers, "the number of service reflection workers")
+	flags.UintVar(&c.EndpointSliceWorkers, "endpointslice-reflection-workers", c.EndpointSliceWorkers,
+		"the number of endpointslice reflection workers")
 
 	flags.DurationVar(&c.InformerResyncPeriod, "full-resync-period", c.InformerResyncPeriod,
 		"how often to perform a full resync of pods between kubernetes and the provider")

cmd/virtual-kubelet/root/opts.go

@@ -32,7 +32,10 @@ const (
 	DefaultLiqoInformerResyncPeriod time.Duration = 0
 	DefaultMetricsAddr                            = ":10255"
 	DefaultListenPort                             = 10250
-	DefaultPodSyncWorkers                         = 10
+
+	DefaultPodSyncWorkers       = 10
+	DefaultServiceWorkers       = 3
+	DefaultEndpointSliceWorkers = 10
 
 	DefaultKubeletNamespace = "default"
 	DefaultLiqoIpamServer   = consts.NetworkManagerServiceName
@@ -54,8 +57,10 @@ type Opts struct {
 
 	MetricsAddr string
 
-	// Number of workers to use to handle pod notifications
-	PodSyncWorkers uint
+	// Number of workers to use to handle pod notifications and resource reflection
+	PodSyncWorkers       uint
+	ServiceWorkers       uint
+	EndpointSliceWorkers uint
 
 	InformerResyncPeriod     time.Duration
 	LiqoInformerResyncPeriod time.Duration
@@ -94,6 +99,14 @@ func SetDefaultOpts(c *Opts) error {
 		c.PodSyncWorkers = DefaultPodSyncWorkers
 	}
 
+	if c.ServiceWorkers == 0 {
+		c.ServiceWorkers = DefaultServiceWorkers
+	}
+
+	if c.EndpointSliceWorkers == 0 {
+		c.EndpointSliceWorkers = DefaultEndpointSliceWorkers
+	}
+
 	if c.ListenPort == 0 {
 		if kp := os.Getenv("KUBELET_PORT"); kp != "" {
 			p, err := strconv.ParseInt(kp, 10, 32)

cmd/virtual-kubelet/root/root.go

@@ -112,10 +112,14 @@ func runRootCommand(ctx context.Context, c *Opts) error {
 		HomeClusterID:   c.HomeClusterID,
 		RemoteClusterID: c.ForeignClusterID,
 
-		Namespace:            c.KubeletNamespace,
-		NodeName:             c.NodeName,
+		Namespace: c.KubeletNamespace,
+		NodeName:  c.NodeName,
+
 		LiqoIpamServer:       c.LiqoIpamServer,
 		InformerResyncPeriod: c.InformerResyncPeriod,
+
+		ServiceWorkers:       c.ServiceWorkers,
+		EndpointSliceWorkers: c.EndpointSliceWorkers,
 	}
 
 	podProvider, err := podprovider.NewLiqoProvider(ctx, &podcfg)

internal/crdReplicator/reflection/handler_test.go

@@ -174,7 +174,7 @@ var _ = Describe("Handler tests", func() {
 				})
 
 				It("should succeed", func() { Expect(err).ToNot(HaveOccurred()) })
-				It("should remote the finalizer from the local object", func() {
+				It("should remove the finalizer from the local object", func() {
 					Expect(localAfter.Finalizers).ToNot(ContainElement(finalizer))
 				})
 				It("the remote object should not be created", func() {

pkg/virtualKubelet/apiReflection/const.go

@@ -19,7 +19,6 @@ const (
 	EndpointSlices
 	Pods
 	ReplicaSets
-	Services
 	Secrets
 )
 
@@ -30,7 +29,6 @@ var ApiNames = map[ApiType]string{
 	EndpointSlices: "endpointslices",
 	Pods:           "pods",
 	ReplicaSets:    "replicasets",
-	Services:       "services",
 	Secrets:        "secrets",
 }
 

pkg/virtualKubelet/apiReflection/reflectors/blacklists.go

@@ -28,7 +28,4 @@ var Blacklist = map[apimgmt.ApiType]blackListType{
 		"default/kubernetes": struct{}{},
 	},
 	apimgmt.Pods: {},
-	apimgmt.Services: {
-		"default/kubernetes": struct{}{},
-	},
 }

pkg/virtualKubelet/apiReflection/reflectors/outgoing/apiTypes.go

@@ -32,7 +32,6 @@ var ReflectorBuilders = map[apimgmt.ApiType]func(reflector ri.APIReflector, opts
 	apimgmt.Configmaps:     configmapsReflectorBuilder,
 	apimgmt.EndpointSlices: endpointslicesReflectorBuilder,
 	apimgmt.Secrets:        secretsReflectorBuilder,
-	apimgmt.Services:       servicesReflectorBuilder,
 }
 
 func configmapsReflectorBuilder(reflector ri.APIReflector, _ map[options.OptionKey]options.Option) ri.OutgoingAPIReflector {
@@ -58,7 +57,3 @@ func endpointslicesReflectorBuilder(reflector ri.APIReflector, opts map[options.
 func secretsReflectorBuilder(reflector ri.APIReflector, _ map[options.OptionKey]options.Option) ri.OutgoingAPIReflector {
 	return &SecretsReflector{APIReflector: reflector}
 }
-
-func servicesReflectorBuilder(reflector ri.APIReflector, _ map[options.OptionKey]options.Option) ri.OutgoingAPIReflector {
-	return &ServicesReflector{APIReflector: reflector}
-}

pkg/virtualKubelet/apiReflection/reflectors/outgoing/services.go

@@ -23,6 +23,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	discoveryv1beta1 "k8s.io/api/discovery/v1beta1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/klog/v2"
 
@@ -33,6 +34,9 @@ import (
 	"github.com/liqotech/liqo/pkg/virtualKubelet/options/types"
 )
 
+// ReflectionFieldManager -> The name associated with the fields modified by virtual kubelet reflection.
+const ReflectionFieldManager = "reflection.liqo.io"
+
 var (
 	// LocalClusterID -> the cluster ID associated with the local cluster.
 	LocalClusterID string
@@ -66,8 +70,6 @@ func HomeToForeign(homeObj, foreignObj runtime.Object, reflectionType string) (r
 		return forger.endpointsliceHomeToForeign(homeObj.(*discoveryv1beta1.EndpointSlice), foreignObj.(*discoveryv1beta1.EndpointSlice))
 	case *corev1.Pod:
 		return forger.podHomeToForeign(homeObj, foreignObj, reflectionType)
-	case *corev1.Service:
-		return forger.serviceHomeToForeign(homeObj.(*corev1.Service), foreignObj.(*corev1.Service))
 	}
 
 	return nil, errors.Errorf("error while creating foreign object from home: api %s unhandled", reflect.TypeOf(homeObj).String())
@@ -114,7 +116,15 @@ func initIpamClient() {
 		grpc.WithInsecure(),
 		grpc.WithBlock())
 	if err != nil {
-		klog.Error(err)
+		klog.Fatalf("Failed to initialize IPAM client: %v", err)
 	}
 	forger.ipamClient = liqonetIpam.NewIpamClient(conn)
 }
+
+// ApplyOptions returns the apply options configured for object reflection.
+func ApplyOptions() metav1.ApplyOptions {
+	return metav1.ApplyOptions{
+		Force:        true,
+		FieldManager: ReflectionFieldManager,
+	}
+}

pkg/virtualKubelet/forge/apiForger.go

@@ -21,6 +21,11 @@ import (
 	. "github.com/onsi/gomega"
 )
 
+const (
+	LocalClusterID  = "local-cluster"
+	RemoteClusterID = "remote-cluster"
+)
+
 func TestForge(t *testing.T) {
 	RegisterFailHandler(Fail)
 	RunSpecs(t, "Forge Suite")