Merge pull request #4 from deckhouse/v-kramarenko-adds-extended-sched…

…uler

[controller] Adds a scheduler-extender

Chart.yaml

@@ -1,2 +1,2 @@
-name: sds-lvm-provisioner
+name: sds-lvm
 version: 0.0.1

images/sds-lvm-scheduler-extender/cmd/cmd/root.go

@@ -4,62 +4,70 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	v1 "k8s.io/api/core/v1"
+	sv1 "k8s.io/api/storage/v1"
+	extv1 "k8s.io/apiextensions-apiserver/pkg/apis/apiextensions/v1"
+	"k8s.io/apimachinery/pkg/runtime"
 	"net/http"
 	"os"
 	"os/signal"
+	"sds-lvm-scheduler-extender/pkg/api/v1alpha1"
+	"sds-lvm-scheduler-extender/pkg/kubutils"
+	"sds-lvm-scheduler-extender/pkg/logger"
+	"sds-lvm-scheduler-extender/pkg/scheduler"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sync"
 	"syscall"
 	"time"
 
-	"sds-lvm-scheduler-extender/pkg/scheduler"
+	apiruntime "k8s.io/apimachinery/pkg/runtime"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 
 	"github.com/spf13/cobra"
-	"github.com/topolvm/topolvm"
-	ctrl "sigs.k8s.io/controller-runtime"
-	"sigs.k8s.io/controller-runtime/pkg/log"
-	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 	"sigs.k8s.io/yaml"
 )
 
 var cfgFilePath string
-var zapOpts zap.Options
 
-const defaultDivisor = 1
-const defaultListenAddr = ":8000"
+var resourcesSchemeFuncs = []func(*apiruntime.Scheme) error{
+	v1alpha1.AddToScheme,
+	clientgoscheme.AddToScheme,
+	extv1.AddToScheme,
+	v1.AddToScheme,
+	sv1.AddToScheme,
+}
+
+const (
+	defaultDivisor    = 1
+	defaultListenAddr = ":8000"
+)
 
-// Config represents configuration parameters for topolvm-scheduler
 type Config struct {
-	// ListenAddr is listen address of topolvm-scheduler.
-	ListenAddr string `json:"listen"`
-	// Divisors is a mapping between device-class names and their divisors.
-	Divisors map[string]float64 `json:"divisors"`
-	// DefaultDivisor is the default divisor value.
+	ListenAddr     string  `json:"listen"`
 	DefaultDivisor float64 `json:"default-divisor"`
+	LogLevel       string  `json:"log-level"`
 }
 
 var config = &Config{
 	ListenAddr:     defaultListenAddr,
 	DefaultDivisor: defaultDivisor,
+	LogLevel:       "2",
 }
 
 var rootCmd = &cobra.Command{
-	Use:     "topolvm-scheduler",
-	Version: topolvm.Version,
-	Short:   "a scheduler-extender for TopoLVM",
-	Long: `A scheduler-extender for TopoLVM.
-
+	Use:     "sds-lvm-scheduler",
+	Version: "development",
+	Short:   "a scheduler-extender for SDS-LVM",
+	Long: `A scheduler-extender for SDS-LVM.
 The extender implements filter and prioritize verbs.
-
-The filter verb is "predicate" and served at "/predicate" via HTTP.
+The filter verb is "filter" and served at "/filter" via HTTP.
 It filters out nodes that have less storage capacity than requested.
 The requested capacity is read from "capacity.topolvm.io/<device-class>"
 resource value.
-
 The prioritize verb is "prioritize" and served at "/prioritize" via HTTP.
 It scores nodes with this formula:
-
     min(10, max(0, log2(capacity >> 30 / divisor)))
-
 The default divisor is 1.  It can be changed with a command-line option.
 `,
 	RunE: func(cmd *cobra.Command, args []string) error {
@@ -69,9 +77,6 @@ The default divisor is 1.  It can be changed with a command-line option.
 }
 
 func subMain(parentCtx context.Context) error {
-	ctrl.SetLogger(zap.New(zap.UseFlagOptions(&zapOpts)))
-	logger := log.FromContext(parentCtx)
-
 	if len(cfgFilePath) != 0 {
 		b, err := os.ReadFile(cfgFilePath)
 		if err != nil {
@@ -83,32 +88,59 @@ func subMain(parentCtx context.Context) error {
 		}
 	}
 
-	h, err := scheduler.NewHandler(config.DefaultDivisor, config.Divisors)
+	log, err := logger.NewLogger(logger.Verbosity(config.LogLevel))
+	if err != nil {
+		fmt.Println(fmt.Sprintf("[subMain] unable to initialize logger, err: %s", err.Error()))
+	}
+	log.Info(fmt.Sprintf("[subMain] logger has been initialized, log level: %s", config.LogLevel))
+	ctrl.SetLogger(log.GetLogger())
+
+	kConfig, err := kubutils.KubernetesDefaultConfigCreate()
+	if err != nil {
+		log.Error(err, "[subMain] unable to KubernetesDefaultConfigCreate")
+	}
+	log.Info("[subMain] kubernetes config has been successfully created.")
+
+	scheme := runtime.NewScheme()
+	for _, f := range resourcesSchemeFuncs {
+		err := f(scheme)
+		if err != nil {
+			log.Error(err, "[subMain] unable to add scheme to func")
+			os.Exit(1)
+		}
+	}
+	log.Info("[subMain] successfully read scheme CR")
+
+	cl, err := client.New(kConfig, client.Options{
+		Scheme:         scheme,
+		WarningHandler: client.WarningHandlerOptions{},
+	})
+
+	h, err := scheduler.NewHandler(cl, *log, config.DefaultDivisor)
 	if err != nil {
 		return err
 	}
+	log.Info("[subMain] scheduler handler initialized")
 
 	serv := &http.Server{
 		Addr:        config.ListenAddr,
 		Handler:     accessLogHandler(parentCtx, h),
 		ReadTimeout: 30 * time.Second,
 	}
-
 	var wg sync.WaitGroup
 	defer wg.Wait()
-
 	ctx, stop := signal.NotifyContext(parentCtx, os.Interrupt, syscall.SIGTERM)
 	defer stop() // stop() should be called before wg.Wait() to stop the goroutine correctly.
-
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
 		<-ctx.Done()
 		if err := serv.Shutdown(parentCtx); err != nil {
-			logger.Error(err, "failed to shutdown gracefully")
+			log.Error(err, "failed to shutdown gracefully")
 		}
 	}()
 
+	log.Info(fmt.Sprintf("[subMain] starts serving on: %s", config.ListenAddr))
 	err = serv.ListenAndServe()
 	if !errors.Is(err, http.ErrServerClosed) {
 		return err
@@ -124,7 +156,6 @@ func Execute() {
 		os.Exit(1)
 	}
 }
-
 func init() {
 	rootCmd.PersistentFlags().StringVar(&cfgFilePath, "config", "", "config file")
 }

images/sds-lvm-scheduler-extender/config/config.go

@@ -0,0 +1,45 @@
+/*
+Copyright 2023 Flant JSC
+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 config
+
+import (
+	"os"
+	"sds-lvm-scheduler-extender/pkg/logger"
+)
+
+const (
+	NodeName = "KUBE_NODE_NAME"
+	LogLevel = "LOG_LEVEL"
+)
+
+type Options struct {
+	NodeName string
+	Version  string
+	Loglevel logger.Verbosity
+}
+
+func NewConfig() *Options {
+	var opts Options
+
+	loglevel := os.Getenv(LogLevel)
+	if loglevel == "" {
+		opts.Loglevel = logger.DebugLevel
+	} else {
+		opts.Loglevel = logger.Verbosity(loglevel)
+	}
+
+	opts.Version = "dev"
+
+	return &opts
+}

images/sds-lvm-scheduler-extender/go.mod

@@ -3,12 +3,16 @@ module sds-lvm-scheduler-extender
 go 1.21
 
 require (
+	github.com/go-logr/logr v1.4.1
 	github.com/go-logr/zapr v1.3.0
 	github.com/spf13/cobra v1.8.0
 	github.com/topolvm/topolvm v0.25.0
 	go.uber.org/zap v1.26.0
 	k8s.io/api v0.29.1
+	k8s.io/apiextensions-apiserver v0.29.0
 	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
 	sigs.k8s.io/yaml v1.4.0
 )
@@ -20,7 +24,6 @@ require (
 	github.com/emicklei/go-restful/v3 v3.11.0 // indirect
 	github.com/evanphx/json-patch/v5 v5.8.0 // indirect
 	github.com/fsnotify/fsnotify v1.7.0 // indirect
-	github.com/go-logr/logr v1.4.1 // indirect
 	github.com/go-openapi/jsonpointer v0.19.6 // indirect
 	github.com/go-openapi/jsonreference v0.20.2 // indirect
 	github.com/go-openapi/swag v0.22.3 // indirect
@@ -60,10 +63,7 @@ require (
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
-	k8s.io/apiextensions-apiserver v0.29.0 // indirect
-	k8s.io/client-go v0.29.0 // indirect
 	k8s.io/component-base v0.29.0 // indirect
-	k8s.io/klog/v2 v2.110.1 // 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