feat(logs): support Windows logs, dynamic date paths, and global metrics

.github/workflows/ci.yml

@@ -39,19 +39,18 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 1
-      - uses: WillAbides/setup-go-faster@v1.14.0
+      - uses: actions/setup-go@v6
         with:
           go-version: "1.25"
       - name: Tidy Go modules
         run: go mod tidy
       - name: Run golang-ci-lint
-        uses: golangci/golangci-lint-action@v3.6.0
+        uses: golangci/golangci-lint-action@v9
         with:
-          version: v1.53
+          version: v2.12
           args: --timeout=10m --tests=false
           only-new-issues: true
-          skip-pkg-cache: true
-          skip-build-cache: true
+          skip-cache: true
       - name: Run go version check
         run: make go-version-check
       - name: Run go vet check
@@ -64,9 +63,9 @@ jobs:
       - uses: actions/checkout@v4
         with:
           fetch-depth: 1
-      - uses: WillAbides/setup-go-faster@v1.14.0
+      - uses: actions/setup-go@v6
         with:
-          go-version: "1.24"
+          go-version: "1.25"
       - name: make vendor
         run: make vendor-ci
       - name: go build linux

agent/logs_agent.go

@@ -94,12 +94,34 @@ func NewLogsAgent() AgentModule {
 	if os.IsNotExist(err) {
 		os.MkdirAll(coreconfig.GetLogRunPath(), 0755)
 	}
-	auditor := auditor.New(coreconfig.GetLogRunPath(), auditor.DefaultRegistryFilename, auditorTTL)
+	auditorIns := auditor.New(coreconfig.GetLogRunPath(), auditor.DefaultRegistryFilename, auditorTTL)
+
+	auditor.RegisterCollector(auditorIns, func() []auditor.SourceMeta {
+		srcs := sources.GetSources()
+		meta := make([]auditor.SourceMeta, 0, len(srcs))
+		for _, s := range srcs {
+			if s.Config == nil {
+				continue
+			}
+			sm := auditor.SourceMeta{
+				ConfigPath: s.Config.Path,
+				SourceType: s.Config.Type,
+				Source:     s.Config.Source,
+				Service:    s.Config.Service,
+			}
+			if expandedTags := s.Config.Tags; len(expandedTags) > 0 {
+				sm.Tags = auditor.ParseTags(expandedTags)
+			}
+			meta = append(meta, sm)
+		}
+		return meta
+	})
+
 	destinationsCtx := client.NewDestinationsContext()
 	diagnosticMessageReceiver := diagnostic.NewBufferedMessageReceiver()
 
 	// setup the pipeline provider that provides pairs of processor and sender
-	pipelineProvider := pipeline.NewProvider(coreconfig.NumberOfPipelines(), auditor, diagnosticMessageReceiver, processingRules, endpoints, destinationsCtx)
+	pipelineProvider := pipeline.NewProvider(coreconfig.NumberOfPipelines(), auditorIns, diagnosticMessageReceiver, processingRules, endpoints, destinationsCtx)
 
 	validatePodContainerID := coreconfig.ValidatePodContainerID()
 	//
@@ -114,10 +136,10 @@ func NewLogsAgent() AgentModule {
 
 	// setup the inputs
 	inputs := []restart.Restartable{
-		file.NewScanner(sources, coreconfig.OpenLogsLimit(), pipelineProvider, auditor,
+		file.NewScanner(sources, coreconfig.OpenLogsLimit(), coreconfig.MaxTraverseLimit(), coreconfig.MaxDepthLimit(), pipelineProvider, auditorIns,
 			file.DefaultSleepDuration, validatePodContainerID, time.Duration(time.Duration(coreconfig.FileScanPeriod())*time.Second)),
 		listener.NewLauncher(sources, coreconfig.LogFrameSize(), pipelineProvider),
-		journald.NewLauncher(sources, pipelineProvider, auditor),
+		journald.NewLauncher(sources, pipelineProvider, auditorIns),
 	}
 	if coreconfig.EnableCollectContainer() {
 		log.Println("collect docker logs...")
@@ -129,7 +151,7 @@ func NewLogsAgent() AgentModule {
 		services:                  services,
 		processingRules:           processingRules,
 		endpoints:                 endpoints,
-		auditor:                   auditor,
+		auditor:                   auditorIns,
 		destinationsCtx:           destinationsCtx,
 		pipelineProvider:          pipelineProvider,
 		inputs:                    inputs,
@@ -188,6 +210,7 @@ func (a *LogsAgent) Flush(ctx context.Context) {
 // Stop stops all the elements of the data pipeline
 // in the right order to prevent data loss
 func (a *LogsAgent) Stop() error {
+	auditor.UnregisterCollector(a.auditor)
 	inputs := restart.NewParallelStopper()
 	for _, input := range a.inputs {
 		inputs.Add(input)

config/logs.go

@@ -26,6 +26,8 @@ type (
 		BatchWait             int                          `json:"batch_wait" toml:"batch_wait"`
 		RunPath               string                       `json:"run_path" toml:"run_path"`
 		OpenFilesLimit        int                          `json:"open_files_limit" toml:"open_files_limit"`
+		MaxTraverseLimit      int                          `json:"max_traverse_limit" toml:"max_traverse_limit"`
+		MaxDepthLimit         int                          `json:"max_depth_limit" toml:"max_depth_limit"`
 		ScanPeriod            int                          `json:"scan_period" toml:"scan_period"`
 		FrameSize             int                          `json:"frame_size" toml:"frame_size"`
 		CollectContainerAll   bool                         `json:"collect_container_all" toml:"collect_container_all"`
@@ -91,6 +93,20 @@ func OpenLogsLimit() int {
 	return Config.Logs.OpenFilesLimit
 }
 
+func MaxTraverseLimit() int {
+	if Config.Logs.MaxTraverseLimit <= 0 {
+		return 100000
+	}
+	return Config.Logs.MaxTraverseLimit
+}
+
+func MaxDepthLimit() int {
+	if Config.Logs.MaxDepthLimit <= 0 {
+		return 15
+	}
+	return Config.Logs.MaxDepthLimit
+}
+
 func FileScanPeriod() int {
 	if Config.Logs.ScanPeriod == 0 {
 		Config.Logs.ScanPeriod = 10

go.mod

@@ -215,6 +215,7 @@ require (
 	github.com/bits-and-blooms/bitset v1.13.0
 	github.com/blang/semver/v4 v4.0.0
 	github.com/bmatcuk/doublestar/v3 v3.0.0
+	github.com/bmatcuk/doublestar/v4 v4.10.0
 	github.com/cenkalti/backoff/v4 v4.3.0
 	github.com/coreos/go-systemd/v22 v22.5.0
 	github.com/dennwc/btrfs v0.0.0-20230312211831-a1f570bd01a1

go.sum

@@ -785,6 +785,8 @@ github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM
 github.com/blang/semver/v4 v4.0.0/go.mod h1:IbckMUScFkM3pff0VJDNKRiT6TG/YpiHIM2yvyW5YoQ=
 github.com/bmatcuk/doublestar/v3 v3.0.0 h1:TQtVPlDnAYwcrVNB2JiGuMc++H5qzWZd9PhkNo5WyHI=
 github.com/bmatcuk/doublestar/v3 v3.0.0/go.mod h1:6PcTVMw80pCY1RVuoqu3V++99uQB3vsSYKPTd8AWA0k=
+github.com/bmatcuk/doublestar/v4 v4.10.0 h1:zU9WiOla1YA122oLM6i4EXvGW62DvKZVxIe6TYWexEs=
+github.com/bmatcuk/doublestar/v4 v4.10.0/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc=
 github.com/boombuler/barcode v1.0.0/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
 github.com/boombuler/barcode v1.0.1/go.mod h1:paBWMcWSl3LHKBqUq+rly7CNSldXjb2rDl3JlRe0mD8=
 github.com/bytedance/gopkg v0.1.3 h1:TPBSwH8RsouGCBcMBktLt1AymVo2TVsBVCY4b6TnZ/M=

inputs/elasticsearch/collector/data_stream_test.go

@@ -64,12 +64,12 @@ func TestDataStream(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			c := NewDataStream(http.DefaultClient, u)
+			c, err := NewDataStream(u, http.DefaultClient)
 			if err != nil {
 				t.Fatal(err)
 			}
 
-			if err := testutil.CollectAndCompare(c, strings.NewReader(tt.want)); err != nil {
+			if err := testutil.CollectAndCompare(wrapCollector{c}, strings.NewReader(tt.want)); err != nil {
 				t.Fatal(err)
 			}
 		})

inputs/elasticsearch/collector/ilm_test.go

@@ -24,7 +24,6 @@ import (
 	"testing"
 
 	"github.com/prometheus/client_golang/prometheus/testutil"
-	"github.com/prometheus/common/promslog"
 )
 
 func TestILM(t *testing.T) {
@@ -81,7 +80,7 @@ func TestILM(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			c, err := NewILM(promslog.NewNopLogger(), u, http.DefaultClient)
+			c, err := NewILM(u, http.DefaultClient)
 			if err != nil {
 				t.Fatal(err)
 			}

inputs/elasticsearch/collector/indices_test.go

@@ -1,6 +1,7 @@
 package collector
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
@@ -38,7 +39,7 @@ func TestIndices(t *testing.T) {
 		indicesIncluded := make([]string, 0)
 		maxIndicesIncludeCount := 80
 		i := NewIndices(http.DefaultClient, u, false, true, indicesIncluded, maxIndicesIncludeCount)
-		stats, err := i.fetchAndDecodeIndexStats()
+		stats, err := i.fetchAndDecodeIndexStats(context.Background())
 		if err != nil {
 			t.Fatalf("Failed to fetch or decode indices stats: %s", err)
 		}
@@ -115,7 +116,7 @@ func TestAliases(t *testing.T) {
 		indicesIncluded := make([]string, 0)
 		maxIndicesIncludeCount := 80
 		i := NewIndices(http.DefaultClient, u, false, true, indicesIncluded, maxIndicesIncludeCount)
-		stats, err := i.fetchAndDecodeIndexStats()
+		stats, err := i.fetchAndDecodeIndexStats(context.Background())
 		if err != nil {
 			t.Fatalf("Failed to fetch or decode indices stats: %s", err)
 		}

inputs/elasticsearch/collector/nodes_test.go

@@ -58,7 +58,7 @@ func TestNodesStats(t *testing.T) {
 				}
 				u.User = url.UserPassword("elastic", "changeme")
 				nodeStats := make([]string, 0)
-				c := NewNodes(http.DefaultClient, u, true, "_local", false, nodeStats, "test")
+				c := NewNodes(http.DefaultClient, u, true, "_local", false, nodeStats)
 				nsr, err := c.fetchAndDecodeNodeStats()
 				if err != nil {
 					t.Fatalf("Failed to fetch or decode node stats: %s", err)

inputs/elasticsearch/collector/slm.go

@@ -140,6 +140,22 @@ type SLMStatusResponse struct {
 	OperationMode string `json:"operation_mode"`
 }
 
+func (s *SLM) fetchAndDecodeSLMStats(ctx context.Context) (SLMStatsResponse, error) {
+	u := s.u.ResolveReference(&url.URL{Path: "/_slm/stats"})
+	var slmStatsResp SLMStatsResponse
+
+	resp, err := getURL(ctx, s.hc, u.String())
+	if err != nil {
+		return slmStatsResp, err
+	}
+
+	err = json.Unmarshal(resp, &slmStatsResp)
+	if err != nil {
+		return slmStatsResp, err
+	}
+	return slmStatsResp, nil
+}
+
 func (s *SLM) Update(ctx context.Context, ch chan<- prometheus.Metric) error {
 	u := s.u.ResolveReference(&url.URL{Path: "/_slm/status"})
 	var slmStatusResp SLMStatusResponse
@@ -154,15 +170,7 @@ func (s *SLM) Update(ctx context.Context, ch chan<- prometheus.Metric) error {
 		return err
 	}
 
-	u = s.u.ResolveReference(&url.URL{Path: "/_slm/stats"})
-	var slmStatsResp SLMStatsResponse
-
-	resp, err = getURL(ctx, s.hc, u.String())
-	if err != nil {
-		return err
-	}
-
-	err = json.Unmarshal(resp, &slmStatsResp)
+	slmStatsResp, err := s.fetchAndDecodeSLMStats(ctx)
 	if err != nil {
 		return err
 	}

inputs/elasticsearch/collector/slm_test.go

@@ -14,6 +14,7 @@
 package collector
 
 import (
+	"context"
 	"fmt"
 	"net/http"
 	"net/http/httptest"
@@ -42,8 +43,12 @@ func TestSLM(t *testing.T) {
 		if err != nil {
 			t.Fatalf("Failed to parse URL: %s", err)
 		}
-		s := NewSLM(http.DefaultClient, u)
-		stats, err := s.fetchAndDecodeSLMStats()
+		c, err := NewSLM(u, http.DefaultClient)
+		if err != nil {
+			t.Fatalf("Failed to create SLM: %s", err)
+		}
+		s := c.(*SLM)
+		stats, err := s.fetchAndDecodeSLMStats(context.Background())
 		if err != nil {
 			t.Fatalf("Failed to fetch or decode snapshots stats: %s", err)
 		}

inputs/elasticsearch/collector/snapshots_test.go

@@ -208,15 +208,12 @@ func TestSnapshots(t *testing.T) {
 				t.Fatal(err)
 			}
 
-			s := NewSnapshots(http.DefaultClient, u)
-
-			// TODO: Convert to collector interface
-			// c, err := NewSnapshots(log.NewNopLogger(), u, http.DefaultClient)
-			// if err != nil {
-			// 	t.Fatal(err)
-			// }
+			s, err := NewSnapshots(u, http.DefaultClient)
+			if err != nil {
+				t.Fatal(err)
+			}
 
-			if err := testutil.CollectAndCompare(s, strings.NewReader(tt.want)); err != nil {
+			if err := testutil.CollectAndCompare(wrapCollector{s}, strings.NewReader(tt.want)); err != nil {
 				t.Fatal(err)
 			}
 		})

inputs/elasticsearch/pkg/clusterinfo/clusterinfo_test.go

@@ -19,14 +19,13 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/url"
-	"os"
+
 	"reflect"
 	"sync"
 	"testing"
 	"time"
 
 	"github.com/blang/semver/v4"
-	"github.com/prometheus/common/promslog"
 )
 
 const (
@@ -118,7 +117,7 @@ func TestNew(t *testing.T) {
 	if err != nil {
 		t.Skipf("internal test error: %s", err)
 	}
-	r := New(promslog.NewNopLogger(), http.DefaultClient, u, 0)
+	r := New(http.DefaultClient, u, 0)
 	if r.url != u {
 		t.Errorf("new Retriever mal-constructed")
 	}
@@ -130,7 +129,7 @@ func TestRetriever_RegisterConsumer(t *testing.T) {
 	if err != nil {
 		t.Fatalf("internal test error: %s", err)
 	}
-	retriever := New(promslog.NewNopLogger(), mockES.Client(), u, 0)
+	retriever := New(mockES.Client(), u, 0)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	consumerNames := []string{"consumer-1", "consumer-2"}
@@ -169,7 +168,7 @@ func TestRetriever_fetchAndDecodeClusterInfo(t *testing.T) {
 	if err != nil {
 		t.Skipf("internal test error: %s", err)
 	}
-	retriever := New(promslog.NewNopLogger(), mockES.Client(), u, 0)
+	retriever := New(mockES.Client(), u, 0)
 	ci, err := retriever.fetchAndDecodeClusterInfo()
 	if err != nil {
 		t.Fatalf("failed to retrieve cluster info: %s", err)
@@ -189,7 +188,7 @@ func TestRetriever_Run(t *testing.T) {
 	}
 
 	// setup cluster info retriever
-	retriever := New(promslog.New(&promslog.Config{Writer: os.Stdout}), mockES.Client(), u, 0)
+	retriever := New(mockES.Client(), u, 0)
 
 	// setup mock consumer
 	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)