初步描述控制器

Author: googs1025

pkg/apis/jobflow/v1alpha1/types.go

@@ -24,13 +24,16 @@ type JobFlowSpec struct {
 }
 
 type Flow struct {
+	// job name, namespace 就是默认 namespace
 	Name string `json:"name"`
-	// cd
-	JobTemplate  v1.JobSpec `json:"jobTemplate"`
-	Dependencies []string   `json:"dependencies"`
+	// 用于赋值
+	JobTemplate v1.JobSpec `json:"jobTemplate"`
+	// 依赖项
+	Dependencies []string `json:"dependencies"`
 }
 
 type JobFlowStatus struct {
+	// 用于存储 map 是 name/namespace 的方式 或是只要 name就行
 	JobStatusList map[string]v1.JobStatus `json:"jobStatusList,omitempty"`
 }
 

pkg/apis/jobflow/v1alpha1/zz_generated.deepcopy.go

@@ -2,16 +2,24 @@ package controller
 
 import (
 	"context"
+	"fmt"
 	"github.com/go-logr/logr"
+	jobflowv1alpha1 "github.com/myoperator/jobflowoperator/pkg/apis/jobflow/v1alpha1"
+	v1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/tools/record"
+	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+	"time"
 )
 
 type JobFlowController struct {
-	client client.Client
-	Scheme *runtime.Scheme
-	log    logr.Logger
+	client   client.Client
+	Scheme   *runtime.Scheme
+	Recorder record.EventRecorder
+	log      logr.Logger
 }
 
 func NewJobFlowController(client client.Client, log logr.Logger, scheme *runtime.Scheme) *JobFlowController {
@@ -25,5 +33,27 @@ func NewJobFlowController(client client.Client, log logr.Logger, scheme *runtime
 // Reconcile 调协 loop
 func (r *JobFlowController) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
 
+	klog.Info("start jobFlow Reconcile..........")
+	klog.Info(fmt.Sprintf("req.%v", req))
+
+	// load JobFlow by namespace
+	jobFlow := &jobflowv1alpha1.JobFlow{}
+	time.Sleep(time.Second)
+	err := r.client.Get(ctx, req.NamespacedName, jobFlow)
+	if err != nil {
+		// If no instance is found, it will be returned directly
+		if errors.IsNotFound(err) {
+			klog.Info(fmt.Sprintf("not found jobFlow : %v", req.Name))
+			return reconcile.Result{}, nil
+		}
+		klog.Error(err, err.Error())
+		r.Recorder.Eventf(jobFlow, v1.EventTypeWarning, "Created", err.Error())
+		return reconcile.Result{}, err
+	}
+
+	// 启动 依序启动 job 任务
+
+	// 改变 job 状态
+
 	return reconcile.Result{}, nil
 }