added dataPath and logsPath, store pq data and checkpoints at dataPath by default

Author: AndrewChubatiuk

api/operator/v1/vlagent_types.go

@@ -95,8 +95,9 @@ type VLAgentK8sCollector struct {
 	// By default VLAgent collects logs from /var/log/containers
 	LogsPath string `json:"logsPath,omitempty"`
 
-	// CheckpointsPath configures path where logs checkpoints are stored
-	CheckpointsPath string `json:"checkpointsPath,omitempty"`
+	// DataPath configures host path where logs checkpoints are stored.
+	// By default it also contains persistent queue
+	DataPath string `json:"checkpointsPath,omitempty"`
 
 	// TenantID defines default tenant ID to use for logs collected from pods in format: <accountID>:<projectID>
 	TenantID string `json:"tenantID,omitempty"`

config/base-with-webhook/kustomization.yaml

@@ -80,3 +80,9 @@ replacements:
       group: cert-manager.io
       kind: Certificate
       version: v1
+apiVersion: kustomize.config.k8s.io/v1beta1
+kind: Kustomization
+images:
+- name: manager
+  newName: localhost:5001/victoriametrics/operator
+  newTag: heads-vlagent-logs-collection-0-gd8757fec-dirty-60c84815

config/crd/overlay/crd.yaml

@@ -741,8 +741,9 @@ spec:
                   K8s pods
                 properties:
                   checkpointsPath:
-                    description: CheckpointsPath configures path where logs checkpoints
-                      are stored
+                    description: |-
+                      DataPath configures host path where logs checkpoints are stored.
+                      By default it also contains persistent queue
                     type: string
                   decolorizeFields:
                     description: DecolorizeFields defines fields to remove ANSI color

config/examples/vlagent-collector.yaml

@@ -3,21 +3,14 @@ kind: VLAgent
 metadata:
   name: example
 spec:
-  replicaCount: 2
   resources:
     requests:
       cpu: "50m"
       memory: "350Mi"
     limits:
       cpu: "500m"
       memory: "850Mi"
-  persistentVolumeClaimRetentionPolicy:
-    whenDeleted: Delete
   k8sCollector:
     enabled: true
   remoteWrite:
     - url: "http://vlsingle-example-0.default.svc:9428/internal/insert"
-      maxDiskUsage: 10GB
-  remoteWriteSettings:
-    # ~ 5GB in bytes
-    maxBlockSize: 30MB

docs/api.md

@@ -180,7 +180,7 @@ Appears in: [VLAgentSpec](#vlagentspec)
 
 | Field | Description |
 | --- | --- |
-| checkpointsPath<a href="#vlagentk8scollector-checkpointspath" id="vlagentk8scollector-checkpointspath">#</a><br/>_string_ | _(Required)_<br/>CheckpointsPath configures path where logs checkpoints are stored |
+| checkpointsPath<a href="#vlagentk8scollector-checkpointspath" id="vlagentk8scollector-checkpointspath">#</a><br/>_string_ | _(Required)_<br/>DataPath configures host path where logs checkpoints are stored.<br />By default it also contains persistent queue |
 | decolorizeFields<a href="#vlagentk8scollector-decolorizefields" id="vlagentk8scollector-decolorizefields">#</a><br/>_string array_ | _(Required)_<br/>DecolorizeFields defines fields to remove ANSI color codes across logs ingested from Kubernetes |
 | enabled<a href="#vlagentk8scollector-enabled" id="vlagentk8scollector-enabled">#</a><br/>_boolean_ | _(Required)_<br/>Enabled switches VLAgent to log collection mode.<br />Note, for this purpose operator uses DaemonSet, while by default VLAgent uses StatefulSet.<br />It means that switching this option will drop all persisted data. |
 | extraFields<a href="#vlagentk8scollector-extrafields" id="vlagentk8scollector-extrafields">#</a><br/>_string_ | _(Required)_<br/>ExtraFields defines extra fields to add to each collected log line |

internal/controller/operator/factory/finalize/vlagent.go

@@ -5,15 +5,24 @@ import (
 
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
+	rbacv1 "k8s.io/api/rbac/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	vmv1 "github.com/VictoriaMetrics/operator/api/operator/v1"
+	"github.com/VictoriaMetrics/operator/internal/config"
 )
 
 // OnVLAgentDelete deletes all vlagent related resources
 func OnVLAgentDelete(ctx context.Context, rclient client.Client, cr *vmv1.VLAgent) error {
-	if err := removeFinalizeObjByName(ctx, rclient, &appsv1.StatefulSet{}, cr.PrefixedName(), cr.Namespace); err != nil {
-		return err
+	if cr.Spec.K8sCollector.Enabled {
+		if err := removeFinalizeObjByName(ctx, rclient, &appsv1.DaemonSet{}, cr.PrefixedName(), cr.Namespace); err != nil {
+			return err
+		}
+	} else {
+		if err := removeFinalizeObjByName(ctx, rclient, &appsv1.StatefulSet{}, cr.PrefixedName(), cr.Namespace); err != nil {
+			return err
+		}
 	}
 	// check service
 	if err := removeFinalizeObjByName(ctx, rclient, &corev1.Service{}, cr.PrefixedName(), cr.Namespace); err != nil {
@@ -39,5 +48,20 @@ func OnVLAgentDelete(ctx context.Context, rclient client.Client, cr *vmv1.VLAgen
 	if err := deleteSA(ctx, rclient, cr); err != nil {
 		return err
 	}
+	if config.IsClusterWideAccessAllowed() {
+		if err := removeFinalizeObjByName(ctx, rclient, &rbacv1.ClusterRoleBinding{}, cr.GetClusterRoleName(), cr.GetNamespace()); err != nil {
+			return err
+		}
+		if err := removeFinalizeObjByName(ctx, rclient, &rbacv1.ClusterRole{}, cr.GetClusterRoleName(), cr.GetNamespace()); err != nil {
+			return err
+		}
+		if err := SafeDelete(ctx, rclient, &rbacv1.ClusterRoleBinding{ObjectMeta: metav1.ObjectMeta{Name: cr.GetClusterRoleName(), Namespace: cr.GetNamespace()}}); err != nil {
+			return err
+		}
+
+		if err := SafeDelete(ctx, rclient, &rbacv1.ClusterRole{ObjectMeta: metav1.ObjectMeta{Name: cr.GetClusterRoleName(), Namespace: cr.GetNamespace()}}); err != nil {
+			return err
+		}
+	}
 	return nil
 }

internal/controller/operator/factory/vlagent/rbac.go

@@ -35,7 +35,7 @@ var (
 
 // createK8sAPIAccess - creates RBAC access rules for vlagent
 func createK8sAPIAccess(ctx context.Context, rclient client.Client, cr, prevCR *vmv1.VLAgent) error {
-	if config.IsClusterWideAccessAllowed() {
+	if !config.IsClusterWideAccessAllowed() {
 		logger.WithContext(ctx).Info(fmt.Sprintf("skipping cluster role and binding for vlagent=%s/%s since operator has WATCH_NAMESPACE set", cr.Namespace, cr.Name))
 		return nil
 	}

internal/controller/operator/factory/vlagent/vlagent.go

@@ -31,6 +31,10 @@ const (
 	persistentQueueSTSDir    = "/vlagent_pq/vlagent-remotewrite-data"
 	persistentQueueMountName = "persistent-queue-data"
 
+	defaultLogsPath = "/var/log/containers"
+	defaultDataPath = "/var/lib/vlagent"
+	dataVolumeName  = "data"
+
 	remoteWriteAssetsMounthPath = "/etc/vl/remote-write-assets"
 	tlsServerConfigMountPath    = "/etc/vl/tls-server-secrets"
 )
@@ -78,9 +82,6 @@ func CreateOrUpdate(ctx context.Context, cr *vmv1.VLAgent, rclient client.Client
 	if cr.ParsedLastAppliedSpec != nil {
 		prevCR = cr.DeepCopy()
 		prevCR.Spec = *cr.ParsedLastAppliedSpec
-		if err := deleteOrphaned(ctx, rclient, cr); err != nil {
-			return fmt.Errorf("cannot delete objects from prev state: %w", err)
-		}
 	}
 	if cr.IsOwnsServiceAccount() {
 		var prevSA *corev1.ServiceAccount
@@ -293,54 +294,70 @@ func newPodSpec(cr *vmv1.VLAgent) (*corev1.PodSpec, error) {
 		if len(cr.Spec.K8sCollector.ExtraFields) > 0 {
 			args = append(args, fmt.Sprintf("-kubernetesCollector.extraField=%q", cr.Spec.K8sCollector.ExtraFields))
 		}
-		if len(cr.Spec.K8sCollector.CheckpointsPath) > 0 {
-			args = append(args, fmt.Sprintf("-kubernetesCollector.checkpointsPath=%q", cr.Spec.K8sCollector.CheckpointsPath))
-		}
-		if len(cr.Spec.K8sCollector.LogsPath) > 0 {
-			args = append(args, fmt.Sprintf("-kubernetesCollector.logsPath=%q", cr.Spec.K8sCollector.LogsPath))
-		}
-		volumes = append(volumes, corev1.Volume{
-			Name: "varlog",
-			VolumeSource: corev1.VolumeSource{
-				HostPath: &corev1.HostPathVolumeSource{
-					Path: "/var/log",
-				},
-			},
-		}, corev1.Volume{
-			Name: "varlib",
-			VolumeSource: corev1.VolumeSource{
-				HostPath: &corev1.HostPathVolumeSource{
-					Path: "/var/lib",
+
+		if len(cr.Spec.K8sCollector.LogsPath) == 0 || cr.Spec.K8sCollector.LogsPath == defaultLogsPath {
+			logVolumeName := "varlog"
+			logVolumePath := "/var/log"
+			libVolumeName := "varlib"
+			libVolumePath := "/var/lib"
+			volumes = append(volumes, corev1.Volume{
+				Name: logVolumeName,
+				VolumeSource: corev1.VolumeSource{
+					HostPath: &corev1.HostPathVolumeSource{
+						Path: logVolumePath,
+					},
 				},
-			},
-		}, corev1.Volume{
-			Name: "vl-collector-data",
-			VolumeSource: corev1.VolumeSource{
-				HostPath: &corev1.HostPathVolumeSource{
-					Path: "/var/lib/vl-collector",
+			}, corev1.Volume{
+				Name: libVolumeName,
+				VolumeSource: corev1.VolumeSource{
+					HostPath: &corev1.HostPathVolumeSource{
+						Path: libVolumePath,
+					},
 				},
-			},
-		})
-		if cr.Spec.RemoteWriteSettings == nil || cr.Spec.RemoteWriteSettings.TmpDataPath == nil {
+			})
+			agentVolumeMounts = append(agentVolumeMounts, corev1.VolumeMount{
+				Name:      logVolumeName,
+				MountPath: logVolumePath,
+				ReadOnly:  true,
+			}, corev1.VolumeMount{
+				Name:      libVolumeName,
+				MountPath: libVolumePath,
+				ReadOnly:  true,
+			})
+		} else {
+			args = append(args, fmt.Sprintf("-kubernetesCollector.logsPath=%q", cr.Spec.K8sCollector.LogsPath))
+		}
+		dataPath := defaultDataPath
+		var pqSrc corev1.VolumeSource
+		if len(cr.Spec.K8sCollector.DataPath) == 0 || cr.Spec.K8sCollector.DataPath == defaultDataPath {
 			volumes = append(volumes, corev1.Volume{
-				Name: persistentQueueMountName,
+				Name: dataVolumeName,
 				VolumeSource: corev1.VolumeSource{
-					EmptyDir: &corev1.EmptyDirVolumeSource{},
+					HostPath: &corev1.HostPathVolumeSource{
+						Path: dataPath,
+					},
 				},
 			})
+			agentVolumeMounts = append(agentVolumeMounts, corev1.VolumeMount{
+				Name:      dataVolumeName,
+				MountPath: dataPath,
+			})
+			pqSrc.HostPath = &corev1.HostPathVolumeSource{
+				Path: path.Join(dataPath, persistentQueueSTSDir),
+			}
+		} else {
+			dataPath = cr.Spec.K8sCollector.DataPath
+			pqSrc.EmptyDir = &corev1.EmptyDirVolumeSource{}
+		}
+		checkpointsPath := path.Join(dataPath, "checkpoints.json")
+		args = append(args, fmt.Sprintf("-kubernetesCollector.checkpointsPath=%q", checkpointsPath))
+
+		if cr.Spec.RemoteWriteSettings == nil || cr.Spec.RemoteWriteSettings.TmpDataPath == nil {
+			volumes = append(volumes, corev1.Volume{
+				Name:         persistentQueueMountName,
+				VolumeSource: pqSrc,
+			})
 		}
-		agentVolumeMounts = append(agentVolumeMounts, corev1.VolumeMount{
-			Name:      "varlog",
-			MountPath: "/var/log",
-			ReadOnly:  true,
-		}, corev1.VolumeMount{
-			Name:      "varlib",
-			MountPath: "/var/lib",
-			ReadOnly:  true,
-		}, corev1.VolumeMount{
-			Name:      "vl-collector-data",
-			MountPath: "/vl-collector",
-		})
 	}
 
 	var envs []corev1.EnvVar

internal/controller/operator/factory/vlagent/vlagent_test.go

@@ -1081,6 +1081,7 @@ containers:
     args:
       - -httpListenAddr=:9425
       - -kubernetesCollector
+      - -kubernetesCollector.checkpointsPath="/var/lib/vlagent/checkpoints.json"
       - -kubernetesCollector.msgField="msg,message"
       - -remoteWrite.maxDiskUsagePerURL=10GB,10GB,
       - -remoteWrite.tmpDataPath=/vlagent_pq/vlagent-remotewrite-data
@@ -1100,8 +1101,8 @@ containers:
       - name: varlib
         readonly: true
         mountpath: /var/lib
-      - name: vl-collector-data
-        mountpath: /vl-collector
+      - name: data
+        mountpath: /var/lib/vlagent
       - name: persistent-queue-data
         readonly: false
         mountpath: /vlagent_pq/vlagent-remotewrite-data
@@ -1140,13 +1141,14 @@ volumes:
   volumesource:
     hostpath:
       path: /var/lib
-- name: vl-collector-data
+- name: data
   volumesource:
     hostpath:
-      path: /var/lib/vl-collector
+      path: /var/lib/vlagent
 - name: persistent-queue-data
   volumesource:
-    emptydir: {}
+    hostpath:
+      path: /var/lib/vlagent/vlagent_pq/vlagent-remotewrite-data
 
     `)