Add storage extension to save filelog receivers' offsets

pkg/resources/collector/templates.go

@@ -19,7 +19,7 @@ var (
 filelog/k8s:
   include: [ /var/log/pods/*/*/*.log ]
   exclude: []
-  start_at: beginning
+  storage: file_storage
   include_file_path: true
   include_file_name: false
   operators:
@@ -87,7 +87,7 @@ filelog/k8s:
 	templateLogAgentRKE = `
 filelog/rke:
   include: [ /var/lib/rancher/rke/log/*.log ]
-  start_at: beginning
+  storage: file_storage
   include_file_path: true
   include_file_name: false
   operators:
@@ -106,7 +106,7 @@ journald/k3s:
 	templateKubeAuditLogs = template.Must(template.New("kubeauditlogsreceiver").Parse(`
 filelog/kubeauditlogs:
   include: [ {{ . }} ]
-  start_at: beginning
+  storage: file_storage
   include_file_path: false
   include_file_name: false
   operators:
@@ -157,7 +157,7 @@ journald/rke2:
   directory: {{ . }}
 filelog/rke2:
   include: [ /var/lib/rancher/rke2/agent/logs/kubelet.log ]
-  start_at: beginning
+  storage: file_storage
   include_file_path: true
   include_file_name: false
   operators:
@@ -230,7 +230,12 @@ processors:
       - key: tier
       - key: component
     {{ template "metrics-system-processor" . }}
+extensions:
+  file_storage:
+    directory: /var/otel/filestorage
+    timeout: 1s
 service:
+  extensions: [file_storage]
   telemetry:
     logs:
       level: {{ .LogLevel }}

pkg/resources/collector/workloads.go

@@ -22,16 +22,19 @@ import (
 )
 
 const (
-	receiversKey       = "receivers.yaml"
-	mainKey            = "config.yaml"
-	aggregatorKey      = "aggregator.yaml"
+	receiversKey  = "receivers.yaml"
+	mainKey       = "config.yaml"
+	aggregatorKey = "aggregator.yaml"
+
 	collectorImageRepo = "ghcr.io"
 	collectorImage     = "rancher-sandbox/opni-otel-collector"
-	collectorVersion   = "v0.1.2-0.74.0"
+	collectorVersion   = "v0.1.4-rc1-0.85.0"
 	reloaderImage      = "rancher-sandbox/config-reloader"
 	reloaderVersion    = "v0.1.2"
+
 	otelColBinaryName  = "otelcol-custom"
 	otelConfigDir      = "/etc/otel"
+	otelFileStorageDir = "/var/otel/filestorage"
 
 	otlpGRPCPort    = int32(4317)
 	rke2AgentLogDir = "/var/lib/rancher/rke2/agent/logs/"
@@ -213,6 +216,10 @@ func (r *Reconciler) daemonSet() resources.Resource {
 			Name:      "collector-config",
 			MountPath: otelConfigDir,
 		},
+		{
+			Name:      "filestorage-extension",
+			MountPath: otelFileStorageDir,
+		},
 	}
 	volumes := []corev1.Volume{
 		{
@@ -225,6 +232,15 @@ func (r *Reconciler) daemonSet() resources.Resource {
 				},
 			},
 		},
+		{
+			Name: "filestorage-extension",
+			VolumeSource: corev1.VolumeSource{
+				HostPath: &corev1.HostPathVolumeSource{
+					Path: otelFileStorageDir,
+					Type: &directoryOrCreate,
+				},
+			},
+		},
 	}
 
 	// Short circuit and remove the daemonset if we don't need it

pkg/resources/preprocessor/workloads.go

@@ -21,7 +21,7 @@ import (
 
 const (
 	configKey             = "config.yaml"
-	preprocessorVersion   = "v0.1.3-rc2-0.74.0"
+	preprocessorVersion   = "v0.1.4-rc1-0.85.0"
 	preprocessorImageRepo = "ghcr.io/rancher-sandbox"
 	preprocessorImage     = "opni-otel-collector"
 	otlpGRPCPort          = 4317
@@ -70,7 +70,6 @@ processors:
     log_statements:
     - context: log
       statements:
-      - merge_maps(attributes, body, "upsert") where attributes["filename"] == nil
       - set(attributes["COMM"], attributes["_COMM"])
       - delete_matching_keys(attributes, "^_.*")
     - context: resource

plugins/logging/pkg/agent/drivers/kubernetes_manager/kubernetes_manager.go

@@ -153,7 +153,7 @@ func (m *KubernetesManagerDriver) buildLoggingCollectorConfig() *opniloggingv1be
 		Spec: opniloggingv1beta1.CollectorConfigSpec{
 			Provider: opniloggingv1beta1.LogProvider(m.provider),
 			KubeAuditLogs: &opniloggingv1beta1.KubeAuditLogsSpec{
-				Enabled: true,
+				Enabled: false,
 			},
 		},
 	}