feat: implemented Datasource through Cue template rendering

apis/monitor/v1alpha1/collectordatasource_types.go

@@ -37,16 +37,38 @@ const (
 	LogsDataSourceType    DataSourceType = "logs"
 )
 
+type DataSource struct {
+	// Name is the name of the data source
+	// +kubebuilder:validation:Required
+	Name string `json:"name"`
+
+	// Parameter is the parameter of the data source
+	Parameter *string `json:"parameter,omitempty"`
+}
+
 // CollectorDataSourceSpec defines the desired state of CollectorDataSource
 type CollectorDataSourceSpec struct {
 	// INSERT ADDITIONAL SPEC FIELDS - desired state of cluster
 	// Important: Run "make" to regenerate code after modifying this file
 
+	// Name is the name of the data source
+	// +kubebuilder:validation:Required
+	Name string `json:"name"`
+
 	// ExporterRef is the exporters to export data source
+	// +kubebuilder:validation:Required
 	ExporterRef `json:",inline"`
 
 	// Type is the type of the data source
+	// +kubebuilder:validation:Required
 	Type DataSourceType `json:"type"`
+
+	// CollectionInterval is the interval of the data source
+	CollectionInterval string `json:"collectionInterval,omitempty"`
+
+	// DataSourceList is the list of the data source
+	// +kubebuilder:validation:Required
+	DataSourceList []DataSource `json:"dataSourceList"`
 }
 
 // CollectorDataSourceStatus defines the observed state of CollectorDataSource

config/crd/bases/monitor.kubeblocks.io_collectordatasources.yaml

@@ -1,10 +1,11 @@
----
 apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   annotations:
     controller-gen.kubebuilder.io/version: v0.9.0
   creationTimestamp: null
+  labels:
+    app.kubernetes.io/name: kubeblocks
   name: collectordatasources.monitor.kubeblocks.io
 spec:
   group: monitor.kubeblocks.io
@@ -36,15 +37,37 @@ spec:
           spec:
             description: CollectorDataSourceSpec defines the desired state of CollectorDataSource
             properties:
+              collectionInterval:
+                description: CollectionInterval is the interval of the data source
+                type: string
+              dataSourceList:
+                description: DataSourceList is the list of the data source
+                items:
+                  properties:
+                    name:
+                      description: Name is the name of the data source
+                      type: string
+                    parameter:
+                      description: Parameter is the parameter of the data source
+                      type: string
+                  required:
+                  - name
+                  type: object
+                type: array
               exporterRef:
                 items:
                   type: string
                 type: array
+              name:
+                description: Name is the name of the data source
+                type: string
               type:
                 description: Type is the type of the data source
                 type: string
             required:
+            - dataSourceList
             - exporterRef
+            - name
             - type
             type: object
           status:

config/crd/bases/monitor.kubeblocks.io_logsexportersinks.yaml

@@ -1,10 +1,11 @@
----
 apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   annotations:
     controller-gen.kubebuilder.io/version: v0.9.0
   creationTimestamp: null
+  labels:
+    app.kubernetes.io/name: kubeblocks
   name: logsexportersinks.monitor.kubeblocks.io
 spec:
   group: monitor.kubeblocks.io

config/crd/bases/monitor.kubeblocks.io_metricsexportersinks.yaml

@@ -1,10 +1,11 @@
----
 apiVersion: apiextensions.k8s.io/v1
 kind: CustomResourceDefinition
 metadata:
   annotations:
     controller-gen.kubebuilder.io/version: v0.9.0
   creationTimestamp: null
+  labels:
+    app.kubernetes.io/name: kubeblocks
   name: metricsexportersinks.monitor.kubeblocks.io
 spec:
   group: monitor.kubeblocks.io

controllers/monitor/config/builder.go

@@ -0,0 +1,62 @@
+package monitor
+
+import (
+	"cuelang.org/go/cue"
+	"embed"
+	intctrlutil "github.com/apecloud/kubeblocks/internal/controllerutil"
+	"github.com/leaanthony/debme"
+	"gopkg.in/yaml.v3"
+)
+
+var (
+	//go:embed cue/*
+	cueTemplates embed.FS
+	cacheCtx     = map[string]interface{}{}
+)
+
+func getCacheCUETplValue(key string, valueCreator func() (*intctrlutil.CUETpl, error)) (*intctrlutil.CUETpl, error) {
+	vIf, ok := cacheCtx[key]
+	if ok {
+		return vIf.(*intctrlutil.CUETpl), nil
+	}
+	v, err := valueCreator()
+	if err != nil {
+		return nil, err
+	}
+	cacheCtx[key] = v
+	return v, err
+}
+
+func buildFromCUEForOTel(tplName string, fillMap map[string]any, lookupKey string) ([]byte, error) {
+	cueFS, _ := debme.FS(cueTemplates, "cue")
+	cueTpl, err := getCacheCUETplValue(tplName, func() (*intctrlutil.CUETpl, error) {
+		return intctrlutil.NewCUETplFromBytes(cueFS.ReadFile(tplName))
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	cueValue := intctrlutil.NewCUEBuilder(*cueTpl)
+
+	for k, v := range fillMap {
+		if err := cueValue.FillObj(k, v); err != nil {
+			return nil, err
+		}
+	}
+
+	value := cueValue.Value.LookupPath(cue.ParsePath(lookupKey))
+	bytes, err := value.MarshalJSON()
+	if err != nil {
+		return nil, err
+	}
+
+	var jsonObj interface{}
+	if err := yaml.Unmarshal(bytes, &jsonObj); err != nil {
+		return nil, err
+	}
+	yamlBytes, err := yaml.Marshal(jsonObj)
+	if err != nil {
+		return nil, err
+	}
+	return yamlBytes, nil
+}

controllers/monitor/config/builder_test.go

@@ -0,0 +1,39 @@
+package monitor
+
+import (
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	yamlv2 "gopkg.in/yaml.v2"
+	"gopkg.in/yaml.v3"
+)
+
+var _ = Describe("monitor_controller", func() {
+
+	It("should generate config correctly from config yaml", func() {
+		Eventually(func(g Gomega) {
+			valMap := map[string]any{
+				"meta.transport": "http",
+				//"meta.allow_native_password": false,
+				//"meta.endpoint":              "http://",
+				"meta.password": "labels[\"pass\"]",
+			}
+			tplName := "test.cue"
+			bytes, err := buildFromCUEForOTel(tplName, valMap, "output")
+			node := yaml.Node{}
+			err = node.Encode(bytes)
+
+			Expect(err).ShouldNot(HaveOccurred())
+			slice := yamlv2.MapSlice{}
+			err = yamlv2.Unmarshal(bytes, &slice)
+			Expect(err).ShouldNot(HaveOccurred())
+			Expect(bytes).ShouldNot(BeNil())
+
+		}).Should(Succeed())
+	})
+
+	It("should generate config correctly from config yaml", func() {
+		Eventually(func(g Gomega) {
+
+		}).Should(Succeed())
+	})
+})

controllers/monitor/config/cue/exporter/loki.cue

@@ -0,0 +1,9 @@
+parameters: {
+	endpoint: ""
+}
+
+output:
+  loki:
+  	endpoint: parameters.endpoint
+
+

controllers/monitor/config/cue/extension/apecloud_engine_observer.cue

@@ -0,0 +1,15 @@
+parameters: {
+	pod_observer: *"apecloud_k8s_observer" | string
+  container_observer: *"runtime_container" | string
+  scraper_config_file: *"/opt/oteld/etc/scraper.yaml" | string
+}
+
+
+
+output:
+  apecloud_engine_observer: {
+	  pod_observer: parameters.pod_observer
+	  container_observer: parameters.container_observer
+	  scraper_config_file: parameters.scraper_config_file
+  }
+

controllers/monitor/config/cue/extension/runtime_container.cue

@@ -0,0 +1,13 @@
+parameters: {
+	enable: *true | bool
+  container_runtime_type: *"containerd" | string
+}
+
+
+output:
+  runtime_contaienr: {
+	enable: parameters.enable
+  container_runtime_type: parameters.container_runtime_type
+  }
+
+

controllers/monitor/config/cue/receiver/logs/mysql.cue

@@ -0,0 +1,15 @@
+output: {
+	"filelog/mysql/error": {
+		type:
+		include: [ "/data/mysql/log/mysqld-error.log"]
+		include_file_name: false
+		start_at: "beginning"
+	}
+
+	"filelog/mysql/slow":{
+		type: "slow"
+		include:[ "/data/mysql/log/mysqld-slowquery.log"]
+		include_file_name: false
+		start_at: "beginning"
+	}
+}

controllers/monitor/config/cue/receiver/metrics/apecloudkubeletstats.cue

@@ -0,0 +1,21 @@
+parameters: {
+	auth_type: *"service_account" | string
+	collection_interval: *"10s" | string
+	endpoint: *"http://localhost:10255" | string
+}
+
+output:
+  apecloudkubeletstats: {
+	  auth_type: parameters.auth_type
+    collection_interval: parameters.collection_interval
+    endpoint: parameters.endpoint
+	  extra_metadata_labels:
+	  	- k8s.volume.type
+	  	- kubeblocks
+	  metric_groups:
+	  	- container
+	  	- pod
+  		- volume
+  }
+
+

controllers/monitor/config/cue/receiver/metrics/apecloudmysql.cue

@@ -0,0 +1,19 @@
+parameters: {
+	username: *"${env:MYSQL_USER}" | string
+  password: *"${env:MYSQL_PASSWORD}" | string
+  allow_native_passwords: *true | bool
+  transport: *"tcp" | string
+  collection_interval: *"${env:COLLECTION_INTERVAL}" | string
+}
+
+
+
+output:
+  apecloudmysql: {
+    username: parameters.username
+    password: parameters.password
+    allow_native_passwords: parameters.allow_native_passwords
+    transport: parameters.transport
+    collection_interval: parameters.collection_interval
+  }
+

controllers/monitor/config/cue/receiver/metrics/apecloudnode.cue

@@ -0,0 +1,10 @@
+parameters: {
+	collection_interval: *"15s" | string
+}
+
+
+output:
+  apecloudnode: {
+  	collection_interval: parameters.collection_interval
+  }
+

controllers/monitor/config/cue/receiver/metrics/k8s_cluster.cue

@@ -0,0 +1,10 @@
+parameters: {
+	collection_interval: *"15s" | string
+}
+
+
+output:
+	k8s_cluster: {
+		collection_interval: parameters.collection_interval
+	}
+

controllers/monitor/config/cue/test.cue

@@ -0,0 +1,18 @@
+#meta: {
+	endpoint: *"test" | string
+	username: *"labels[\"conn_username\"]" | string
+	password: *"labels[\"conn_password\"]" | string
+	collection_interval: *"15s" | string
+	transport: *"tcp" | string
+}
+
+output:
+	apecloudmysql: {
+		endpoint: parameter.endpoint
+		username: parameter.username
+		password: parameter.password
+		collection_interval: parameter.collection_interval
+		transport: parameter.transport
+	}
+
+parameter: #meta