[controller] Add cache for extender-scheduler (#23)

Signed-off-by: Viktor Kramarenko <viktor.kramarenko@flant.com>

images/sds-local-volume-controller/cmd/main.go

@@ -118,6 +118,4 @@ func main() {
 		log.Error(err, "[main] unable to mgr.Start")
 		os.Exit(1)
 	}
-
-	log.Info("[main] successfully starts the manager")
 }

images/sds-local-volume-scheduler-extender/Dockerfile

@@ -1,7 +1,7 @@
-ARG BASE_GOLANG_20_ALPINE=registry.deckhouse.io/base_images/golang:1.20.4-alpine3.18@sha256:5f403dd08db2f0b40d4416e29d3080eec41cd6cf53a05d5e4bcece3a5c7a8ce6
-ARG BASE_GOLANG_20_ALPINE_BUILDER=$BASE_GOLANG_20_ALPINE
+ARG BASE_GOLANG_22_ALPINE=registry.deckhouse.io/base_images/golang:1.22.1-alpine@sha256:0de6cf7cceab6ecbf0718bdfb675b08b78113c3709c5e4b99456cdb2ae8c2495
+ARG BASE_GOLANG_22_ALPINE_BUILDER=$BASE_GOLANG_22_ALPINE
 
-FROM $BASE_GOLANG_20_ALPINE_BUILDER as builder
+FROM $BASE_GOLANG_22_ALPINE_BUILDER as builder
 
 WORKDIR /go/src
 
@@ -15,7 +15,8 @@ COPY . .
 WORKDIR /go/src/cmd
 RUN GOOS=linux GOARCH=amd64 go build -o sds-local-volume-scheduler-extender
 
-FROM --platform=linux/amd64 $BASE_GOLANG_20_ALPINE
+FROM --platform=linux/amd64 $BASE_GOLANG_22_ALPINE
 COPY --from=builder /go/src/cmd/sds-local-volume-scheduler-extender /go/src/cmd/sds-local-volume-scheduler-extender
+RUN apk add curl
 
 ENTRYPOINT ["/go/src/cmd/sds-local-volume-scheduler-extender"]

images/sds-local-volume-scheduler-extender/cmd/cmd/root.go

@@ -24,21 +24,23 @@ import (
 	"os"
 	"os/signal"
 	"sds-local-volume-scheduler-extender/api/v1alpha1"
+	"sds-local-volume-scheduler-extender/pkg/cache"
+	"sds-local-volume-scheduler-extender/pkg/controller"
 	"sds-local-volume-scheduler-extender/pkg/kubutils"
 	"sds-local-volume-scheduler-extender/pkg/logger"
 	"sds-local-volume-scheduler-extender/pkg/scheduler"
+	"sigs.k8s.io/controller-runtime/pkg/healthz"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"sync"
 	"syscall"
 	"time"
 
+	"github.com/spf13/cobra"
 	v1 "k8s.io/api/core/v1"
 	sv1 "k8s.io/api/storage/v1"
 	"k8s.io/apimachinery/pkg/runtime"
-	ctrl "sigs.k8s.io/controller-runtime"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-
-	"github.com/spf13/cobra"
 	apiruntime "k8s.io/apimachinery/pkg/runtime"
+	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/yaml"
 )
 
@@ -53,18 +55,21 @@ var resourcesSchemeFuncs = []func(*apiruntime.Scheme) error{
 const (
 	defaultDivisor    = 1
 	defaultListenAddr = ":8000"
+	defaultCacheSize  = 10
 )
 
 type Config struct {
 	ListenAddr     string  `json:"listen"`
 	DefaultDivisor float64 `json:"default-divisor"`
 	LogLevel       string  `json:"log-level"`
+	CacheSize      int     `json:"cache-size"`
 }
 
 var config = &Config{
 	ListenAddr:     defaultListenAddr,
 	DefaultDivisor: defaultDivisor,
 	LogLevel:       "2",
+	CacheSize:      defaultCacheSize,
 }
 
 var rootCmd = &cobra.Command{
@@ -122,22 +127,56 @@ func subMain(parentCtx context.Context) error {
 	}
 	log.Info("[subMain] successfully read scheme CR")
 
-	cl, err := client.New(kConfig, client.Options{
-		Scheme:         scheme,
-		WarningHandler: client.WarningHandlerOptions{},
-	})
+	managerOpts := manager.Options{
+		Scheme: scheme,
+		Logger: log.GetLogger(),
+	}
+
+	mgr, err := manager.New(kConfig, managerOpts)
+	if err != nil {
+		return err
+	}
+
+	schedulerCache := cache.NewCache(*log)
+	log.Info("[subMain] scheduler cache was initialized")
 
-	h, err := scheduler.NewHandler(ctx, cl, *log, config.DefaultDivisor)
+	h, err := scheduler.NewHandler(ctx, mgr.GetClient(), *log, schedulerCache, config.DefaultDivisor)
 	if err != nil {
 		return err
 	}
 	log.Info("[subMain] scheduler handler initialized")
 
+	_, err = controller.RunLVGWatcherCacheController(mgr, *log, schedulerCache)
+	if err != nil {
+		log.Error(err, fmt.Sprintf("[subMain] unable to run %s controller", controller.LVGWatcherCacheCtrlName))
+	}
+	log.Info(fmt.Sprintf("[subMain] successfully ran %s controller", controller.LVGWatcherCacheCtrlName))
+
+	err = controller.RunPVCWatcherCacheController(mgr, *log, schedulerCache)
+	if err != nil {
+		log.Error(err, fmt.Sprintf("[subMain] unable to run %s controller", controller.PVCWatcherCacheCtrlName))
+	}
+	log.Info(fmt.Sprintf("[subMain] successfully ran %s controller", controller.PVCWatcherCacheCtrlName))
+
+	if err = mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
+		log.Error(err, "[subMain] unable to mgr.AddHealthzCheck")
+		os.Exit(1)
+	}
+	log.Info("[subMain] successfully AddHealthzCheck")
+
+	if err = mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
+		log.Error(err, "[subMain] unable to mgr.AddReadyzCheck")
+		os.Exit(1)
+	}
+	log.Info("[subMain] successfully AddReadyzCheck")
+
 	serv := &http.Server{
 		Addr:        config.ListenAddr,
 		Handler:     accessLogHandler(parentCtx, h),
 		ReadTimeout: 30 * time.Second,
 	}
+	log.Info("[subMain] server was initialized")
+
 	var wg sync.WaitGroup
 	defer wg.Wait()
 	ctx, stop := signal.NotifyContext(parentCtx, os.Interrupt, syscall.SIGTERM)
@@ -151,11 +190,22 @@ func subMain(parentCtx context.Context) error {
 		}
 	}()
 
+	go func() {
+		log.Info("[subMain] kube manager will start now")
+		err = mgr.Start(ctx)
+		if err != nil {
+			log.Error(err, "[subMain] unable to mgr.Start")
+			os.Exit(1)
+		}
+	}()
+
 	log.Info(fmt.Sprintf("[subMain] starts serving on: %s", config.ListenAddr))
 	err = serv.ListenAndServe()
 	if !errors.Is(err, http.ErrServerClosed) {
+		log.Error(err, "[subMain] unable to run the server")
 		return err
 	}
+
 	return nil
 }
 

images/sds-local-volume-scheduler-extender/go.mod

@@ -13,7 +13,8 @@ require (
 	k8s.io/apimachinery v0.29.1
 	k8s.io/client-go v0.29.0
 	k8s.io/klog/v2 v2.110.1
-	sigs.k8s.io/controller-runtime v0.17.0
+	k8s.io/utils v0.0.0-20240310230437-4693a0247e57
+	sigs.k8s.io/controller-runtime v0.17.2
 	sigs.k8s.io/yaml v1.4.0
 )
 
@@ -66,7 +67,6 @@ require (
 	k8s.io/apiextensions-apiserver v0.29.0 // indirect
 	k8s.io/component-base v0.29.0 // indirect
 	k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 // indirect
-	k8s.io/utils v0.0.0-20230726121419-3b25d923346b // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
 )

images/sds-local-volume-scheduler-extender/go.sum

@@ -194,10 +194,10 @@ k8s.io/klog/v2 v2.110.1 h1:U/Af64HJf7FcwMcXyKm2RPM22WZzyR7OSpYj5tg3cL0=
 k8s.io/klog/v2 v2.110.1/go.mod h1:YGtd1984u+GgbuZ7e08/yBuAfKLSO0+uR1Fhi6ExXjo=
 k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00 h1:aVUu9fTY98ivBPKR9Y5w/AuzbMm96cd3YHRTU83I780=
 k8s.io/kube-openapi v0.0.0-20231010175941-2dd684a91f00/go.mod h1:AsvuZPBlUDVuCdzJ87iajxtXuR9oktsTctW/R9wwouA=
-k8s.io/utils v0.0.0-20230726121419-3b25d923346b h1:sgn3ZU783SCgtaSJjpcVVlRqd6GSnlTLKgpAAttJvpI=
-k8s.io/utils v0.0.0-20230726121419-3b25d923346b/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
-sigs.k8s.io/controller-runtime v0.17.0 h1:fjJQf8Ukya+VjogLO6/bNX9HE6Y2xpsO5+fyS26ur/s=
-sigs.k8s.io/controller-runtime v0.17.0/go.mod h1:+MngTvIQQQhfXtwfdGw/UOQ/aIaqsYywfCINOtwMO/s=
+k8s.io/utils v0.0.0-20240310230437-4693a0247e57 h1:gbqbevonBh57eILzModw6mrkbwM0gQBEuevE/AaBsHY=
+k8s.io/utils v0.0.0-20240310230437-4693a0247e57/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0=
+sigs.k8s.io/controller-runtime v0.17.2 h1:FwHwD1CTUemg0pW2otk7/U5/i5m2ymzvOXdbeGOUvw0=
+sigs.k8s.io/controller-runtime v0.17.2/go.mod h1:+MngTvIQQQhfXtwfdGw/UOQ/aIaqsYywfCINOtwMO/s=
 sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd h1:EDPBXCAspyGV4jQlpZSudPeMmr1bNJefnuqLsRAsHZo=
 sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd/go.mod h1:B8JuhiUyNFVKdsE8h686QcCxMaH6HrOAZj4vswFpcB0=
 sigs.k8s.io/structured-merge-diff/v4 v4.4.1 h1:150L+0vs/8DA78h1u02ooW1/fFq/Lwr+sGiqlzvrtq4=