Merge pull request #38 from mindwm/neo4j-cdc-vector-knative

Add vector helm chart deployment with kafka basic config to context x… Fixes #14
mindwm · Jun 11, 2024 · f649611 · f649611
2 parents a3d863b + eb1c5b1
commit f649611
Show file tree

Hide file tree

Showing 3 changed files with 135 additions and 2 deletions.
diff --git a/config.k b/config.k
@@ -155,6 +155,17 @@ neo4j = {
     }
 }
 
+vector = {
+    image = {
+        repo = "ghcr.io/mindwm/vector-image/vector-image"
+        tag = "main"
+    } 
+    chart_version = "0.33.0"
+    crossplane_name = lambda context_name : str -> str {
+        context_name + delimiter + "vector"
+    }
+} 
+
 kiali_operator = {
     namespace = "kiali-operator"
     release_name = "kiali-operator"

diff --git a/crossplane/xrd_kcl_function.k b/crossplane/xrd_kcl_function.k
@@ -223,6 +223,128 @@ items = [
       }
     }
   }
+  {
+    apiVersion = "helm.crossplane.io/v1beta1"
+    kind = "Release"
+    metadata.name = vector.crossplane_name("{{ context_name }}")
+    spec = {
+      forProvider = {
+        chart = {
+          name = "${charts.vector.name}"
+          repository = "${charts.vector.repoURL}"
+          version = "${config.vector.chart_version}"
+        }
+        namespace = makeContextNamespace("{{ context_name }}")
+        values = {
+            image = { 
+                # TODO(@metacoma) parametrize
+                repository = "${config.vector.image.repo}"
+                tag = "${config.vector.image.tag}"
+            }
+            serviceHeadless.enalbed = False
+            customConfig = {
+                transforms = {
+                    cloudevent = {
+                        type = "lua"
+                        version = "2" 
+                        inputs = ["neo4j-cdc"] 
+                        hooks = {
+                            process = "process"
+                            init = "seed_init" 
+                        } 
+                        source = '''
+        function seed_init() 
+          uuid = require("uuid")
+          local socket = require("socket")  -- gettime() has higher precision than os.time()
+          uuid.seed()
+        end
+        function process(event, emit)
+          local cloudevent = {}
+          cloudevent.log = {
+            id = uuid(),
+            data = event.log,
+            source = "graph." .. event.log["payload"]["type"] ,
+            specversion = "1.0",
+            datacontenttype = "application/json",
+            type = event.log["meta"]["operation"],
+            time = os.date("!%Y-%m-%dT%H:%M:%SZ", os.time()),
+
+            traceparent = "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",
+            tracestate = "rojo=00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01,congo=lZWRzIHRoNhcm5hbCBwbGVhc3VyZS4"
+          }
+          if (cloudevent.log.type == "created") then
+            if (event.log["payload"]["type"] == "node") then
+              if event.log["payload"]["after"]["labels"] then
+                cloudevent.log["subject"] = table.concat(event.log["payload"]["after"]["labels"], ",")
+                else
+                  cloudevent.log["subject"] = "unknown_node"
+              end
+            end
+            if (event.log["payload"]["type"] == "relationship") then
+              if (event.log["payload"]["label"]) then
+                  cloudevent.log["subject"] = event.log["payload"]["label"]
+                else
+                  cloudevent.log["subject"] = "unknown_relationship"
+              end
+            end
+          end
+          emit(cloudevent)
+        end
+'''
+                    } 
+                } 
+                sources = {
+                    "neo4j-cdc" = {
+                        type = "kafka"
+                        bootstrap_servers: "${config.kafka.brokerEndpoint}" 
+                        group_id = "{{ context_name }}-vector"
+                        topics = [ "{{ context_name }}-cdc-topic" ]
+                        decoding.codec = "json"
+
+                        librdkafka_options = {
+                            "api.version.request" = 'true'
+                            "partition.assignment.strategy" = "roundrobin"
+                            "log_level" = "6"
+                            "session.timeout.ms" = "10000"
+                            "max.poll.interval.ms" = "300000"
+                            "socket.nagle.disable" = "false"
+                            "socket.keepalive.enable" = "true"
+                            "socket.max.fails" = "3"
+                            "fetch.min.bytes" = "1000000"
+                        }
+
+                    } 
+                } 
+                sinks = {
+                    console = {
+                        type = "console"
+                        inputs = [ "neo4j-cdc" ]
+                        encoding.codec = "json"
+                    } 
+                    "context-broker" = {
+                        type = "http"
+                        inputs = [ "cloudevent" ]
+                        encoding.codec = "json"
+                        method = "post"
+                        batch = {
+                            max_size = 1
+                            max_events = 1
+                        } 
+                        framing.method = "newline_delimited"
+                        uri = "http://broker-ingress.${config.knative.eventing.namespace}.svc.cluster.local/" +  makeContextNamespace("{{ context_name }}") + "/" + contextBroker.name("{{ context_name }}")
+                        request = {
+                            concurrency = 1
+                            headers = {
+                                "Content-Type" = "application/cloudevents+json"
+                            }
+                        }
+                    } 
+                } 
+            } 
+        }
+      }
+    }
+  }
   {
     apiVersion = "kubernetes.crossplane.io/v1alpha2"
     kind = "Object"

diff --git a/kcl.mod b/kcl.mod
@@ -3,11 +3,11 @@ name = "mindwm-gitops"
 version = "0.1.0"
 
 [dependencies]
-json_merge_patch = { oci = "oci://ghcr.io/kcl-lang/json_merge_patch", tag = "0.1.0" }
 argo-cd-order = { oci = "oci://ghcr.io/kcl-lang/argo-cd-order", tag = "0.2.0" }
 knative-operator = { oci = "oci://ghcr.io/kcl-lang/knative-operator", tag = "0.1.0" }
 k8s = { oci = "oci://ghcr.io/kcl-lang/k8s", tag = "1.29" }
 argoproj = { oci = "oci://ghcr.io/kcl-lang/argoproj", tag = "0.1.0" }
-
+json_merge_patch = { oci = "oci://ghcr.io/kcl-lang/json_merge_patch", tag = "0.1.0" }
 [profile]
 entries = ["main.k"]
+