增加 hpa

.air.conf

@@ -0,0 +1,71 @@
+# Config file for [Air](https://github.com/cosmtrek/air) in TOML format
+
+# Working directory
+# . or absolute path, please note that the directories following must be under root.
+root = "."
+tmp_dir = "tmp"
+
+[build]
+# Just plain old shell command. You could use `make` as well. go build -o bin/manager cmd/main.go
+cmd = "go build -o /home/tmp/main cmd/main.go"
+# Binary file yields from `cmd`.
+bin = "/home/tmp/main"
+# Customize binary, can setup environment variables when run your app.
+full_bin = "/home/tmp/main"
+# Watch these filename extensions.
+include_ext = ["go", "tpl", "tmpl", "html","mod","sum","conf","yaml"]
+# Ignore these filename extensions or directories.
+exclude_dir = ["assets", "tmp", "vendor", "frontend/node_modules"]
+# Watch these directories if you specified.
+include_dir = []
+# Watch these files.
+include_file = []
+# Exclude files.
+exclude_file = []
+# Exclude specific regular expressions.
+exclude_regex = ["_test\\.go"]
+# Exclude unchanged files.
+exclude_unchanged = true
+# Follow symlink for directories
+follow_symlink = true
+# This log file places in your tmp_dir.
+log = "air.log"
+# Poll files for changes instead of using fsnotify.
+poll = false
+# Poll interval (defaults to the minimum interval of 500ms).
+poll_interval = 5000 # ms
+# It's not necessary to trigger build each time file changes if it's too frequent.
+delay = 5000 # ms
+# Stop running old binary when build errors occur.
+stop_on_error = true
+# Send Interrupt signal before killing process (windows does not support this feature)
+send_interrupt = true
+# Delay after sending Interrupt signal
+kill_delay = 5000 # ms
+# Rerun binary or not
+rerun = false
+# Delay after each executions
+rerun_delay = 500
+# Add additional arguments when running binary (bin/full_bin). Will run './tmp/main hello world'.
+# args_bin = ["--configFile", "config/conf.yaml"]
+
+[log]
+# Show log time
+time = true
+# Only show main log (silences watcher, build, runner)
+main_only = false
+
+[color]
+# Customize each part's color. If no color found, use the raw app log.
+main = "magenta"
+watcher = "cyan"
+build = "yellow"
+runner = "green"
+
+[misc]
+# Delete tmp directory on exit
+clean_on_exit = true
+
+[screen]
+clear_on_rebuild = true
+keep_scroll = true

api/v1/taskcrd_types.go

@@ -28,11 +28,13 @@ type TaskCrdSpec struct {
 	// INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
 	// Important: Run "make" to regenerate code after modifying this file
 
-	Deploy        string `json:"deploy,omitempty"`
-	Start         string `json:"start,omitempty"`
-	StartReplicas int32  `json:"start_replicas,omitempty"`
-	End           string `json:"end,omitempty"`
-	EndReplicas   int32  `json:"end_replicas,omitempty"`
+	Deployment         string `json:"deployment,omitempty"`
+	HPA                string `json:"hpa,omitempty"`
+	AverageUtilization int32  `json:"average_utilization,omitempty"`
+	Start              string `json:"start,omitempty"`
+	MaxReplicas        int32  `json:"max_replicas,omitempty"`
+	End                string `json:"end,omitempty"`
+	EndReplicas        int32  `json:"end_replicas,omitempty"`
 }
 
 // TaskCrdStatus defines the observed state of TaskCrd

config/samples/webapp_v1_taskcrd.yaml

@@ -5,6 +5,12 @@ metadata:
     app.kubernetes.io/name: task-crd
     app.kubernetes.io/managed-by: kustomize
   name: taskcrd-sample
+  namespace: m1
 spec:
-  deploy: nginx
-  replicas: 2
+  deployment: nginx
+  hpa: php-apache
+  average_utilization: 10
+  start: "* * * * * *"
+  max_replicas: 10
+  end: "10 * * * * *"
+  end_replicas: 1

internal/controller/taskcrd_controller.go

@@ -18,9 +18,13 @@ package controller
 
 import (
 	"context"
+	"sync/atomic"
+	"time"
 
 	"github.com/robfig/cron/v3"
 	v1 "k8s.io/api/apps/v1"
+	v2 "k8s.io/api/autoscaling/v2"
+	corev1 "k8s.io/api/core/v1"
 	"k8s.io/apimachinery/pkg/runtime"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -50,44 +54,101 @@ type TaskCrdReconciler struct {
 func (r *TaskCrdReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	_ = log.FromContext(ctx)
 
+	// 获取 CRD
 	var taskCrd webappv1.TaskCrd
 	err := r.Client.Get(ctx, req.NamespacedName, &taskCrd)
 	if err != nil {
 		return ctrl.Result{}, err
 	}
 
-	if taskCrd.Spec.Deploy != "" {
-		var deploys v1.DeploymentList
-		err = r.Client.List(ctx, &deploys)
-		if err != nil {
-			return ctrl.Result{}, err
-		}
+	// 获取 deployment
+	if taskCrd.Spec.Deployment == "" {
+		return ctrl.Result{}, nil
+	}
 
-		c := cron.New()
+	var deploy v1.Deployment
+	err = r.Client.Get(ctx, client.ObjectKey{Namespace: taskCrd.Namespace, Name: taskCrd.Spec.Deployment}, &deploy)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
 
-		for _, deploy := range deploys.Items {
-			if deploy.ObjectMeta.Name == taskCrd.Spec.Deploy {
+	// 获取 hpa
+	if taskCrd.Spec.HPA == "" {
+		return ctrl.Result{}, nil
+	}
 
-				c.AddFunc(taskCrd.Spec.Start, func() {
-					deploy.Spec.Replicas = &taskCrd.Spec.StartReplicas
+	hpa := v2.HorizontalPodAutoscaler{}
+	err = r.Client.Get(ctx, client.ObjectKey{Namespace: req.Namespace, Name: taskCrd.Spec.HPA}, &hpa)
+	if err != nil {
+		return ctrl.Result{}, err
+	}
+
+	var (
+		start = make(chan struct{})
+		end   = make(chan struct{})
+	)
+	// 都存在，则开始执行
+	c := cron.New()
+	c.AddFunc(taskCrd.Spec.Start, func() {
+		start <- struct{}{}
+	})
+
+	c.AddFunc(taskCrd.Spec.End, func() {
+		end <- struct{}{}
+	})
+
+	for {
+	OUT:
+		select {
+		case <-ctx.Done():
+			return ctrl.Result{}, ctx.Err()
+		case <-start:
+			log.Log.Info("task start: ", time.Now())
+			tick := time.NewTicker(10 * time.Second)
+			for {
+				select {
+				case <-ctx.Done():
+					return ctrl.Result{}, ctx.Err()
+				case <-end:
+					log.Log.Info("task end: ", time.Now())
+					atomic.StoreInt32(deploy.Spec.Replicas, taskCrd.Spec.EndReplicas)
 					err = r.Client.Update(ctx, &deploy)
 					if err != nil {
-						return
+						return ctrl.Result{}, err
 					}
-				})
-
-				c.AddFunc(taskCrd.Spec.End, func() {
-					deploy.Spec.Replicas = &taskCrd.Spec.EndReplicas
-					err = r.Client.Update(ctx, &deploy)
+					break OUT
+				case <-tick.C:
+					err = r.Client.Get(ctx, client.ObjectKey{Namespace: req.Namespace, Name: taskCrd.Spec.HPA}, &hpa)
 					if err != nil {
-						return
+						return ctrl.Result{}, err
 					}
-				})
+					for _, metrics := range hpa.Status.CurrentMetrics {
+						// CPU 的平均使用率小于设定值，则修改对应 deployment 副本数目
+						if metrics.Resource.Name == corev1.ResourceCPU && *metrics.Resource.Current.AverageUtilization < taskCrd.Spec.AverageUtilization {
+							err = r.Client.Get(ctx, client.ObjectKey{Namespace: taskCrd.Namespace, Name: taskCrd.Spec.Deployment}, &deploy)
+							if err != nil {
+								return ctrl.Result{}, err
+							}
+							log.Log.Info("get deploy", deploy.Name, "replicas", deploy.Status.Replicas)
+							if deploy.Status.Replicas < taskCrd.Spec.MaxReplicas {
+								atomic.AddInt32(deploy.Spec.Replicas, 1)
+								err = r.Client.Update(ctx, &deploy)
+								if err != nil {
+									return ctrl.Result{}, err
+								}
+							} else {
+								// 不小于，则结束
+								tick.Stop()
+							}
+						}
+					}
+				}
 			}
 		}
-
 	}
 
+	c.Stop()
+
 	return ctrl.Result{}, nil
 }