Chore: refactor the cue engine with cuex (#162)

* Chore: refactor the cue engine with cuex Signed-off-by: FogDong <[email protected]> * use singelton for compiler Signed-off-by: FogDong <[email protected]> * fix email test casses Signed-off-by: FogDong <[email protected]> * fix test Signed-off-by: FogDong <[email protected]> * add unmarshal to as util func Signed-off-by: FogDong <[email protected]> * add time provider Signed-off-by: FogDong <[email protected]> * fix: fix compiler Signed-off-by: FogDong <[email protected]> * fix: fix lint and rebase Signed-off-by: FogDong <[email protected]> * chore: update cue and fix context key Signed-off-by: FogDong <[email protected]> * fix: fix input fill val Signed-off-by: FogDong <[email protected]> * fix: fix checkpending field path Signed-off-by: FogDong <[email protected]> * fix: fix nil pointer return in kube Signed-off-by: FogDong <[email protected]> * fix: fix error in return Signed-off-by: FogDong <[email protected]> * fix: add mutex for syntax Signed-off-by: FogDong <[email protected]> * fix: add kubeclient runtime param Signed-off-by: FogDong <[email protected]> * chore: clean up stdlib Signed-off-by: FogDong <[email protected]> * fix: do not override runtime params Signed-off-by: FogDong <[email protected]> --------- Signed-off-by: FogDong <[email protected]> Signed-off-by: FogDong <[email protected]>
kubevela · Jul 27, 2024 · 7d94489 · 7d94489
1 parent c3331e7
commit 7d94489
Show file tree

Hide file tree

Showing 95 changed files with 3,913 additions and 9,175 deletions.
diff --git a/cmd/main.go b/cmd/main.go
@@ -53,7 +53,6 @@ import (
 	"github.com/kubevela/workflow/controllers"
 	"github.com/kubevela/workflow/pkg/backup"
 	"github.com/kubevela/workflow/pkg/common"
-	"github.com/kubevela/workflow/pkg/cue/packages"
 	"github.com/kubevela/workflow/pkg/features"
 	"github.com/kubevela/workflow/pkg/monitor/watcher"
 	"github.com/kubevela/workflow/pkg/types"
@@ -222,15 +221,6 @@ func main() {
 		}
 	}
 
-	pd, err := packages.NewPackageDiscover(mgr.GetConfig())
-	if err != nil {
-		klog.Error(err, "Failed to create CRD discovery for CUE package client")
-		if !packages.IsCUEParseErr(err) {
-			os.Exit(1)
-		}
-	}
-	controllerArgs.PackageDiscover = pd
-
 	if useWebhook {
 		klog.InfoS("Enable webhook", "server port", strconv.Itoa(webhookPort))
 		webhook.Register(mgr, controllerArgs)

diff --git a/controllers/suite_test.go b/controllers/suite_test.go
@@ -33,8 +33,8 @@ import (
 	logf "sigs.k8s.io/controller-runtime/pkg/log"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
 
+	"github.com/kubevela/pkg/util/singleton"
 	"github.com/kubevela/workflow/api/v1alpha1"
-	"github.com/kubevela/workflow/pkg/cue/packages"
 	//+kubebuilder:scaffold:imports
 )
 
@@ -69,9 +69,6 @@ var _ = BeforeSuite(func() {
 	Expect(err).NotTo(HaveOccurred())
 	Expect(cfg).NotTo(BeNil())
 
-	pd, err := packages.NewPackageDiscover(cfg)
-	Expect(err).To(BeNil())
-
 	testScheme = scheme.Scheme
 	err = v1alpha1.AddToScheme(testScheme)
 	Expect(err).NotTo(HaveOccurred())
@@ -81,14 +78,12 @@ var _ = BeforeSuite(func() {
 	k8sClient, err = client.New(cfg, client.Options{Scheme: testScheme})
 	Expect(err).NotTo(HaveOccurred())
 	Expect(k8sClient).NotTo(BeNil())
+	singleton.KubeClient.Set(k8sClient)
 
 	reconciler = &WorkflowRunReconciler{
 		Client:   k8sClient,
 		Scheme:   testScheme,
 		Recorder: event.NewAPIRecorder(recorder),
-		Args: Args{
-			PackageDiscover: pd,
-		},
 	}
 
 }, 60)

diff --git a/controllers/testdata/apply-object.yaml b/controllers/testdata/apply-object.yaml
@@ -6,6 +6,18 @@ metadata:
 spec:
   schematic:
     cue:
-      template: "import (\n\t\"vela/op\"\n)\n\napply: op.#Apply & {\n\tvalue:   parameter.value\n\tcluster:
-        parameter.cluster\n}\nparameter: {\n\t// +usage=Specify the value of the object\n\tvalue:
-        {...}\n\t// +usage=Specify the cluster of the object\n\tcluster: *\"\" | string\n}\n"
+      template: |
+        import (
+                "vela/op"
+        )
+
+        apply: op.#Apply & {
+                value:   parameter.value
+                cluster: parameter.cluster
+        }
+        parameter: {
+                // +usage=Specify the value of the object
+                value: {...}
+                // +usage=Specify the cluster of the object
+                cluster: *"" | string
+        }
diff --git a/controllers/testdata/test-apply.yaml b/controllers/testdata/test-apply.yaml
@@ -6,14 +6,46 @@ metadata:
 spec:
   schematic:
     cue:
-      template: "import (\t\"vela/op\"\n)\n\noutput: op.#Apply & {\n\tvalue: {\n\t\tapiVersion:
-        \"apps/v1\"\n\t\tkind:       \"Deployment\"\n\t\tmetadata: {\n\t\t\tname:
-        \     context.stepName\n\t\t\tnamespace: context.namespace\n\t\t}\n\t\tspec:
-        {\n\t\t\tselector: matchLabels: wr: context.stepName\n\t\t\ttemplate: {\n\t\t\t\tmetadata:
-        labels: wr: context.stepName\n\t\t\t\tspec: containers: [{\n\t\t\t\t\tname:
-        \ context.stepName\n\t\t\t\t\timage: parameter.image\n\t\t\t\t\tif parameter[\"cmd\"]
-        != _|_ {\n\t\t\t\t\t\tcommand: parameter.cmd\n\t\t\t\t\t}\n\t\t\t\t\tif parameter[\"message\"]
-        != _|_ {\n\t\t\t\t\t\tenv: [{\n\t\t\t\t\t\t\tname:  \"MESSAGE\"\n\t\t\t\t\t\t\tvalue:
-        parameter.message\n\t\t\t\t\t\t}]\n\t\t\t\t\t}\n\t\t\t\t}]\n\t\t\t}\n\t\t}\n\t}\n}\nwait:
-        op.#ConditionalWait & {\n\tcontinue: output.value.status.readyReplicas ==
-        1\n}\nparameter: {\n\timage:    string\n\tcmd?:     [...string]\n\tmessage?: string\n}\n"
+      template: |
+        import ( "vela/op"
+        )
+
+        output: op.#Apply & {
+                value: {
+                        apiVersion: "apps/v1"
+                        kind:       "Deployment"
+                        metadata: {
+                                name:      context.stepName
+                                namespace: context.namespace
+                        }
+                        spec: {
+                                selector: matchLabels: wr: context.stepName
+                                template: {
+                                        metadata: labels: wr: context.stepName
+                                        spec: containers: [{
+                                                name:  context.stepName
+                                                image: parameter.image
+                                                if parameter["cmd"] != _|_ {
+                                                        command: parameter.cmd
+                                                }
+                                                if parameter["message"] != _|_ {
+                                                        env: [{
+                                                                name:  "MESSAGE"
+                                                                value: parameter.message
+                                                        }]
+                                                }
+                                        }]
+                                }
+                        }
+                }
+        }
+        wait: op.#ConditionalWait & {
+                if len(output.value.status) > 0 if output.value.status.readyReplicas == 1 {
+                        continue: true
+                }
+        }
+        parameter: {
+                image: string
+                cmd?: [...string]
+                message?: string
+        }
diff --git a/controllers/workflow_test.go b/controllers/workflow_test.go
@@ -789,11 +789,19 @@ var _ = Describe("Test Workflow", func() {
 		checkRun := &v1alpha1.WorkflowRun{}
 		Expect(k8sClient.Get(ctx, wrKey, checkRun)).Should(BeNil())
 
+		for i := 0; i < wfTypes.MaxWorkflowStepErrorRetryTimes; i++ {
+			tryReconcile(reconciler, wr.Name, wr.Namespace)
+			Expect(k8sClient.Get(ctx, wrKey, checkRun)).Should(BeNil())
+			Expect(checkRun.Status.Message).Should(BeEquivalentTo(""))
+			Expect(checkRun.Status.Phase).Should(BeEquivalentTo(v1alpha1.WorkflowStateExecuting))
+			Expect(checkRun.Status.Steps[0].Phase).Should(BeEquivalentTo(v1alpha1.WorkflowStepPhaseFailed))
+		}
+
 		tryReconcile(reconciler, wr.Name, wr.Namespace)
 		Expect(k8sClient.Get(ctx, wrKey, checkRun)).Should(BeNil())
 		Expect(checkRun.Status.Phase).Should(BeEquivalentTo(v1alpha1.WorkflowStateFailed))
 		Expect(checkRun.Status.Steps[0].Phase).Should(BeEquivalentTo(v1alpha1.WorkflowStepPhaseFailed))
-		Expect(checkRun.Status.Steps[0].Reason).Should(BeEquivalentTo(wfTypes.StatusReasonRendering))
+		Expect(checkRun.Status.Steps[0].Reason).Should(BeEquivalentTo(wfTypes.StatusReasonFailedAfterRetries))
 		Expect(checkRun.Status.Steps[1].Phase).Should(BeEquivalentTo(v1alpha1.WorkflowStepPhaseSkipped))
 	})
 

diff --git a/controllers/workflowrun_controller.go b/controllers/workflowrun_controller.go
@@ -44,11 +44,11 @@ import (
 	"github.com/kubevela/workflow/api/condition"
 	"github.com/kubevela/workflow/api/v1alpha1"
 	wfContext "github.com/kubevela/workflow/pkg/context"
-	"github.com/kubevela/workflow/pkg/cue/packages"
 	"github.com/kubevela/workflow/pkg/executor"
 	"github.com/kubevela/workflow/pkg/features"
 	"github.com/kubevela/workflow/pkg/generator"
 	"github.com/kubevela/workflow/pkg/monitor/metrics"
+	providertypes "github.com/kubevela/workflow/pkg/providers/types"
 	"github.com/kubevela/workflow/pkg/types"
 )
 
@@ -58,8 +58,6 @@ type Args struct {
 	ConcurrentReconciles int
 	// IgnoreWorkflowWithoutControllerRequirement indicates that workflow controller will not process the workflowrun without 'workflowrun.oam.dev/controller-version-require' annotation.
 	IgnoreWorkflowWithoutControllerRequirement bool
-	// PackageDiscover discover the packages
-	PackageDiscover *packages.PackageDiscover
 }
 
 // WorkflowRunReconciler reconciles a WorkflowRun object
@@ -90,6 +88,10 @@ func (r *WorkflowRunReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	defer cancel()
 
 	ctx = types.SetNamespaceInCtx(ctx, req.Namespace)
+	ctx = providertypes.WithLabelParams(ctx, map[string]string{
+		types.LabelWorkflowRunName:      req.Name,
+		types.LabelWorkflowRunNamespace: req.Namespace,
+	})
 
 	logCtx := monitorContext.NewTraceContext(ctx, "").AddTag("workflowrun", req.String())
 	logCtx.Info("Start reconcile workflowrun")
@@ -128,10 +130,7 @@ func (r *WorkflowRunReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	}
 	isUpdate := instance.Status.Message != ""
 
-	runners, err := generator.GenerateRunners(logCtx, instance, types.StepGeneratorOptions{
-		PackageDiscover: r.PackageDiscover,
-		Client:          r.Client,
-	})
+	runners, err := generator.GenerateRunners(logCtx, instance, types.StepGeneratorOptions{})
 	if err != nil {
 		logCtx.Error(err, "[generate runners]")
 		r.Recorder.Event(run, event.Warning(v1alpha1.ReasonGenerate, errors.WithMessage(err, v1alpha1.MessageFailedGenerate)))
@@ -143,7 +142,7 @@ func (r *WorkflowRunReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		Client: r.Client,
 		run:    run,
 	}
-	executor := executor.New(instance, r.Client, patcher.patchStatus)
+	executor := executor.New(instance, executor.WithStatusPatcher(patcher.patchStatus))
 	state, err := executor.ExecuteRunners(logCtx, runners)
 	if err != nil {
 		logCtx.Error(err, "[execute runners]")

diff --git a/e2e/e2e_suite_test.go b/e2e/e2e_suite_test.go
@@ -31,6 +31,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/config"
 
+	"github.com/kubevela/pkg/util/singleton"
 	"github.com/kubevela/pkg/util/test/definition"
 
 	"github.com/kubevela/workflow/api/v1alpha1"
@@ -58,9 +59,11 @@ var k8sClient client.Client
 var _ = BeforeSuite(func() {
 	conf, err := config.GetConfig()
 	Expect(err).Should(BeNil())
+	singleton.KubeConfig.Set(conf)
 
 	k8sClient, err = client.New(conf, client.Options{Scheme: scheme})
 	Expect(err).Should(BeNil())
+	singleton.KubeClient.Set(k8sClient)
 
 	prepareWorkflowDefinitions()
 })