Virtual kubelet: reflection-based pod offloading

Author: giorio94

cmd/virtual-kubelet/root/flag.go

@@ -26,7 +26,7 @@ func InstallFlags(flags *pflag.FlagSet, c *Opts) {
 	flags.StringVar(&c.NodeName, "nodename", c.NodeName, "kubernetes node name")
 	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.PodWorkers, "pod-reflection-workers", c.PodWorkers, "the number of pod reflection 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")

cmd/virtual-kubelet/root/http.go

@@ -28,7 +28,7 @@ import (
 	"github.com/virtual-kubelet/virtual-kubelet/node/api"
 	"k8s.io/klog/v2"
 
-	podprovider "github.com/liqotech/liqo/pkg/virtualKubelet/provider"
+	"github.com/liqotech/liqo/pkg/virtualKubelet/reflection/workload"
 )
 
 // AcceptedCiphers is the list of accepted TLS ciphers, with known weak ciphers elided
@@ -59,8 +59,7 @@ func loadTLSConfig(certPath, keyPath string) (*tls.Config, error) {
 	}, nil
 }
 
-func setupHTTPServer(ctx context.Context, p *podprovider.LiqoProvider, cfg *apiServerConfig,
-	getPodsFromKubernetes api.PodListerFunc) (_ func(), retErr error) {
+func setupHTTPServer(ctx context.Context, handler workload.PodHandler, cfg *apiServerConfig) (_ func(), retErr error) {
 	var closers []io.Closer
 	cancel := func() {
 		for _, c := range closers {
@@ -76,7 +75,7 @@ func setupHTTPServer(ctx context.Context, p *podprovider.LiqoProvider, cfg *apiS
 	if cfg.CertPath == "" || cfg.KeyPath == "" {
 		klog.Error("TLS certificates not provided, not setting up pod http server")
 	} else {
-		s, err := startPodHandlerServer(ctx, p, cfg, getPodsFromKubernetes)
+		s, err := startPodHandlerServer(ctx, handler, cfg)
 		if err != nil {
 			return nil, err
 		}
@@ -96,7 +95,7 @@ func setupHTTPServer(ctx context.Context, p *podprovider.LiqoProvider, cfg *apiS
 		mux := http.NewServeMux()
 
 		podMetricsRoutes := api.PodMetricsConfig{
-			GetStatsSummary: p.GetStatsSummary,
+			GetStatsSummary: handler.Stats,
 		}
 		api.AttachPodMetricsRoutes(podMetricsRoutes, mux)
 		s := &http.Server{
@@ -109,8 +108,7 @@ func setupHTTPServer(ctx context.Context, p *podprovider.LiqoProvider, cfg *apiS
 	return cancel, nil
 }
 
-func startPodHandlerServer(ctx context.Context, p *podprovider.LiqoProvider,
-	cfg *apiServerConfig, getPodsFromKubernetes api.PodListerFunc) (*http.Server, error) {
+func startPodHandlerServer(ctx context.Context, handler workload.PodHandler, cfg *apiServerConfig) (*http.Server, error) {
 	tlsCfg, err := loadTLSConfig(cfg.CertPath, cfg.KeyPath)
 	if err != nil {
 		klog.Error(err)
@@ -126,11 +124,11 @@ func startPodHandlerServer(ctx context.Context, p *podprovider.LiqoProvider,
 	mux := http.NewServeMux()
 
 	podRoutes := api.PodHandlerConfig{
-		RunInContainer:        p.RunInContainer,
-		GetContainerLogs:      p.GetContainerLogs,
-		GetPodsFromKubernetes: getPodsFromKubernetes,
-		GetStatsSummary:       p.GetStatsSummary,
-		GetPods:               p.GetPods,
+		RunInContainer:        handler.Exec,
+		GetContainerLogs:      handler.Logs,
+		GetStatsSummary:       handler.Stats,
+		GetPodsFromKubernetes: handler.List,
+		GetPods:               handler.List,
 	}
 
 	api.AttachPodRoutes(podRoutes, mux, true)

cmd/virtual-kubelet/root/opts.go

@@ -15,6 +15,7 @@
 package root
 
 import (
+	"fmt"
 	"os"
 	"strconv"
 	"time"
@@ -33,7 +34,7 @@ const (
 	DefaultMetricsAddr                            = ":10255"
 	DefaultListenPort                             = 10250
 
-	DefaultPodSyncWorkers       = 10
+	DefaultPodWorkers           = 10
 	DefaultServiceWorkers       = 3
 	DefaultEndpointSliceWorkers = 10
 	DefaultConfigMapWorkers     = 3
@@ -59,8 +60,8 @@ type Opts struct {
 
 	MetricsAddr string
 
-	// Number of workers to use to handle pod notifications and resource reflection
-	PodSyncWorkers       uint
+	// Number of workers to use to handle pod and resource reflection
+	PodWorkers           uint
 	ServiceWorkers       uint
 	EndpointSliceWorkers uint
 	ConfigMapWorkers     uint
@@ -99,8 +100,8 @@ func SetDefaultOpts(c *Opts) error {
 		c.MetricsAddr = DefaultMetricsAddr
 	}
 
-	if c.PodSyncWorkers == 0 {
-		c.PodSyncWorkers = DefaultPodSyncWorkers
+	if c.PodWorkers == 0 {
+		c.PodWorkers = DefaultPodWorkers
 	}
 
 	if c.ServiceWorkers == 0 {
@@ -138,7 +139,7 @@ func SetDefaultOpts(c *Opts) error {
 		c.HomeKubeconfig = os.Getenv("KUBECONFIG")
 	}
 	if c.LiqoIpamServer == "" {
-		c.LiqoIpamServer = DefaultLiqoIpamServer
+		c.LiqoIpamServer = fmt.Sprintf("%v:%v", consts.NetworkManagerServiceName, consts.NetworkManagerIpamPort)
 	}
 
 	return nil

cmd/virtual-kubelet/root/root.go

@@ -18,31 +18,23 @@ package root
 import (
 	"context"
 	"os"
-	"path"
-	"time"
 
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
 	"github.com/virtual-kubelet/virtual-kubelet/node"
 	corev1 "k8s.io/api/core/v1"
 	k8serrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/fields"
 	"k8s.io/client-go/discovery"
-	kubeinformers "k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/record"
-	"k8s.io/client-go/util/workqueue"
 	"k8s.io/klog/v2"
 
 	"github.com/liqotech/liqo/internal/utils/errdefs"
 	"github.com/liqotech/liqo/pkg/utils"
 	"github.com/liqotech/liqo/pkg/utils/restcfg"
 	"github.com/liqotech/liqo/pkg/virtualKubelet"
 	nodeprovider "github.com/liqotech/liqo/pkg/virtualKubelet/liqoNodeProvider"
-	"github.com/liqotech/liqo/pkg/virtualKubelet/manager"
 	podprovider "github.com/liqotech/liqo/pkg/virtualKubelet/provider"
 )
 
@@ -70,8 +62,8 @@ func runRootCommand(ctx context.Context, c *Opts) error {
 		return errors.New("cluster id is mandatory")
 	}
 
-	if c.PodSyncWorkers == 0 {
-		return errdefs.InvalidInput("pod sync workers must be greater than 0")
+	if c.PodWorkers == 0 || c.ServiceWorkers == 0 || c.EndpointSliceWorkers == 0 || c.ConfigMapWorkers == 0 || c.SecretWorkers == 0 {
+		return errdefs.InvalidInput("reflection workers must be greater than 0")
 	}
 
 	config, err := utils.GetRestConfig(c.HomeKubeconfig)
@@ -85,27 +77,6 @@ func runRootCommand(ctx context.Context, c *Opts) error {
 		return err
 	}
 
-	// Create a shared informer factory for Kubernetes pods in the current namespace (if specified) and scheduled to the current node.
-	podInformerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(
-		client,
-		c.InformerResyncPeriod,
-		kubeinformers.WithTweakListOptions(func(options *metav1.ListOptions) {
-			options.FieldSelector = fields.OneTermEqualSelector("spec.nodeName", c.NodeName).String()
-		}))
-	podInformer := podInformerFactory.Core().V1().Pods()
-
-	// Create another shared informer factory for Kubernetes secrets and configmaps (not subject to any selectors).
-	scmInformerFactory := kubeinformers.NewSharedInformerFactoryWithOptions(client, c.InformerResyncPeriod)
-	// Create a secret informer and a config map informer so we can pass their listers to the resource manager.
-	secretInformer := scmInformerFactory.Core().V1().Secrets()
-	configMapInformer := scmInformerFactory.Core().V1().ConfigMaps()
-	serviceInformer := scmInformerFactory.Core().V1().Services()
-
-	rm, err := manager.NewResourceManager(podInformer.Lister(), secretInformer.Lister(), configMapInformer.Lister(), serviceInformer.Lister())
-	if err != nil {
-		return errors.Wrap(err, "could not create resource manager")
-	}
-
 	// Initialize the pod provider
 	podcfg := podprovider.InitConfig{
 		HomeConfig:      config,
@@ -114,10 +85,12 @@ func runRootCommand(ctx context.Context, c *Opts) error {
 
 		Namespace: c.KubeletNamespace,
 		NodeName:  c.NodeName,
+		NodeIP:    os.Getenv("VKUBELET_POD_IP"),
 
 		LiqoIpamServer:       c.LiqoIpamServer,
 		InformerResyncPeriod: c.InformerResyncPeriod,
 
+		PodWorkers:           c.PodWorkers,
 		ServiceWorkers:       c.ServiceWorkers,
 		EndpointSliceWorkers: c.EndpointSliceWorkers,
 		ConfigMapWorkers:     c.ConfigMapWorkers,
@@ -195,55 +168,12 @@ func runRootCommand(ctx context.Context, c *Opts) error {
 		return err
 	}
 
-	eb := record.NewBroadcaster()
-	pc, err := node.NewPodController(node.PodControllerConfig{
-		PodClient:                            client.CoreV1(),
-		PodInformer:                          podInformer,
-		EventRecorder:                        eb.NewRecorder(scheme.Scheme, corev1.EventSource{Component: path.Join(c.NodeName, "pod-controller")}),
-		Provider:                             podProvider,
-		SecretInformer:                       secretInformer,
-		ConfigMapInformer:                    configMapInformer,
-		ServiceInformer:                      serviceInformer,
-		SyncPodsFromKubernetesRateLimiter:    newPodControllerWorkqueueRateLimiter(),
-		SyncPodStatusFromProviderRateLimiter: newPodControllerWorkqueueRateLimiter(),
-		DeletePodsFromKubernetesRateLimiter:  newPodControllerWorkqueueRateLimiter(),
-	})
-	if err != nil {
-		return errors.Wrap(err, "error setting up pod controller")
-	}
-
-	go podInformerFactory.Start(ctx.Done())
-	go scmInformerFactory.Start(ctx.Done())
-
-	cancelHTTP, err := setupHTTPServer(ctx, podProvider, getAPIConfig(c), func(context.Context) ([]*corev1.Pod, error) {
-		return rm.GetPods(), nil
-	})
+	cancelHTTP, err := setupHTTPServer(ctx, podProvider.PodHandler(), getAPIConfig(c))
 	if err != nil {
 		return errors.Wrap(err, "error while setting up HTTP server")
 	}
 	defer cancelHTTP()
 
-	go func() {
-		if err := pc.Run(ctx, int(c.PodSyncWorkers)); err != nil && errors.Is(err, context.Canceled) {
-			klog.Fatal(errors.Wrap(err, "error in pod controller running"))
-		}
-	}()
-
-	if c.StartupTimeout > 0 {
-		ctx, cancel := context.WithTimeout(ctx, c.StartupTimeout)
-		klog.Info("Waiting for pod controller / VK to be ready")
-		select {
-		case <-ctx.Done():
-			cancel()
-			return ctx.Err()
-		case <-pc.Ready():
-		}
-		cancel()
-		if err := pc.Err(); err != nil {
-			return err
-		}
-	}
-
 	go func() {
 		if err := nodeRunner.Run(ctx); err != nil {
 			klog.Error(err, "error in pod controller running")
@@ -259,14 +189,6 @@ func runRootCommand(ctx context.Context, c *Opts) error {
 	return nil
 }
 
-// newPodControllerWorkqueueRateLimiter returns a new custom rate limiter to be assigned to the pod controller workqueues.
-// Differently from the standard workqueue.DefaultControllerRateLimiter(), composed of an overall bucket rate limiter
-// and a per-item exponential rate limiter to address failures, this includes only the latter component. Hance avoiding
-// performance limitations when processing a high number of pods in parallel.
-func newPodControllerWorkqueueRateLimiter() workqueue.RateLimiter {
-	return workqueue.NewItemExponentialFailureRateLimiter(5*time.Millisecond, 1000*time.Second)
-}
-
 func getVersion(config *rest.Config) string {
 	client, err := discovery.NewDiscoveryClientForConfig(config)
 	if err != nil {

deployments/liqo/files/liqo-virtual-kubelet-remote-ClusterRole.yaml

@@ -3,7 +3,6 @@ rules:
   - ""
   resources:
   - configmaps
-  - pods
   - secrets
   - services
   verbs:
@@ -17,44 +16,40 @@ rules:
 - apiGroups:
   - ""
   resources:
-  - pods/status
-  - services/status
+  - pods
   verbs:
-  - create
-  - delete
   - get
   - list
-  - patch
-  - update
   - watch
 - apiGroups:
-  - apps
+  - discovery.k8s.io
   resources:
-  - replicasets
+  - endpointslices
   verbs:
   - create
   - delete
   - get
   - list
+  - patch
   - update
   - watch
 - apiGroups:
-  - discovery.k8s.io
+  - metrics.k8s.io
   resources:
-  - endpointslices
+  - pods
   verbs:
-  - create
-  - delete
   - get
   - list
-  - patch
-  - update
   - watch
 - apiGroups:
-  - metrics.k8s.io
+  - virtualkubelet.liqo.io
   resources:
-  - pods
+  - shadowpods
   verbs:
+  - create
+  - delete
   - get
   - list
+  - patch
+  - update
   - watch

go.mod

@@ -18,7 +18,6 @@ require (
 	github.com/metal-stack/go-ipam v1.8.4-0.20210322080203-5a9da5064b27
 	github.com/miekg/dns v1.1.35
 	github.com/mittwald/go-helm-client v0.8.3-0.20211026133933-a26889186afc
-	github.com/modern-go/reflect2 v1.0.1
 	github.com/onsi/ginkgo v1.16.4
 	github.com/onsi/gomega v1.16.0
 	github.com/openshift/api v0.0.0-20210521075222-e273a339932a
@@ -155,6 +154,7 @@ require (
 	github.com/moby/sys/mountinfo v0.4.1 // indirect
 	github.com/moby/term v0.0.0-20210610120745-9d4ed1856297 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.1 // indirect
 	github.com/monochromegane/go-gitignore v0.0.0-20200626010858-205db1a8cc00 // indirect
 	github.com/morikuni/aec v1.0.0 // indirect
 	github.com/nxadm/tail v1.4.8 // indirect

pkg/liqonet/ipam/ipam.go

@@ -1096,7 +1096,7 @@ func (liqoIPAM *IPAM) GetHomePodIP(ctx context.Context, request *GetHomePodIPReq
 	homeIP, err := liqoIPAM.getHomePodIPInternal(request.GetClusterID(), request.GetIp())
 	if err != nil {
 		return &GetHomePodIPResponse{}, fmt.Errorf("cannot get home Pod IP starting from IP %s: %w",
-			request.GetClusterID(), err)
+			request.GetIp(), err)
 	}
 	return &GetHomePodIPResponse{HomeIP: homeIP}, nil
 }