koordlet: support extension controllers (#2459)

Signed-off-by: shaloulcy <[email protected]>
Co-authored-by: shaloulcy <[email protected]>
Co-authored-by: wangjianyu <[email protected]>

Author: 3 people

cmd/koordlet/options/options.go

@@ -40,3 +40,7 @@ func InstallExtendedHTTPHandler(mux *http.ServeMux) {
 		klog.V(4).Infof("extended HTTP handler is registered on path %s", path)
 	}
 }
+
+func RegisterHTTPHandler(path string, handler func() http.HandlerFunc) {
+	ExtendedHTTPHandlerRegistry[path] = handler
+}

pkg/koordlet/extension/interface.go

@@ -0,0 +1,30 @@
+/*
+Copyright 2022 The Koordinator Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package extension
+
+import (
+	clientset "k8s.io/client-go/kubernetes"
+
+	"github.com/koordinator-sh/koordinator/pkg/koordlet/statesinformer"
+)
+
+type Controller interface {
+	Run(stopCh <-chan struct{}) error
+	Name() string
+}
+
+type ControllerInitFunc = func(nodeName string, kubeClient clientset.Interface, statesInformer statesinformer.StatesInformer) Controller

pkg/koordlet/koordlet.go

@@ -32,6 +32,7 @@ import (
 	clientsetbeta1 "github.com/koordinator-sh/koordinator/pkg/client/clientset/versioned"
 	"github.com/koordinator-sh/koordinator/pkg/client/clientset/versioned/typed/scheduling/v1alpha1"
 	"github.com/koordinator-sh/koordinator/pkg/koordlet/config"
+	"github.com/koordinator-sh/koordinator/pkg/koordlet/extension"
 	"github.com/koordinator-sh/koordinator/pkg/koordlet/metriccache"
 	"github.com/koordinator-sh/koordinator/pkg/koordlet/metrics"
 	"github.com/koordinator-sh/koordinator/pkg/koordlet/metricsadvisor"
@@ -47,6 +48,8 @@ import (
 
 var (
 	scheme = apiruntime.NewScheme()
+
+	extensionControllerInitFuncs = map[string]extension.ControllerInitFunc{}
 )
 
 func init() {
@@ -65,6 +68,8 @@ type daemon struct {
 	runtimeHook    runtimehooks.RuntimeHook
 	predictServer  prediction.PredictServer
 	executor       resourceexecutor.ResourceUpdateExecutor
+
+	extensionControllers []extension.Controller
 }
 
 func NewDaemon(config *config.Configuration) (Daemon, error) {
@@ -111,6 +116,11 @@ func NewDaemon(config *config.Configuration) (Daemon, error) {
 		return nil, err
 	}
 
+	extensionControllers := []extension.Controller{}
+	for _, initFunc := range extensionControllerInitFuncs {
+		extensionControllers = append(extensionControllers, initFunc(nodeName, kubeClient, statesInformer))
+	}
+
 	d := &daemon{
 		metricAdvisor:  collectorService,
 		statesInformer: statesInformer,
@@ -119,6 +129,8 @@ func NewDaemon(config *config.Configuration) (Daemon, error) {
 		runtimeHook:    runtimeHook,
 		predictServer:  predictServer,
 		executor:       resourceexecutor.NewResourceUpdateExecutor(),
+
+		extensionControllers: extensionControllers,
 	}
 
 	return d, nil
@@ -182,6 +194,16 @@ func (d *daemon) Run(stopCh <-chan struct{}) {
 		}
 	}()
 
+	for _, c := range d.extensionControllers {
+		go func(controller extension.Controller) {
+			name := controller.Name()
+			klog.Infof("starting extension controller %v", name)
+			if err := controller.Run(stopCh); err != nil {
+				klog.Fatalf("Unable to start the extension controller %v, err: %v", name, err)
+			}
+		}(c)
+	}
+
 	klog.Info("Start daemon successfully")
 	<-stopCh
 	klog.Info("Shutting down daemon")