update vector config

Author: soerenschneider

apps/vector/daemonset.yaml

@@ -94,7 +94,7 @@ spec:
             name: "vector-config"
         - name: data
           hostPath:
-            path: "/var/lib/vector"
+            path: "/var/lib/vector-k8s"
         - hostPath:
             path: "/var/log/"
           name: "var-log"

clusters/rs.soeren.cloud/monitoring/victoriametrics/virtualservice.yaml

@@ -5,6 +5,16 @@ namespace: vector
 resources:
   - ../../../apps/vector
   - namespace.yaml
+patches:
+  - target:
+      kind: DaemonSet
+      name: vector
+    patch: |-
+      - op: add
+        path: /spec/template/spec/containers/0/env/-
+        value:
+          name: "DC"
+          value: "dd"
 configMapGenerator:
   - name: vector-config
     behavior: merge

clusters/svc.dd.soeren.cloud/vector/kustomization.yaml

@@ -1,18 +1,52 @@
 ---
+transforms:
+  # Parse and enrich journald logs
+  k8s_parser:
+    type: "remap"
+    inputs: ["k8s"]
+    source: |
+      .dc = get_env_var("DC") ?? ""
+      .cluster = get_env_var("CLUSTER") ?? ""
+      
+      #.app_name = replace!(to_string!(.unit_name) || "unknown", r'\.(service|timer|socket)', "")
+
+      if exists(.message) {
+        message_str = string(.message) ?? ""
+
+        # Try JSON first
+        json_result = parse_json(message_str) ?? {}
+        if exists(json_result.level) && is_string(json_result.level) {
+          .log_level = downcase!(json_result.level)
+        } else {
+          # Comprehensive regex for common log level patterns
+          level_match = parse_regex(message_str, r'(?i)(?:\[|<|\(|^|\s|level[=:]\s*|severity[=:]\s*|\|)(?P<level>info|error|warn|warning|debug|trace|fatal|critical)(?:\]|>|\)|[:\s]|\||$)') ?? {}
+
+          if exists(level_match.level) {
+            found_level = downcase!(level_match.level)
+
+            # Normalize variations
+            .log_level = found_level
+          }
+        }
+      }
+
+
 sinks:
   prom_exporter:
     type: "prometheus_exporter"
     inputs: ["host_metrics", "internal_metrics"]
     address: "0.0.0.0:9090"
-  loki:
-    type: "loki"
-    inputs: ["k8s"]
-    encoding:
-      codec: "json"
-    endpoint: "http://loki.loki:3100"
-    out_of_order_action: "accept"
-    tenant_id: "soeren"
-    labels:
-      datacenter: "dd"
-      cluster: "svc.dd.soeren.cloud"
-      app: "{{ .app }}"
+
+  vlogs:
+    inputs: ["k8s_parser"]
+    type: "elasticsearch"
+    endpoints:
+      - https://logs.rs.soeren.cloud/insert/elasticsearch/
+    api_version: v8
+    compression: gzip
+    healthcheck:
+      enabled: false
+    query:
+      _msg_field: message
+      _time_field: timestamp
+      _stream_fields: kubernetes.pod_node_name,kubernetes.pod_namespace,kubernetes.pod_name,kubernetes.container_name,dc,cluster

clusters/svc.dd.soeren.cloud/vector/sinks.yaml

@@ -5,6 +5,16 @@ namespace: vector
 resources:
   - ../../../apps/vector
   - namespace.yaml
+patches:
+  - target:
+      kind: DaemonSet
+      name: vector
+    patch: |-
+      - op: add
+        path: /spec/template/spec/containers/0/env/-
+        value:
+          name: "DC"
+          value: "ez"
 configMapGenerator:
   - name: vector-config
     behavior: merge

clusters/svc.ez.soeren.cloud/grafana/virtualservice.yaml

@@ -1,18 +1,52 @@
 ---
+transforms:
+  # Parse and enrich journald logs
+  k8s_parser:
+    type: "remap"
+    inputs: ["k8s"]
+    source: |
+      .dc = get_env_var("DC") ?? ""
+      .cluster = get_env_var("CLUSTER") ?? ""
+      
+      #.app_name = replace!(to_string!(.unit_name) || "unknown", r'\.(service|timer|socket)', "")
+
+      if exists(.message) {
+        message_str = string(.message) ?? ""
+
+        # Try JSON first
+        json_result = parse_json(message_str) ?? {}
+        if exists(json_result.level) && is_string(json_result.level) {
+          .log_level = downcase!(json_result.level)
+        } else {
+          # Comprehensive regex for common log level patterns
+          level_match = parse_regex(message_str, r'(?i)(?:\[|<|\(|^|\s|level[=:]\s*|severity[=:]\s*|\|)(?P<level>info|error|warn|warning|debug|trace|fatal|critical)(?:\]|>|\)|[:\s]|\||$)') ?? {}
+
+          if exists(level_match.level) {
+            found_level = downcase!(level_match.level)
+
+            # Normalize variations
+            .log_level = found_level
+          }
+        }
+      }
+
+
 sinks:
   prom_exporter:
     type: "prometheus_exporter"
     inputs: ["host_metrics", "internal_metrics"]
     address: "0.0.0.0:9090"
-  loki:
-    type: "loki"
-    inputs: ["k8s"]
-    encoding:
-      codec: "json"
-    endpoint: "http://loki.loki:3100"
-    out_of_order_action: "accept"
-    tenant_id: "soeren"
-    labels:
-      datacenter: "ez"
-      cluster: "svc.ez.soeren.cloud"
-      app: "{{ .app }}"
+
+  vlogs:
+    inputs: ["k8s_parser"]
+    type: "elasticsearch"
+    endpoints:
+      - https://logs.rs.soeren.cloud/insert/elasticsearch/
+    api_version: v8
+    compression: gzip
+    healthcheck:
+      enabled: false
+    query:
+      _msg_field: message
+      _time_field: timestamp
+      _stream_fields: kubernetes.pod_node_name,kubernetes.pod_namespace,kubernetes.pod_name,kubernetes.container_name,dc,cluster