Add support for v0.9.0 (#9)

* Add support for v0.9.0 - Create processors directory when version is >= 0.9.0 - When version cannot be detected (i.e. not semver), processors path will not be added. - Allow for different conduit image to be passed. * fix typo
ConduitIO · Mar 18, 2024 · 5547c63 · 5547c63
1 parent 80df83f
commit 5547c63
Show file tree

Hide file tree

Showing 15 changed files with 242 additions and 55 deletions.
diff --git a/api/v1/conduit_types.go b/api/v1/conduit_types.go
@@ -38,47 +38,55 @@ var conduitConditions = NewConditionSet(
 )
 
 const (
-	ConduitVersion             = "0.8.0"
+	ConduitVersion             = "v0.9.0"
 	ConduitImage               = "ghcr.io/conduitio/conduit"
 	ConduitContainerName       = "conduit-server"
 	ConduitPipelinePath        = "/conduit.pipelines"
 	ConduitVolumePath          = "/conduit.storage"
 	ConduitDBPath              = "/conduit.storage/db"
 	ConduitConnectorsPath      = "/conduit.storage/connectors"
+	ConduitProcessorsPath      = "/conduit.storage/processors"
 	ConduitStorageVolumeMount  = "conduit-storage"
 	ConduitPipelineVolumeMount = "conduit-pipelines"
 	ConduitInitImage           = "golang:1.22"
-	ConduitInitContainerName   = "conduit-connector-init"
+	ConduitInitContainerName   = "conduit-init"
 )
 
-var ConduitPipelineFile = path.Join(ConduitPipelinePath, "pipeline.yaml")
+var (
+	ConduitPipelineFile          = path.Join(ConduitPipelinePath, "pipeline.yaml")
+	ConduitWithProcessorsVersion = "0.9.0"
+)
 
 // ConduitSpec defines the desired state of Conduit
 type ConduitSpec struct {
-	Running     bool                `json:"running,omitempty"`
-	Name        string              `json:"name,omitempty"`
-	Version     string              `json:"version,omitempty"`
-	Description string              `json:"description,omitempty"`
-	Connectors  []*ConduitConnector `json:"connectors,omitempty"`
-	Processors  []*ConduitProcessor `json:"processors,omitempty"`
+	Name        string `json:"name,omitempty"`
+	Description string `json:"description,omitempty"`
+	Image       string `json:"image,omitempty"`
+	Running     bool   `json:"running,omitempty"`
+	Version     string `json:"version,omitempty"`
+
+	Connectors []*ConduitConnector `json:"connectors,omitempty"`
+	Processors []*ConduitProcessor `json:"processors,omitempty"`
 }
 
 type ConduitConnector struct {
-	Name          string              `json:"name,omitempty"`
-	Plugin        string              `json:"plugin,omitempty"`
-	PluginPkg     string              `json:"pluginPkg,omitempty"`
-	PluginName    string              `json:"pluginName,omitempty"`
-	PluginVersion string              `json:"pluginVersion,omitempty"`
-	Type          string              `json:"type,omitempty"`
-	Settings      []SettingsVar       `json:"settings,omitempty"`
-	Processors    []*ConduitProcessor `json:"processors,omitempty"`
+	Name          string `json:"name,omitempty"`
+	Type          string `json:"type,omitempty"`
+	Plugin        string `json:"plugin,omitempty"`
+	PluginName    string `json:"pluginName,omitempty"`
+	PluginPkg     string `json:"pluginPkg,omitempty"`
+	PluginVersion string `json:"pluginVersion,omitempty"`
+
+	Settings   []SettingsVar       `json:"settings,omitempty"`
+	Processors []*ConduitProcessor `json:"processors,omitempty"`
 }
 
 type ConduitProcessor struct {
-	Name     string        `json:"name,omitempty"`
-	Type     string        `json:"type,omitempty"`
+	Name    string `json:"name,omitempty"`
+	Type    string `json:"type,omitempty"`
+	Workers int    `json:"workers,omitempty"`
+
 	Settings []SettingsVar `json:"settings,omitempty"`
-	Workers  int           `json:"workers,omitempty"`
 }
 
 type GlobalConfigMapRef struct {
@@ -132,15 +140,11 @@ type Conduit struct {
 
 func (r *Conduit) NamespacedName() types.NamespacedName {
 	return types.NamespacedName{
-		Name:      fmt.Sprintf("conduit-server-%s", r.Name),
+		Name:      fmt.Sprint("conduit-server-", r.Name),
 		Namespace: r.Namespace,
 	}
 }
 
-func (r *Conduit) ImageName() string {
-	return fmt.Sprintf("%s:v%s", ConduitImage, r.Spec.Version)
-}
-
 //+kubebuilder:object:root=true
 
 // ConduitList contains a list of Conduit

diff --git a/api/v1/conduit_webhook.go b/api/v1/conduit_webhook.go
@@ -32,6 +32,10 @@ func (r *Conduit) Default() {
 		}
 	}
 
+	if r.Spec.Image == "" {
+		r.Spec.Image = ConduitImage
+	}
+
 	if r.Spec.Version == "" {
 		r.Spec.Version = ConduitVersion
 	}

diff --git a/charts/conduit-operator/templates/crd.yaml b/charts/conduit-operator/templates/crd.yaml
@@ -159,6 +159,8 @@ spec:
                 type: array
               description:
                 type: string
+              image:
+                type: string
               name:
                 type: string
               processors:

diff --git a/config/crd/bases/operator.conduit.io_conduits.yaml b/config/crd/bases/operator.conduit.io_conduits.yaml
@@ -148,6 +148,8 @@ spec:
                 type: array
               description:
                 type: string
+              image:
+                type: string
               name:
                 type: string
               processors:

diff --git a/config/samples/conduit-generator-image-ver.yaml b/config/samples/conduit-generator-image-ver.yaml
@@ -0,0 +1,24 @@
+apiVersion: operator.conduit.io/v1
+kind: Conduit
+metadata:
+  name: conduit-generator
+spec:
+  running: true
+  name: generator.log
+  description: generator pipeline
+  image: ghcr.io/conduitio/conduit
+  version: v0.9.0
+  connectors:
+    - name: source-connector
+      type: source
+      plugin: builtin:generator
+      settings:
+        - name: format.type
+          value: structured
+        - name: format.options
+          value: "id:int,name:string,company:string,trial:bool"
+        - name: recordCount
+          value: "3"
+    - name: destination-connector
+      type: destination
+      plugin: builtin:log
diff --git a/config/samples/conduit-generator-secrets.yaml b/config/samples/conduit-generator-secrets.yaml
@@ -14,7 +14,7 @@ metadata:
 spec:
   running: true
   name: generator.standalone.log
-  description: generator pipline
+  description: generator pipeline
   connectors:
     - name: source-connector
       type: source

diff --git a/config/samples/conduit-generator.yaml b/config/samples/conduit-generator.yaml
@@ -5,7 +5,7 @@ metadata:
 spec:
   running: true
   name: generator.log
-  description: generator pipline
+  description: generator pipeline
   connectors:
     - name: source-connector
       type: source

diff --git a/config/samples/conduit-with-proccessors.yaml b/config/samples/conduit-with-proccessors.yaml
@@ -5,7 +5,7 @@ metadata:
 spec:
   running: true
   name: generator.proc.log
-  description: redacting generator pipline
+  description: redacting generator pipeline
   connectors:
     - name: source-connector
       type: source

diff --git a/config/samples/invalid-plugin.yaml b/config/samples/invalid-plugin.yaml
@@ -5,7 +5,7 @@ metadata:
 spec:
   running: true
   name: file-pipeline
-  description: test pipline
+  description: test pipeline
   connectors:
     - name: source-connector
       type: source

diff --git a/controllers/conduit_containers.go b/controllers/conduit_containers.go
@@ -77,6 +77,24 @@ func (c *commandBuilder) addConnectorBuild(b connectorBuild) {
 func ConduitInitContainers(cc []*v1.ConduitConnector) []corev1.Container {
 	builder := &commandBuilder{}
 
+	containers := []corev1.Container{
+		{
+			Name:            v1.ConduitInitContainerName,
+			Image:           v1.ConduitInitImage,
+			ImagePullPolicy: corev1.PullIfNotPresent,
+			Args: []string{
+				"sh", "-xe", "-c",
+				fmt.Sprintf("mkdir -p %s %s", v1.ConduitProcessorsPath, v1.ConduitConnectorsPath),
+			},
+			VolumeMounts: []corev1.VolumeMount{
+				{
+					Name:      v1.ConduitStorageVolumeMount,
+					MountPath: v1.ConduitVolumePath,
+				},
+			},
+		},
+	}
+
 	for _, c := range cc {
 		if strings.HasPrefix(c.Plugin, "builtin") {
 			continue
@@ -90,13 +108,9 @@ func ConduitInitContainers(cc []*v1.ConduitConnector) []corev1.Container {
 		})
 	}
 
-	if builder.empty() {
-		return []corev1.Container{}
-	}
-
-	return []corev1.Container{
-		{
-			Name:            v1.ConduitInitContainerName,
+	if !builder.empty() {
+		containers = append(containers, corev1.Container{
+			Name:            fmt.Sprint(v1.ConduitInitContainerName, "-connectors"),
 			Image:           v1.ConduitInitImage,
 			ImagePullPolicy: corev1.PullIfNotPresent,
 			Args: []string{
@@ -109,25 +123,33 @@ func ConduitInitContainers(cc []*v1.ConduitConnector) []corev1.Container {
 					MountPath: v1.ConduitVolumePath,
 				},
 			},
-		},
+		})
 	}
+
+	return containers
 }
 
 // ConduitRuntimeContainer returns a Kubernetes container definition
 // todo is the pipelineName supposed to be used?
-func ConduitRuntimeContainer(image string, envVars []corev1.EnvVar) corev1.Container {
+func ConduitRuntimeContainer(image, version string, envVars []corev1.EnvVar) corev1.Container {
+	args := []string{
+		"/app/conduit",
+		"-pipelines.path", v1.ConduitPipelineFile,
+		"-connectors.path", v1.ConduitConnectorsPath,
+		"-db.type", "badger",
+		"-db.badger.path", v1.ConduitDBPath,
+		"-pipelines.exit-on-error",
+	}
+
+	if withProcessors(version) {
+		args = append(args, "-processors.path", v1.ConduitProcessorsPath)
+	}
+
 	return corev1.Container{
 		Name:            v1.ConduitContainerName,
-		Image:           image,
+		Image:           fmt.Sprint(image, ":", version),
 		ImagePullPolicy: corev1.PullAlways,
-		Args: []string{
-			"/app/conduit",
-			"-pipelines.path", v1.ConduitPipelineFile,
-			"-connectors.path", v1.ConduitConnectorsPath,
-			"-db.type", "badger",
-			"-db.badger.path", v1.ConduitDBPath,
-			"-pipelines.exit-on-error",
-		},
+		Args:            args,
 		Ports: []corev1.ContainerPort{
 			{
 				Name:          "http",