syncctl: adjust scheduled pod resource requirement

syncctl: nodeSelecor config setting for scheduled pods
jitsucom · Dec 22, 2023 · 036712a · 036712a
1 parent 813fe75
commit 036712a
Show file tree

Hide file tree

Showing 3 changed files with 21 additions and 8 deletions.
diff --git a/bulkerapp/app/app.go b/bulkerapp/app/app.go
@@ -123,20 +123,21 @@ func (a *Context) ShutdownSignal() error {
 
 // TODO: graceful shutdown and cleanups. Flush producer
 func (a *Context) Cleanup() error {
+	time.Sleep(2 * time.Second)
 	a.cron.Close()
 	_ = a.topicManager.Close()
-	_ = a.batchProducer.Close()
-	_ = a.streamProducer.Close()
 	_ = a.backupsLogger.Close()
 	_ = a.repository.Close()
 	_ = a.configurationSource.Close()
+	_ = a.eventsLogService.Close()
+	_ = a.fastStore.Close()
+	_ = a.batchProducer.Close()
+	_ = a.streamProducer.Close()
 	if a.config.ShutdownExtraDelay > 0 {
 		logging.Infof("Waiting %d seconds before http server shutdown...", a.config.ShutdownExtraDelay)
 		time.Sleep(time.Duration(a.config.ShutdownExtraDelay) * time.Second)
 	}
 	_ = a.metricsServer.Stop()
-	_ = a.eventsLogService.Close()
-	_ = a.fastStore.Close()
 	return nil
 }
 

diff --git a/sync-controller/config.go b/sync-controller/config.go
@@ -28,6 +28,8 @@ type Config struct {
 	KubernetesNamespace    string `mapstructure:"KUBERNETES_NAMESPACE" default:"default"`
 	KubernetesClientConfig string `mapstructure:"KUBERNETES_CLIENT_CONFIG" default:"local"`
 	KubernetesContext      string `mapstructure:"KUBERNETES_CONTEXT"`
+	// nodeSelector for sync pods in json format, e.g: {"disktype": "ssd"}
+	KubernetesNodeSelector string `mapstructure:"KUBERNETES_NODE_SELECTOR"`
 
 	ContainerStatusCheckSeconds int `mapstructure:"CONTAINER_STATUS_CHECK_SECONDS" default:"10"`
 	ContainerInitTimeoutSeconds int `mapstructure:"CONTAINER_INIT_TIMEOUT_SECONDS" default:"180"`

diff --git a/sync-controller/job_runner.go b/sync-controller/job_runner.go
@@ -5,6 +5,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"github.com/hjson/hjson-go/v4"
 	"github.com/jitsucom/bulker/jitsubase/appbase"
 	"github.com/jitsucom/bulker/jitsubase/safego"
 	"github.com/jitsucom/bulker/jitsubase/utils"
@@ -342,6 +343,14 @@ func (j *JobRunner) createPod(podName string, task TaskDescriptor, configuration
 			MountPath: "/pipes",
 		},
 	}
+	var nodeSelector map[string]string
+	if j.config.KubernetesNodeSelector != "" {
+		nodeSelector = map[string]string{}
+		err := hjson.Unmarshal([]byte(j.config.KubernetesNodeSelector), &nodeSelector)
+		if err != nil {
+			j.Errorf("failed to parse node selector from string: %s\nIngoring it. Error: %v", j.config.KubernetesNodeSelector, err)
+		}
+	}
 	if !configuration.IsEmpty() {
 		items := []v1.KeyToPath{}
 		for k := range configuration.ToMap() {
@@ -377,6 +386,7 @@ func (j *JobRunner) createPod(podName string, task TaskDescriptor, configuration
 		},
 		Spec: v1.PodSpec{
 			RestartPolicy: v1.RestartPolicyNever,
+			NodeSelector:  nodeSelector,
 			Containers: []v1.Container{
 				{Name: "source",
 					Image:   fmt.Sprintf("%s:%s", task.Package, task.PackageVersion),
@@ -391,8 +401,8 @@ func (j *JobRunner) createPod(podName string, task TaskDescriptor, configuration
 							v1.ResourceMemory: *resource.NewQuantity(int64(math.Pow(2, 31)), resource.BinarySI),
 						},
 						Requests: v1.ResourceList{
-							v1.ResourceCPU: *resource.NewMilliQuantity(int64(500), resource.DecimalSI),
-							// 512Mi
+							v1.ResourceCPU: *resource.NewMilliQuantity(int64(125), resource.DecimalSI),
+							// 256Mi
 							v1.ResourceMemory: *resource.NewQuantity(int64(math.Pow(2, 29)), resource.BinarySI),
 						},
 					},
@@ -410,9 +420,9 @@ func (j *JobRunner) createPod(podName string, task TaskDescriptor, configuration
 							v1.ResourceMemory: *resource.NewQuantity(int64(math.Pow(2, 29)), resource.BinarySI),
 						},
 						Requests: v1.ResourceList{
-							v1.ResourceCPU: *resource.NewMilliQuantity(int64(100), resource.DecimalSI),
+							v1.ResourceCPU: *resource.NewMilliQuantity(int64(0), resource.DecimalSI),
 							// 256
-							v1.ResourceMemory: *resource.NewQuantity(int64(math.Pow(2, 28)), resource.BinarySI),
+							v1.ResourceMemory: *resource.NewQuantity(int64(0), resource.BinarySI),
 						},
 					},
 				},