vlagent: implement Kubernetes log collector (#815)

Co-authored-by: Aliaksandr Valialkin <[email protected]>

Author: vadimalekseev

.gitignore

@@ -11,6 +11,7 @@
 /gocache-for-docker
 /victoria-logs-data
 /vlagent-remotewrite-data
+/vlagent-kubernetes-checkpoints.json
 .DS_store
 Gemfile.lock
 /_site

app/vlagent/kubernetescollector/README.md

@@ -0,0 +1,39 @@
+# Running vlagent outside Kubernetes for local development
+
+This guide explains how to run vlagent outside a Kubernetes cluster
+for local development and testing purposes.
+
+## Prerequisites
+
+Install [k3d](https://github.com/k3d-io/k3d) - a lightweight tool for running Kubernetes locally,
+with support for mounting `/var/log/*` folders from the guest to the host system.
+
+## Setup
+
+1. Create `/var/log/containers` and `/var/log/pods` directories on your host system:
+
+```bash
+mkdir -p /var/log/containers /var/log/pods
+
+# Ensure both vlagent and k3s have read/write permissions
+chmod a+rw /var/log/pods /var/log/containers
+```
+
+2. Create a k3d cluster with proper volume mounts:
+
+```bash
+k3d cluster create -v /var/log/containers:/var/log/containers@all -v /var/log/pods:/var/log/pods@all
+```
+
+This command will also update the `~/.kube/config` file to use the new k3d cluster.
+vlagent will use this kubeconfig file to connect to the currently selected cluster.
+You can change the kubeconfig path via the `KUBECONFIG` environment variable.
+
+3. Run vlagent with Kubernetes discovery enabled:
+
+```bash
+./vlagent -remoteWrite.url=http://localhost:9428/internal/insert -kubernetes
+```
+
+vlagent connects to the Kubernetes API to discover pods and containers running in the cluster.
+It reads logs from the `/var/log/containers` and `/var/log/pods` directories mounted on the host system.

app/vlagent/kubernetescollector/backoff_timer.go

@@ -0,0 +1,67 @@
+package kubernetescollector
+
+import (
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timerpool"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/timeutil"
+)
+
+// backoffTimer implements an exponential backoff timer with jitter.
+type backoffTimer struct {
+	min     time.Duration
+	max     time.Duration
+	current time.Duration
+
+	timer *time.Timer
+}
+
+// newBackoffTimer returns a new backoffTimer initialized with the given minDelay and maxDelay.
+// The caller must call stop() when the backoffTimer is no longer needed.
+func newBackoffTimer(minDelay, maxDelay time.Duration) backoffTimer {
+	return backoffTimer{
+		min:     minDelay,
+		max:     maxDelay,
+		current: minDelay,
+	}
+}
+
+// wait sleeps for the current delay with jitter, doubling the delay for the next wait.
+// Use currentDelay to get the current backoff duration.
+func (bt *backoffTimer) wait(stopCh <-chan struct{}) {
+	v := timeutil.AddJitterToDuration(bt.current)
+	bt.current *= 2
+	if bt.current > bt.max {
+		bt.current = bt.max
+	}
+
+	if bt.timer == nil {
+		bt.timer = timerpool.Get(v)
+	} else {
+		bt.timer.Reset(v)
+	}
+
+	select {
+	case <-stopCh:
+		bt.timer.Stop()
+	case <-bt.timer.C:
+	}
+}
+
+// currentDelay returns the current backoff duration.
+func (bt *backoffTimer) currentDelay() time.Duration {
+	return bt.current
+}
+
+// reset sets the backoff delay to its minimum.
+func (bt *backoffTimer) reset() {
+	bt.current = bt.min
+}
+
+// stop releases internal resources.
+func (bt *backoffTimer) stop() {
+	if bt.timer != nil {
+		timerpool.Put(bt.timer)
+		bt.timer = nil
+	}
+}

app/vlagent/kubernetescollector/checkpoints_db.go

@@ -0,0 +1,172 @@
+package kubernetescollector
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"slices"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/VictoriaMetrics/VictoriaLogs/lib/logstorage"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/fs"
+	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
+)
+
+// checkpointsDB manages persistent log file reading state checkpoints.
+// It saves reading positions to disk to enable resuming log collection
+// after vlagent restarts without data loss or duplication.
+//
+// The caller is responsible for closing checkpointsDB via stop() method
+// when it's no longer needed.
+type checkpointsDB struct {
+	checkpointsPath string
+
+	checkpoints     map[string]checkpoint
+	checkpointsLock sync.Mutex
+
+	wg     sync.WaitGroup
+	stopCh chan struct{}
+}
+
+// startCheckpointsDB starts a checkpointsDB instance.
+// The caller must call stop() when the checkpointsDB is no longer needed.
+func startCheckpointsDB(path string) (*checkpointsDB, error) {
+	checkpoints, err := readCheckpoints(path)
+	if err != nil {
+		return nil, err
+	}
+
+	checkpointsMap := make(map[string]checkpoint)
+	for _, cp := range checkpoints {
+		checkpointsMap[cp.Path] = cp
+	}
+
+	db := &checkpointsDB{
+		checkpointsPath: path,
+		checkpoints:     checkpointsMap,
+		stopCh:          make(chan struct{}),
+	}
+
+	db.startPeriodicSyncCheckpoints()
+
+	return db, nil
+}
+
+// checkpoint represents a persistent snapshot of a log file reading state.
+//
+// The checkpoint is saved to disk to enable resuming log collection from the exact
+// position after vlagent restarts, preventing:
+//  1. Log duplication when logs are re-read from the beginning.
+//  2. Log loss when a log file was rotated while vlagent was down.
+//     In this case we should find the rotated file.
+//
+// checkpoint includes pod metadata (common and stream fields) to allow immediate log processing
+// without waiting for the Kubernetes API server to provide pod information.
+type checkpoint struct {
+	Path         string             `json:"path"`
+	Inode        uint64             `json:"inode"`
+	Offset       int64              `json:"offset"`
+	CommonFields []logstorage.Field `json:"commonFields"`
+}
+
+func (db *checkpointsDB) set(cp checkpoint) {
+	db.checkpointsLock.Lock()
+	defer db.checkpointsLock.Unlock()
+
+	db.checkpoints[cp.Path] = cp
+}
+
+func (db *checkpointsDB) get(path string) (checkpoint, bool) {
+	db.checkpointsLock.Lock()
+	defer db.checkpointsLock.Unlock()
+
+	cp, ok := db.checkpoints[path]
+	return cp, ok
+}
+
+func (db *checkpointsDB) getAll() []checkpoint {
+	db.checkpointsLock.Lock()
+	defer db.checkpointsLock.Unlock()
+
+	cps := make([]checkpoint, 0, len(db.checkpoints))
+	for _, cp := range db.checkpoints {
+		cps = append(cps, cp)
+	}
+
+	return cps
+}
+
+func (db *checkpointsDB) delete(path string) {
+	db.checkpointsLock.Lock()
+	defer db.checkpointsLock.Unlock()
+
+	delete(db.checkpoints, path)
+}
+
+func (db *checkpointsDB) mustSync() {
+	cps := db.getAll()
+
+	slices.SortFunc(cps, func(a, b checkpoint) int {
+		return strings.Compare(a.Path, b.Path)
+	})
+
+	data, err := json.MarshalIndent(cps, "", "\t")
+	if err != nil {
+		logger.Panicf("BUG: cannot marshal checkpoints: %s", err)
+	}
+
+	fs.MustWriteAtomic(db.checkpointsPath, data, true)
+}
+
+func readCheckpoints(path string) ([]checkpoint, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		if errors.Is(err, os.ErrNotExist) {
+			return nil, nil
+		}
+		return nil, fmt.Errorf("cannot read file checkpoints: %w", err)
+	}
+
+	if len(data) == 0 {
+		return nil, nil
+	}
+
+	var checkpoints []checkpoint
+	if err := json.Unmarshal(data, &checkpoints); err != nil {
+		return nil, fmt.Errorf("cannot unmarshal file checkpoints from %q: %w", path, err)
+	}
+
+	return checkpoints, nil
+}
+
+// startPeriodicFlushCheckpoints periodically persists in-memory checkpoints to disk.
+//
+// It complements the explicit sync performed on graceful stop,
+// ensuring regular persistence even when the process is killed.
+func (db *checkpointsDB) startPeriodicSyncCheckpoints() {
+	db.wg.Add(1)
+	go func() {
+		defer db.wg.Done()
+
+		ticker := time.NewTicker(1 * time.Minute)
+		defer ticker.Stop()
+
+		for {
+			select {
+			case <-ticker.C:
+				db.mustSync()
+			case <-db.stopCh:
+				db.mustSync()
+				return
+			}
+		}
+	}()
+}
+
+func (db *checkpointsDB) stop() {
+	close(db.stopCh)
+	db.wg.Wait()
+}