Fix: enable Observability in runtime clusters (#2886)

* Fix: enable Observability in runtime clusters

Enabled add-on observability in runtime clusters. In each cluster, a sub-domain
will be allocated to the cluste

Signed-off-by: Zheng Xi Zhou <zzxwill@gmail.com>


* Fix：add workflow step definition and ut

Added a new workflow step definition `apply-application-in-parallel`,
and added more unit tests

charts/vela-core/templates/defwithtemplate/apply-application-in-parallel.yaml

@@ -0,0 +1,21 @@
+# Code generated by KubeVela templates. DO NOT EDIT. Please edit the original cue file.
+# Definition source cue file: vela-templates/definitions/internal/apply-application-in-parallel.cue
+apiVersion: core.oam.dev/v1beta1
+kind: WorkflowStepDefinition
+metadata:
+  annotations:
+    definition.oam.dev/description: Apply components of an application in parallel for your workflow steps
+  labels:
+    custom.definition.oam.dev/ui-hidden: "true"
+  name: apply-application-in-parallel
+  namespace: {{.Values.systemDefinitionNamespace}}
+spec:
+  schematic:
+    cue:
+      template: |
+        import (
+        	"vela/op"
+        )
+
+        output: op.#ApplyApplicationInParallel & {}
+

charts/vela-minimal/templates/defwithtemplate/apply-application-in-parallel.yaml

@@ -0,0 +1,21 @@
+# Code generated by KubeVela templates. DO NOT EDIT. Please edit the original cue file.
+# Definition source cue file: vela-templates/definitions/internal/apply-application-in-parallel.cue
+apiVersion: core.oam.dev/v1beta1
+kind: WorkflowStepDefinition
+metadata:
+  annotations:
+    definition.oam.dev/description: Apply components of an application in parallel for your workflow steps
+  labels:
+    custom.definition.oam.dev/ui-hidden: "true"
+  name: apply-application-in-parallel
+  namespace: {{.Values.systemDefinitionNamespace}}
+spec:
+  schematic:
+    cue:
+      template: |
+        import (
+        	"vela/op"
+        )
+
+        output: op.#ApplyApplicationInParallel & {}
+

pkg/addon/addon.go

@@ -17,6 +17,7 @@ limitations under the License.
 package addon
 
 import (
+	"bytes"
 	"context"
 	"encoding/json"
 	"fmt"
@@ -25,12 +26,9 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"text/template"
 	"time"
 
-	"k8s.io/client-go/rest"
-
-	"github.com/oam-dev/kubevela/pkg/definition"
-
 	"cuelang.org/go/cue"
 	cueyaml "cuelang.org/go/encoding/yaml"
 	"github.com/google/go-github/v32/github"
@@ -40,7 +38,10 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime"
 	k8syaml "k8s.io/apimachinery/pkg/runtime/serializer/yaml"
+	"k8s.io/client-go/rest"
+	"k8s.io/klog/v2"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/yaml"
 
@@ -50,6 +51,8 @@ import (
 	utils2 "github.com/oam-dev/kubevela/pkg/controller/utils"
 	cuemodel "github.com/oam-dev/kubevela/pkg/cue/model"
 	"github.com/oam-dev/kubevela/pkg/cue/model/value"
+	"github.com/oam-dev/kubevela/pkg/definition"
+	"github.com/oam-dev/kubevela/pkg/multicluster"
 	"github.com/oam-dev/kubevela/pkg/oam"
 	"github.com/oam-dev/kubevela/pkg/oam/util"
 	"github.com/oam-dev/kubevela/pkg/utils"
@@ -95,6 +98,60 @@ var (
 	EnableLevelOptions = ListOptions{GetDetail: true, GetDefinition: true, GetResource: true, GetTemplate: true, GetParameter: true, GetDefSchema: true}
 )
 
+// ObservabilityEnvironment contains the Observability addon's domain for each cluster
+type ObservabilityEnvironment struct {
+	Cluster string
+	Domain  string
+}
+
+// ObservabilityEnvBindingValues is a list of ObservabilityEnvironment and will be used to render observability-env-binding.yaml
+type ObservabilityEnvBindingValues struct {
+	Envs []ObservabilityEnvironment
+}
+
+const (
+	// ObservabilityEnvBindingEnvTag is the env Tag for env-binding settings for observability addon
+	ObservabilityEnvBindingEnvTag = `        envs:`
+
+	// ObservabilityEnvBindingEnvTmpl is the env values for env-binding settings for observability addon
+	ObservabilityEnvBindingEnvTmpl = `
+        {{ with .Envs}}
+          {{ range . }}
+          - name: {{.Cluster}}
+            placement:
+              clusterSelector:
+                name: {{.Cluster}}
+            patch:
+              components:
+                - name: grafana
+                  type: helm
+                  traits:
+                    - type: pure-ingress
+                      properties:
+                        domain: {{.Domain}}
+          {{ end }}
+        {{ end }}`
+
+	// ObservabilityWorkflowStepsTag is the workflow steps Tag for observability addon
+	ObservabilityWorkflowStepsTag = `steps:`
+
+	// ObservabilityWorkflow4EnvBindingTmpl is the workflow for env-binding settings for observability addon
+	ObservabilityWorkflow4EnvBindingTmpl = `
+{{ with .Envs}}
+  {{ range . }}
+  - name: {{ .Cluster }}
+    type: deploy2env
+    properties:
+      policy: domain
+      env: {{ .Cluster }}
+      parallel: true
+  {{ end }}
+{{ end }}`
+)
+
+// ErrorNoDomain is the error when no domain is found
+var ErrorNoDomain = errors.New("domain is not set")
+
 // GetAddonsFromReader list addons from AsyncReader
 func GetAddonsFromReader(r AsyncReader, opt ListOptions) ([]*Addon, error) {
 	var addons []*Addon
@@ -424,7 +481,7 @@ func genAddonAPISchema(addonRes *Addon) error {
 }
 
 // RenderApp render a K8s application
-func RenderApp(addon *Addon, config *rest.Config, args map[string]interface{}) (*v1beta1.Application, error) {
+func RenderApp(ctx context.Context, k8sClient client.Client, addon *Addon, config *rest.Config, args map[string]interface{}) (*v1beta1.Application, error) {
 	if args == nil {
 		args = map[string]interface{}{}
 	}
@@ -446,9 +503,6 @@ func RenderApp(addon *Addon, config *rest.Config, args map[string]interface{}) (
 	}
 	app.Name = Convert2AppName(addon.Name)
 	app.Labels = util.MergeMapOverrideWithDst(app.Labels, map[string]string{oam.LabelAddonName: addon.Name})
-	if app.Spec.Workflow == nil {
-		app.Spec.Workflow = &v1beta1.Workflow{}
-	}
 	for _, namespace := range addon.NeedNamespace {
 		comp := common2.ApplicationComponent{
 			Type:       "raw",
@@ -476,7 +530,8 @@ func RenderApp(addon *Addon, config *rest.Config, args map[string]interface{}) (
 		app.Spec.Components = append(app.Spec.Components, *comp)
 	}
 
-	if isDeployToRuntimeOnly(addon) {
+	switch {
+	case isDeployToRuntimeOnly(addon):
 		if app.Spec.Workflow == nil {
 			app.Spec.Workflow = &v1beta1.Workflow{Steps: make([]v1beta1.WorkflowStep, 0)}
 		}
@@ -489,7 +544,32 @@ func RenderApp(addon *Addon, config *rest.Config, args map[string]interface{}) (
 				Name: "deploy-runtime",
 				Type: "deploy2runtime",
 			})
-	} else {
+	case addon.Name == "observability":
+		arg, ok := args["domain"]
+		if !ok {
+			return nil, ErrorNoDomain
+		}
+		domain := arg.(string)
+		policies, err := preparePolicies4Observability(ctx, k8sClient, domain)
+		if err != nil {
+			return nil, errors.Wrap(err, "fail to render the policies for Add-on Observability")
+		}
+		app.Spec.Policies = policies
+
+		app.Spec.Workflow = &v1beta1.Workflow{
+			Steps: []v1beta1.WorkflowStep{{
+				Name: "deploy-control-plane",
+				Type: "apply-application-in-parallel",
+			}},
+		}
+
+		workflowSteps, err := prepareWorkflow4Observability(ctx, k8sClient, domain)
+		if err != nil {
+			return nil, errors.Wrap(err, "fail to prepare the workflow for Add-on Observability")
+		}
+		app.Spec.Workflow.Steps = append(app.Spec.Workflow.Steps, workflowSteps...)
+
+	default:
 		for _, def := range addon.Definitions {
 			comp, err := renderRawComponent(def)
 			if err != nil {
@@ -570,6 +650,99 @@ func RenderDefinitionSchema(addon *Addon) ([]*unstructured.Unstructured, error)
 	}
 	return schemaConfigmaps, nil
 }
+
+func allocateDomainForAddon(ctx context.Context, k8sClient client.Client, domain string) ([]ObservabilityEnvironment, error) {
+	secrets, err := multicluster.ListExistingClusterSecrets(ctx, k8sClient)
+	if err != nil {
+		klog.Error(err, "failed to list existing cluster secrets")
+		return nil, err
+	}
+
+	envs := make([]ObservabilityEnvironment, len(secrets))
+
+	for i, secret := range secrets {
+		cluster := secret.Name
+		domain := fmt.Sprintf("%s.%s", cluster, domain)
+		envs[i] = ObservabilityEnvironment{
+			Cluster: cluster,
+			Domain:  domain,
+		}
+	}
+
+	return envs, nil
+}
+
+func preparePolicies4Observability(ctx context.Context, k8sClient client.Client, domain string) ([]v1beta1.AppPolicy, error) {
+	clusters, err := allocateDomainForAddon(ctx, k8sClient, domain)
+	if err != nil {
+		return nil, err
+	}
+
+	envProperties, err := render(clusters, ObservabilityEnvBindingEnvTmpl)
+	if err != nil {
+		return nil, err
+	}
+
+	var properties runtime.RawExtension
+	envs := fmt.Sprintf("%s\n%s", ObservabilityEnvBindingEnvTag, envProperties)
+	envJSON, err := yaml.YAMLToJSON([]byte(envs))
+	if err != nil {
+		return nil, err
+	}
+	err = json.Unmarshal(envJSON, &properties)
+	if err != nil {
+		return nil, err
+	}
+
+	policies := []v1beta1.AppPolicy{{
+		Name:       "domain",
+		Type:       "env-binding",
+		Properties: &properties,
+	}}
+
+	return policies, nil
+}
+
+func prepareWorkflow4Observability(ctx context.Context, k8sClient client.Client, domain string) ([]v1beta1.WorkflowStep, error) {
+	clusters, err := allocateDomainForAddon(ctx, k8sClient, domain)
+	if err != nil {
+		return nil, err
+	}
+
+	envBindingWorkflow, err := render(clusters, ObservabilityWorkflow4EnvBindingTmpl)
+	if err != nil {
+		return nil, err
+	}
+
+	var workflow v1beta1.Workflow
+	envs := fmt.Sprintf("%s\n%s", ObservabilityWorkflowStepsTag, envBindingWorkflow)
+	envJSON, err := yaml.YAMLToJSON([]byte(envs))
+	if err != nil {
+		return nil, err
+	}
+	err = json.Unmarshal(envJSON, &workflow)
+	if err != nil {
+		return nil, err
+	}
+
+	return workflow.Steps, nil
+}
+
+func render(envs []ObservabilityEnvironment, tmpl string) (string, error) {
+	todos := ObservabilityEnvBindingValues{
+		Envs: envs,
+	}
+
+	t := template.Must(template.New("grafana").Parse(tmpl))
+	var rendered bytes.Buffer
+	err := t.Execute(&rendered, todos)
+	if err != nil {
+		return "", err
+	}
+
+	return rendered.String(), nil
+}
+
 func isDeployToRuntimeOnly(addon *Addon) bool {
 	if addon.DeployTo == nil {
 		return false
@@ -766,7 +939,7 @@ func (h *Handler) checkDependencies() error {
 }
 
 func (h *Handler) dispatchAddonResource() error {
-	app, err := RenderApp(h.addon, h.config, h.args)
+	app, err := RenderApp(h.ctx, h.cli, h.addon, h.config, h.args)
 	if err != nil {
 		return errors.Wrap(err, "render addon application fail")
 	}

pkg/addon/addon_test.go

@@ -17,6 +17,8 @@ limitations under the License.
 package addon
 
 import (
+	"context"
+	"encoding/json"
 	"encoding/xml"
 	"net/http"
 	"net/http/httptest"
@@ -25,7 +27,11 @@ import (
 	"strings"
 	"testing"
 
+	v1alpha12 "github.com/oam-dev/cluster-gateway/pkg/apis/cluster/v1alpha1"
 	"gotest.tools/assert"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 )
 
 var paths = []string{
@@ -84,6 +90,8 @@ var ossHandler http.HandlerFunc = func(rw http.ResponseWriter, req *http.Request
 	}
 }
 
+var ctx = context.Background()
+
 func TestGetAddon(t *testing.T) {
 	server := httptest.NewServer(ossHandler)
 	defer server.Close()
@@ -111,9 +119,104 @@ func TestGetAddon(t *testing.T) {
 	assert.Equal(t, items[0].GetPath(), "terraform/metadata.yaml")
 }
 
+func TestRender(t *testing.T) {
+	testcases := []struct {
+		envs   []ObservabilityEnvironment
+		tmpl   string
+		expect string
+		err    error
+	}{
+		{
+			envs: []ObservabilityEnvironment{
+				{
+					Cluster: "c1",
+					Domain:  "a.com",
+				},
+				{
+					Cluster: "c2",
+					Domain:  "b.com",
+				},
+			},
+			tmpl: ObservabilityEnvBindingEnvTmpl,
+			expect: `
+        
+          
+          - name: c1
+            placement:
+              clusterSelector:
+                name: c1
+            patch:
+              components:
+                - name: grafana
+                  type: helm
+                  traits:
+                    - type: pure-ingress
+                      properties:
+                        domain: a.com
+          
+          - name: c2
+            placement:
+              clusterSelector:
+                name: c2
+            patch:
+              components:
+                - name: grafana
+                  type: helm
+                  traits:
+                    - type: pure-ingress
+                      properties:
+                        domain: b.com
+          
+        `,
+
+			err: nil,
+		},
+		{
+			envs: []ObservabilityEnvironment{
+				{
+					Cluster: "c1",
+					Domain:  "a.com",
+				},
+				{
+					Cluster: "c2",
+					Domain:  "b.com",
+				},
+			},
+			tmpl: ObservabilityWorkflow4EnvBindingTmpl,
+			expect: `
+
+  
+  - name: c1
+    type: deploy2env
+    properties:
+      policy: domain
+      env: c1
+      parallel: true
+  
+  - name: c2
+    type: deploy2env
+    properties:
+      policy: domain
+      env: c2
+      parallel: true
+  
+`,
+
+			err: nil,
+		},
+	}
+	for _, tc := range testcases {
+		t.Run("", func(t *testing.T) {
+			rendered, err := render(tc.envs, tc.tmpl)
+			assert.Equal(t, tc.err, err)
+			assert.Equal(t, tc.expect, rendered)
+		})
+	}
+}
+
 func TestRenderApp(t *testing.T) {
 	addon := baseAddon
-	app, err := RenderApp(&addon, nil, map[string]interface{}{})
+	app, err := RenderApp(ctx, nil, &addon, nil, map[string]interface{}{})
 	assert.NilError(t, err, "render app fail")
 	assert.Equal(t, len(app.Spec.Components), 2)
 }
@@ -131,7 +234,7 @@ func TestRenderDeploy2RuntimeAddon(t *testing.T) {
 	assert.Equal(t, def.GetAPIVersion(), "core.oam.dev/v1beta1")
 	assert.Equal(t, def.GetKind(), "TraitDefinition")
 
-	app, err := RenderApp(&addonDeployToRuntime, nil, map[string]interface{}{})
+	app, err := RenderApp(ctx, nil, &addonDeployToRuntime, nil, map[string]interface{}{})
 	assert.NilError(t, err)
 	steps := app.Spec.Workflow.Steps
 	assert.Check(t, len(steps) >= 2)
@@ -182,3 +285,95 @@ template: {
 	parameter: [string]: string
 }
 `
+
+func TestRenderApp4Observability(t *testing.T) {
+	k8sClient := fake.NewClientBuilder().Build()
+	testcases := []struct {
+		addon       Addon
+		args        map[string]interface{}
+		application string
+		err         error
+	}{
+		{
+			addon: Addon{
+				Meta: Meta{
+					Name: "observability",
+				},
+			},
+			args:        map[string]interface{}{},
+			application: "",
+			err:         ErrorNoDomain,
+		},
+		{
+			addon: Addon{
+				Meta: Meta{
+					Name: "observability",
+				},
+			},
+			args: map[string]interface{}{
+				"domain": "a.com",
+			},
+			application: `{"kind":"Application","apiVersion":"core.oam.dev/v1beta1","metadata":{"name":"addon-observability","namespace":"vela-system","creationTimestamp":null,"labels":{"addons.oam.dev/name":"observability"}},"spec":{"components":[],"policies":[{"name":"domain","type":"env-binding","properties":{"envs":null}}],"workflow":{"steps":[{"name":"deploy-control-plane","type":"apply-application-in-parallel"}]}},"status":{}}`,
+		},
+	}
+	for _, tc := range testcases {
+		t.Run("", func(t *testing.T) {
+			app, err := RenderApp(ctx, k8sClient, &tc.addon, nil, tc.args)
+			assert.Equal(t, tc.err, err)
+			if app != nil {
+				data, err := json.Marshal(app)
+				assert.NilError(t, err)
+				assert.Equal(t, tc.application, string(data))
+			}
+		})
+	}
+}
+
+// TestRenderApp4ObservabilityWithEnvBinding tests the case of RenderApp for Addon Observability with some Kubernetes data
+func TestRenderApp4ObservabilityWithK8sData(t *testing.T) {
+	k8sClient := fake.NewClientBuilder().Build()
+	ctx := context.Background()
+	secret1 := &corev1.Secret{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "test-secret",
+			Labels: map[string]string{
+				v1alpha12.LabelKeyClusterCredentialType: string(v1alpha12.CredentialTypeX509Certificate),
+			},
+		},
+		Data: map[string][]byte{
+			"test-key": []byte("test-value"),
+		},
+	}
+	err := k8sClient.Create(ctx, secret1)
+	assert.NilError(t, err)
+
+	testcases := []struct {
+		addon       Addon
+		args        map[string]interface{}
+		application string
+		err         error
+	}{
+		{
+			addon: Addon{
+				Meta: Meta{
+					Name: "observability",
+				},
+			},
+			args: map[string]interface{}{
+				"domain": "a.com",
+			},
+			application: `{"kind":"Application","apiVersion":"core.oam.dev/v1beta1","metadata":{"name":"addon-observability","namespace":"vela-system","creationTimestamp":null,"labels":{"addons.oam.dev/name":"observability"}},"spec":{"components":[],"policies":[{"name":"domain","type":"env-binding","properties":{"envs":[{"name":"test-secret","patch":{"components":[{"name":"grafana","traits":[{"properties":{"domain":"test-secret.a.com"},"type":"pure-ingress"}],"type":"helm"}]},"placement":{"clusterSelector":{"name":"test-secret"}}}]}}],"workflow":{"steps":[{"name":"deploy-control-plane","type":"apply-application-in-parallel"},{"name":"test-secret","type":"deploy2env","properties":{"env":"test-secret","parallel":true,"policy":"domain"}}]}},"status":{}}`,
+		},
+	}
+	for _, tc := range testcases {
+		t.Run("", func(t *testing.T) {
+			app, err := RenderApp(ctx, k8sClient, &tc.addon, nil, tc.args)
+			assert.Equal(t, tc.err, err)
+			if app != nil {
+				data, err := json.Marshal(app)
+				assert.NilError(t, err)
+				assert.Equal(t, tc.application, string(data))
+			}
+		})
+	}
+}

pkg/stdlib/op.cue

@@ -33,6 +33,21 @@ import (
 	} @step(2)
 }
 
+// This operator will dispatch all the components in parallel when applying an application.
+// Currently it works for Addon Observability to speed up the installation. It can also works for other applications, which
+// needs to skip health check for components.
+#ApplyApplicationInParallel: #Steps & {
+	load:       oam.#LoadComponetsInOrder @step(1)
+	components: #Steps & {
+		for name, c in load.value {
+			"\(name)": oam.#ApplyComponent & {
+				value:       c
+				waitHealthy: false
+			}
+		}
+	} @step(2)
+}
+
 #ApplyComponent: oam.#ApplyComponent
 
 #RenderComponent: oam.#RenderComponent

test/e2e-addon-test/addon_test.go

@@ -86,9 +86,9 @@ var _ = Describe("Addon tests", func() {
 		Expect(k8sClient.Delete(ctx, &ns, client.PropagationPolicy(metav1.DeletePropagationBackground))).Should(BeNil())
 	})
 
-	It("Addons Terraform is successfully enables and Terraform application works", func() {
+	It("Addon Terraform is successfully enabled and Terraform application works", func() {
 		By("Install Addon Terraform")
-		output, err := exec.Command("bash", "-c", "/tmp/vela addon enable terraform-alibaba  ALICLOUD_ACCESS_KEY=xxx ALICLOUD_SECRET_KEY=yyy ALICLOUD_REGION=cn-beijing").Output()
+		output, err := exec.Command("bash", "-c", "/tmp/vela addon enable terraform-alibaba ALICLOUD_ACCESS_KEY=xxx ALICLOUD_SECRET_KEY=yyy ALICLOUD_REGION=cn-beijing").Output()
 		var ee *exec.ExitError
 		if errors.As(err, &ee) {
 			fmt.Println("exit code error:", string(ee.Stderr))
@@ -121,4 +121,15 @@ var _ = Describe("Addon tests", func() {
 			},
 			time.Second*30, time.Millisecond*500).ShouldNot(BeNil())
 	})
+
+	PIt("Addon observability is successfully enabled", func() {
+		By("Install Addon Observability")
+		output, err := exec.Command("bash", "-c", "/tmp/vela addon enable observability domain=abc.com disk-size=20Gi").Output()
+		var ee *exec.ExitError
+		if errors.As(err, &ee) {
+			fmt.Println("exit code error:", string(ee.Stderr))
+		}
+		Expect(err).Should(BeNil())
+		Expect(string(output)).Should(ContainSubstring("Successfully enable addon:"))
+	})
 })

vela-templates/definitions/internal/workflowstep/apply-application-in-parallel.cue

@@ -0,0 +1,15 @@
+import (
+	"vela/op"
+)
+
+"apply-application-in-parallel": {
+	type: "workflow-step"
+	annotations: {}
+	labels: {
+		"ui-hidden": "true"
+	}
+	description: "Apply components of an application in parallel for your workflow steps"
+}
+template: {
+	output: op.#ApplyApplicationInParallel & {}
+}