From 5328a4cfd8282b8f8c6a7f3c2300f7681341d485 Mon Sep 17 00:00:00 2001 From: kongfei605 Date: Thu, 18 Jun 2026 12:08:42 +0800 Subject: [PATCH 1/5] feat(logs): support Windows logs, dynamic date paths, and global metrics --- agent/logs_agent.go | 33 +++- config/logs.go | 16 ++ go.mod | 1 + go.sum | 2 + inputs/self_metrics/log_metrics.go | 60 +++++++ inputs/self_metrics/log_metrics_none.go | 7 + inputs/self_metrics/metrics.go | 2 + logs/auditor/auditor.go | 6 + logs/auditor/global.go | 188 ++++++++++++++++++++++ logs/auditor/global_none.go | 26 ++++ logs/auditor/null_auditor.go | 3 + logs/input/file/file_provider.go | 198 ++++++++++++++++++++---- logs/input/file/file_provider_test.go | 191 +++++++++++++++++++++++ logs/input/file/rotate_windows.go | 70 ++++++++- logs/input/file/scanner.go | 6 +- logs/input/file/tailer.go | 7 +- logs/input/file/tailer_windows.go | 29 +++- logs/util/date_format.go | 116 ++++++++++++++ logs/util/date_format_test.go | 61 ++++++++ 19 files changed, 975 insertions(+), 47 deletions(-) create mode 100644 inputs/self_metrics/log_metrics.go create mode 100644 inputs/self_metrics/log_metrics_none.go create mode 100644 logs/auditor/global.go create mode 100644 logs/auditor/global_none.go create mode 100644 logs/input/file/file_provider_test.go create mode 100644 logs/util/date_format.go create mode 100644 logs/util/date_format_test.go diff --git a/agent/logs_agent.go b/agent/logs_agent.go index e141c1eaa..23e2e526e 100644 --- a/agent/logs_agent.go +++ b/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) diff --git a/config/logs.go b/config/logs.go index b7f9e2f88..babbc630f 100644 --- a/config/logs.go +++ b/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 diff --git a/go.mod b/go.mod index db873403a..9d44540a5 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 400db2318..661ed460d 100644 --- a/go.sum +++ b/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= diff --git a/inputs/self_metrics/log_metrics.go b/inputs/self_metrics/log_metrics.go new file mode 100644 index 000000000..686be44e9 --- /dev/null +++ b/inputs/self_metrics/log_metrics.go @@ -0,0 +1,60 @@ +//go:build !no_logs + +package categraf + +import ( + "strconv" + "time" + + config "flashcat.cloud/categraf/config" + "flashcat.cloud/categraf/logs/auditor" + "flashcat.cloud/categraf/types" +) + +// reserved label keys that must not be overridden by user tags +var reservedLabels = map[string]bool{ + "path": true, "source_type": true, "tailing_mode": true, + "source": true, "service": true, +} + +func gatherLogMetrics(slist *types.SampleList) { + stat := auditor.GetCollectorStatus() + + running := 0 + if stat.Running { + running = 1 + } + + pipelines := config.NumberOfPipelines() + + slist.PushSample(defaultPrefix, "log_agent_running", running) + slist.PushSample(defaultPrefix, "log_agent_pipelines", pipelines) + slist.PushSample(defaultPrefix, "log_files_tracked_total", stat.TrackedFiles) + + for _, e := range stat.Entries { + tags := map[string]string{ + "path": e.Path, + "source_type": e.SourceType, + "tailing_mode": e.TailingMode, + } + if e.Source != "" { + tags["source"] = e.Source + } + if e.Service != "" { + tags["service"] = e.Service + } + + // merge user-defined item tags (skip reserved keys) + for k, v := range e.Tags { + if !reservedLabels[k] { + tags[k] = v + } + } + + if offsetVal, err := strconv.ParseInt(e.Offset, 10, 64); err == nil { + slist.PushSample(defaultPrefix, "log_source_offset", offsetVal, tags) + } + ageSec := time.Since(e.LastUpdated).Seconds() + slist.PushSample(defaultPrefix, "log_source_last_updated_seconds", ageSec, tags) + } +} diff --git a/inputs/self_metrics/log_metrics_none.go b/inputs/self_metrics/log_metrics_none.go new file mode 100644 index 000000000..dc35a9649 --- /dev/null +++ b/inputs/self_metrics/log_metrics_none.go @@ -0,0 +1,7 @@ +//go:build no_logs + +package categraf + +import "flashcat.cloud/categraf/types" + +func gatherLogMetrics(slist *types.SampleList) {} diff --git a/inputs/self_metrics/metrics.go b/inputs/self_metrics/metrics.go index 6470b8b4e..adcce28ee 100644 --- a/inputs/self_metrics/metrics.go +++ b/inputs/self_metrics/metrics.go @@ -68,4 +68,6 @@ func (ins *Categraf) Gather(slist *types.SampleList) { } } } + + gatherLogMetrics(slist) } diff --git a/logs/auditor/auditor.go b/logs/auditor/auditor.go index 1f30720a5..a88d8b2d8 100644 --- a/logs/auditor/auditor.go +++ b/logs/auditor/auditor.go @@ -54,6 +54,7 @@ type Auditor interface { Start() Stop() Channel() chan *message.Message + GetAllEntries() map[string]RegistryEntry } // A RegistryAuditor is storing the Auditor information using a registry. @@ -240,6 +241,11 @@ func (a *RegistryAuditor) readOnlyRegistryCopy() map[string]RegistryEntry { return r } +// GetAllEntries returns a read-only copy of all registry entries. +func (a *RegistryAuditor) GetAllEntries() map[string]RegistryEntry { + return a.readOnlyRegistryCopy() +} + // flushRegistry writes on disk the registry at the given path func (a *RegistryAuditor) flushRegistry() error { r := a.readOnlyRegistryCopy() diff --git a/logs/auditor/global.go b/logs/auditor/global.go new file mode 100644 index 000000000..1d64c1910 --- /dev/null +++ b/logs/auditor/global.go @@ -0,0 +1,188 @@ +//go:build !no_logs + +package auditor + +import ( + "path/filepath" + "strings" + "sync" + "time" + + "flashcat.cloud/categraf/logs/util" + "github.com/bmatcuk/doublestar/v4" +) + +// SourceMeta holds per-source runtime metadata for metrics labeling. +type SourceMeta struct { + ConfigPath string + SourceType string + Source string + Service string + Tags map[string]string +} + +type registeredCollector struct { + auditor Auditor + metaFn func() []SourceMeta +} + +var ( + globalMu sync.RWMutex + globalCollectors []registeredCollector +) + +// RegisterCollector registers an auditor with its dynamic source metadata loader callback. +func RegisterCollector(a Auditor, metaFn func() []SourceMeta) { + globalMu.Lock() + defer globalMu.Unlock() + globalCollectors = append(globalCollectors, registeredCollector{ + auditor: a, + metaFn: metaFn, + }) +} + +// UnregisterCollector removes an auditor from global registry. +func UnregisterCollector(a Auditor) { + globalMu.Lock() + defer globalMu.Unlock() + for i, c := range globalCollectors { + if c.auditor == a { + globalCollectors = append(globalCollectors[:i], globalCollectors[i+1:]...) + return + } + } +} + +// LogFileEntry represents a single tracked log entry for metrics. +type LogFileEntry struct { + Path string + SourceType string + TailingMode string + Offset string + LastUpdated time.Time + Source string + Service string + Tags map[string]string +} + +// CollectorStatus holds aggregated log collector status. +type CollectorStatus struct { + Running bool + TrackedFiles int + Entries []LogFileEntry +} + +// GetCollectorStatus returns all tracked entries from all registered auditors. +func GetCollectorStatus() CollectorStatus { + globalMu.RLock() + defer globalMu.RUnlock() + + if len(globalCollectors) == 0 { + return CollectorStatus{} + } + + stat := CollectorStatus{Running: true} + for _, c := range globalCollectors { + allEntries := c.auditor.GetAllEntries() + if allEntries == nil { + continue + } + stat.TrackedFiles += len(allEntries) + + var fnMeta []SourceMeta + if c.metaFn != nil { + fnMeta = c.metaFn() + } + + for id, entry := range allEntries { + resolvedPath := id + sourceType := "unknown" + if strings.HasPrefix(id, "file:") { + sourceType = "file" + resolvedPath = strings.TrimPrefix(id, "file:") + } else if strings.HasPrefix(id, "journald:") { + sourceType = "journald" + resolvedPath = strings.TrimPrefix(id, "journald:") + } + + e := LogFileEntry{ + Path: resolvedPath, + SourceType: sourceType, + TailingMode: entry.TailingMode, + Offset: entry.Offset, + LastUpdated: entry.LastUpdated, + } + + if meta := matchItemMeta(resolvedPath, fnMeta); meta != nil { + e.Source = meta.Source + e.Service = meta.Service + e.Tags = meta.Tags + } + + stat.Entries = append(stat.Entries, e) + } + } + return stat +} + +// matchItemMeta finds the SourceMeta whose ConfigPath matches the resolved path. +func matchItemMeta(resolvedPath string, items []SourceMeta) *SourceMeta { + for i := range items { + configPath := items[i].ConfigPath + if configPath == "" { + continue + } + + if util.ContainsDatePattern(configPath) { + configPath = util.ExpandDatePattern(configPath, time.Now()) + } + + if configPath == resolvedPath { + return &items[i] + } + + configSlash := filepath.ToSlash(configPath) + resolvedSlash := filepath.ToSlash(resolvedPath) + if matched, _ := doublestar.Match(configSlash, resolvedSlash); matched { + return &items[i] + } + } + return nil +} + +// ParseTags parses tag strings into a map, flexibly splitting by the first '=' or ':'. +func ParseTags(tags []string) map[string]string { + result := make(map[string]string, len(tags)) + for _, tag := range tags { + if tag == "" { + continue + } + + iEq := strings.IndexRune(tag, '=') + iColon := strings.IndexRune(tag, ':') + idx := -1 + if iEq >= 0 && iColon >= 0 { + if iEq < iColon { + idx = iEq + } else { + idx = iColon + } + } else if iEq >= 0 { + idx = iEq + } else if iColon >= 0 { + idx = iColon + } + + if idx < 0 { + continue + } + + key := strings.TrimSpace(tag[:idx]) + value := strings.TrimSpace(tag[idx+1:]) + if key == "" { + continue + } + result[key] = value + } + return result +} diff --git a/logs/auditor/global_none.go b/logs/auditor/global_none.go new file mode 100644 index 000000000..bfdaa359b --- /dev/null +++ b/logs/auditor/global_none.go @@ -0,0 +1,26 @@ +//go:build no_logs + +package auditor + +// CollectorStatus holds aggregated log collector status. +type CollectorStatus struct { + Running bool + TrackedFiles int + Entries []LogFileEntry +} + +// LogFileEntry represents a single tracked log entry for metrics. +type LogFileEntry struct { + Path string + SourceType string + TailingMode string + Offset string + Source string + Service string + Tags map[string]string +} + +// GetCollectorStatus returns empty status under no_logs build. +func GetCollectorStatus() CollectorStatus { + return CollectorStatus{} +} diff --git a/logs/auditor/null_auditor.go b/logs/auditor/null_auditor.go index 8b9fe81d6..b59a6af50 100644 --- a/logs/auditor/null_auditor.go +++ b/logs/auditor/null_auditor.go @@ -26,6 +26,9 @@ func NewNullAuditor() *NullAuditor { // GetOffset returns an empty string. func (a *NullAuditor) GetOffset(identifier string) string { return "" } +// GetAllEntries returns nil for NullAuditor. +func (a *NullAuditor) GetAllEntries() map[string]RegistryEntry { return nil } + // GetTailingMode returns an empty string. func (a *NullAuditor) GetTailingMode(identifier string) string { return "" } diff --git a/logs/input/file/file_provider.go b/logs/input/file/file_provider.go index b661ff649..9bcd67e9b 100644 --- a/logs/input/file/file_provider.go +++ b/logs/input/file/file_provider.go @@ -8,20 +8,45 @@ package file import ( + "errors" "fmt" "log" "os" "path/filepath" "sort" + "strings" + "time" + "github.com/bmatcuk/doublestar/v4" + "github.com/prometheus/client_golang/prometheus" + + "flashcat.cloud/categraf/config" logsconfig "flashcat.cloud/categraf/config/logs" "flashcat.cloud/categraf/logs/status" + "flashcat.cloud/categraf/logs/util" ) // OpenFilesLimitWarningType is the key of the message generated when too many // files are tailed const openFilesLimitWarningType = "open_files_limit_warning" +var ErrMaxTraverseLimit = errors.New("max traverse limit reached") + +var fileProviderPermissionDeniedTotal = prometheus.NewCounter( + prometheus.CounterOpts{ + Name: "categraf_logs_file_permission_denied_total", + Help: "Total number of permission denied errors encountered while walking recursive log paths", + }, +) + +func init() { + if err := prometheus.Register(fileProviderPermissionDeniedTotal); err != nil { + if _, ok := err.(prometheus.AlreadyRegisteredError); !ok { + log.Println("W! failed to register fileProviderPermissionDeniedTotal metric:", err) + } + } +} + // File represents a file to tail type File struct { Path string @@ -44,23 +69,36 @@ func NewFile(path string, source *logsconfig.LogSource, isWildcardPath bool) *Fi // If it is a file scanned for a container, it will use the format: / // Otherwise, it will simply use the format: func (t *File) GetScanKey() string { - if t.Source != nil && t.Source.Config != nil && t.Source.Config.Identifier != "" { - return fmt.Sprintf("%s/%s", t.Path, t.Source.Config.Identifier) + key := t.Path + if t.Source != nil && t.Source.Config != nil { + if t.Source.Config.Identifier != "" { + key = fmt.Sprintf("%s/%s", key, t.Source.Config.Identifier) + } } - return t.Path + return key } // Provider implements the logic to retrieve at most filesLimit Files defined in sources type Provider struct { - filesLimit int - shouldLogErrors bool + filesLimit int + maxTraverseLimit int + maxDepthLimit int + shouldLogErrors bool } // NewProvider returns a new Provider -func NewProvider(filesLimit int) *Provider { +func NewProvider(filesLimit, maxTraverseLimit, maxDepthLimit int) *Provider { + if maxTraverseLimit <= 0 { + maxTraverseLimit = config.MaxTraverseLimit() + } + if maxDepthLimit <= 0 { + maxDepthLimit = config.MaxDepthLimit() + } return &Provider{ - filesLimit: filesLimit, - shouldLogErrors: true, + filesLimit: filesLimit, + maxTraverseLimit: maxTraverseLimit, + maxDepthLimit: maxDepthLimit, + shouldLogErrors: true, } } @@ -77,7 +115,7 @@ func (p *Provider) FilesToTail(sources []*logsconfig.LogSource) []*File { source := sources[i] tailedFileCounter := 0 files, err := p.CollectFiles(source) - isWildcardPath := logsconfig.ContainsWildcard(source.Config.Path) + isWildcardPath := logsconfig.ContainsWildcard(source.Config.Path) || util.ContainsDatePattern(source.Config.Path) if err != nil { source.Status.Error(err) if isWildcardPath { @@ -122,6 +160,13 @@ func (p *Provider) FilesToTail(sources []*logsconfig.LogSource) []*File { // CollectFiles returns all the files matching the source path. func (p *Provider) CollectFiles(source *logsconfig.LogSource) ([]*File, error) { path := source.Config.Path + originalPath := path + if util.ContainsDatePattern(path) { + expandedPath := util.ExpandDatePattern(path, time.Now()) + log.Printf("D! Expanded date pattern: %s -> %s", originalPath, expandedPath) + path = expandedPath + } + fileExists := p.exists(path) switch { case fileExists: @@ -132,15 +177,34 @@ func (p *Provider) CollectFiles(source *logsconfig.LogSource) ([]*File, error) { pattern := path return p.searchFiles(pattern, source) default: + if util.ContainsDatePattern(originalPath) { + return nil, fmt.Errorf("file %s does not exist (expanded from %s)", path, originalPath) + } return nil, fmt.Errorf("file %s does not exist", path) } } +// hasDoublestar checks if a pattern contains a ** segment +func hasDoublestar(pattern string) bool { + slashPattern := filepath.ToSlash(pattern) + return strings.Contains(slashPattern, "/**/") || + strings.HasPrefix(slashPattern, "**/") || + strings.HasSuffix(slashPattern, "/**") || + slashPattern == "**" +} + // searchFiles returns all the files matching the source path pattern. func (p *Provider) searchFiles(pattern string, source *logsconfig.LogSource) ([]*File, error) { - paths, err := filepath.Glob(pattern) + var paths []string + var err error + if hasDoublestar(pattern) { + paths, err = p.doublestarWalk(pattern) + } else { + paths, err = filepath.Glob(pattern) + } + if err != nil { - return nil, fmt.Errorf("malformed pattern, could not find any file: %s", pattern) + return nil, fmt.Errorf("malformed pattern or walk failed: %s, error: %w", pattern, err) } if len(paths) == 0 { // no file was found, its parent directories might have wrong permissions or it just does not exist @@ -164,33 +228,111 @@ func (p *Provider) searchFiles(pattern string, source *logsconfig.LogSource) ([] }) // Resolve excluded path(s) - excludedPaths := make(map[string]int) + now := time.Now() + var expandedExcludes []string for _, excludePattern := range source.Config.ExcludePaths { - excludedGlob, err := filepath.Glob(excludePattern) - if err != nil { - return nil, fmt.Errorf("malformed exclusion pattern: %s, %s", excludePattern, err) + if util.ContainsDatePattern(excludePattern) { + expandedExcludes = append(expandedExcludes, util.ExpandDatePattern(excludePattern, now)) + } else { + expandedExcludes = append(expandedExcludes, excludePattern) + } + } + + // Pre-compute slash-normalized excludes and hoist env lookup + normalizedExcludes := make([]string, len(expandedExcludes)) + for i, ep := range expandedExcludes { + normalizedExcludes[i] = filepath.ToSlash(ep) + } + hostMountPrefix := os.Getenv("HOST_MOUNT_PREFIX") + + for _, path := range paths { + isExcluded := false + pathSlash := filepath.ToSlash(path) + for _, excludePattern := range normalizedExcludes { + matched, err := doublestar.Match(excludePattern, pathSlash) + if err != nil { + log.Printf("W! Invalid exclude pattern %q: %v", excludePattern, err) + continue + } + if matched { + isExcluded = true + log.Printf("D! Excluded path: %s", path) + break + } } - for _, excludedPath := range excludedGlob { - log.Println("Adding excluded path:", excludedPath) - excludedPaths[excludedPath]++ - if excludedPaths[excludedPath] > 1 { - log.Println("Overlapping excluded path:", excludedPath) + + if !isExcluded { + if hostMountPrefix != "" { + pt, err := os.Readlink(path) + if err == nil { + path = filepath.Join(hostMountPrefix, pt) + } + // If err != nil (e.g. not a symlink), we silently keep the original unprefixed path } + files = append(files, NewFile(path, source, true)) } } + return files, nil +} + +// doublestarWalk walks the file system based on the given pattern, supporting ** matching, +// with protection mechanisms like maxTraverseLimit, depth limiting, and permission error ignoring. +func (p *Provider) doublestarWalk(pattern string) ([]string, error) { + slashPattern := filepath.ToSlash(pattern) + base, _ := doublestar.SplitPattern(slashPattern) + if base == "" { + base = "." + } + base = filepath.Clean(filepath.FromSlash(base)) + + var paths []string + traverseCount := 0 + + err := filepath.WalkDir(base, func(path string, d os.DirEntry, err error) error { + traverseCount++ + if traverseCount > p.maxTraverseLimit { + return ErrMaxTraverseLimit + } - for i, path := range paths { - if v := os.Getenv("HOST_MOUNT_PREFIX"); v != "" { - p, err := os.Readlink(path) - if err == nil { - paths[i] = filepath.Join(v, p) + if err != nil { + if errors.Is(err, os.ErrPermission) { + fileProviderPermissionDeniedTotal.Inc() + log.Printf("D! Permission denied while scanning %s, ignoring", path) + return nil // ignore and continue } + return err } - if excludedPaths[path] == 0 || excludedPaths[paths[i]] == 0 { - files = append(files, NewFile(paths[i], source, true)) + + if d.IsDir() { + rel, err := filepath.Rel(base, path) + if err == nil && rel != "." { + depth := len(strings.Split(filepath.ToSlash(rel), "/")) + // Limit maximum recursion depth + if depth > p.maxDepthLimit { + log.Printf("D! Skipping directory %s: depth %d exceeds limit %d", path, depth, p.maxDepthLimit) + return filepath.SkipDir + } + } + } else { + ok, _ := doublestar.Match(filepath.ToSlash(pattern), filepath.ToSlash(path)) + if ok { + paths = append(paths, path) + } } + return nil + }) + + if errors.Is(err, ErrMaxTraverseLimit) { + log.Printf("W! Max traverse limit (%d) reached while scanning pattern: %s, returning %d partial results", p.maxTraverseLimit, pattern, len(paths)) + return paths, nil } - return files, nil + + if err != nil && errors.Is(err, os.ErrNotExist) { + log.Printf("D! Base path does not exist for pattern: %s", pattern) + return nil, nil + } + + return paths, err } // exists returns true if the file at path filePath exists diff --git a/logs/input/file/file_provider_test.go b/logs/input/file/file_provider_test.go new file mode 100644 index 000000000..ad43a8f9d --- /dev/null +++ b/logs/input/file/file_provider_test.go @@ -0,0 +1,191 @@ +package file + +import ( + "os" + "path/filepath" + "runtime" + "testing" + + logsconfig "flashcat.cloud/categraf/config/logs" +) + +func TestDoublestarWalk(t *testing.T) { + // Create a temporary directory for tests + tmpDir, err := os.MkdirTemp("", "categraf-doublestar-test") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + // Create test files + filesToCreate := []string{ + "a.log", + "b.txt", + "sub/c.log", + "sub/sub2/d.log", + "sub/sub2/sub3/e.log", + "sub/sub2/sub3/sub4/f.log", + } + + for _, f := range filesToCreate { + path := filepath.Join(tmpDir, filepath.FromSlash(f)) + err := os.MkdirAll(filepath.Dir(path), 0755) + if err != nil { + t.Fatalf("Failed to create dirs for %s: %v", path, err) + } + err = os.WriteFile(path, []byte("test"), 0644) + if err != nil { + t.Fatalf("Failed to write file %s: %v", path, err) + } + } + + provider := NewProvider(100, 10000, 2) // set maxDepthLimit to 2 + + pattern := filepath.Join(tmpDir, "**", "*.log") + paths, err := provider.doublestarWalk(pattern) + if err != nil { + t.Fatalf("doublestarWalk() unexpected error: %v", err) + } + // depth 0: a.log, depth 1: c.log (sub/), depth 2: d.log (sub/sub2/) — all within limit + if len(paths) != 3 { + t.Errorf("doublestarWalk() returned %d paths, want 3", len(paths)) + } +} + +func TestDoublestarWalk_TraverseLimit(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "categraf-traverse-limit-test") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + for _, f := range []string{"a.log", "sub/b.log", "sub/sub2/c.log"} { + path := filepath.Join(tmpDir, filepath.FromSlash(f)) + if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil { + t.Fatalf("Failed to create dirs: %v", err) + } + if err := os.WriteFile(path, []byte("test"), 0644); err != nil { + t.Fatalf("Failed to write file: %v", err) + } + } + + provider := NewProvider(100, 2, 8) // traverse limit = 2 (very small) + pattern := filepath.Join(tmpDir, "**", "*.log") + + paths, err := provider.doublestarWalk(pattern) + if err != nil { + t.Fatalf("doublestarWalk() should not return error on traverse limit, got: %v", err) + } + // With limit=2, WalkDir visits: base dir (count=1), then "a.log" lexically first (count=2, matched). + // Next entry triggers abort (count=3 > 2). Exactly 1 file should be returned. + if len(paths) != 1 { + t.Errorf("expected exactly 1 partial result due to traverse limit, got %d", len(paths)) + } +} + +func TestSearchFiles_ShortCircuitAndExclude(t *testing.T) { + tmpDir, err := os.MkdirTemp("", "categraf-search-test") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + filesToCreate := []string{ + "a.log", + "b.log", + "c.txt", + "sub/d.log", + } + + for _, f := range filesToCreate { + path := filepath.Join(tmpDir, filepath.FromSlash(f)) + if err := os.MkdirAll(filepath.Dir(path), 0755); err != nil { + t.Fatalf("Failed to create dirs for %s: %v", path, err) + } + if err := os.WriteFile(path, []byte("test"), 0644); err != nil { + t.Fatalf("Failed to write file %s: %v", path, err) + } + } + + provider := NewProvider(100, 10000, 8) + + // Test 1: Non-recursive pattern (no **) + sourceNonRec := logsconfig.NewLogSource("", &logsconfig.LogsConfig{ + Path: filepath.Join(tmpDir, "*.log"), + }) + files, err := provider.searchFiles(sourceNonRec.Config.Path, sourceNonRec) + if err != nil { + t.Fatalf("searchFiles failed: %v", err) + } + if len(files) != 2 { // a.log, b.log + t.Errorf("expected 2 files, got %d", len(files)) + } + + // Test 2: Recursive pattern with Exclude + sourceRecExclude := logsconfig.NewLogSource("", &logsconfig.LogsConfig{ + Path: filepath.Join(tmpDir, "**", "*.log"), + ExcludePaths: []string{ + filepath.ToSlash(filepath.Join(tmpDir, "b.log")), + filepath.ToSlash(filepath.Join(tmpDir, "sub", "*.log")), + }, + }) + files2, err := provider.searchFiles(sourceRecExclude.Config.Path, sourceRecExclude) + if err != nil { + t.Fatalf("searchFiles failed: %v", err) + } + if len(files2) != 1 { // only a.log should remain + t.Errorf("expected 1 file, got %d", len(files2)) + } +} + +func TestDoublestarWalk_PermissionDenied(t *testing.T) { + if runtime.GOOS == "windows" { + t.Skip("Skipping permission test on Windows") + } + + tmpDir, err := os.MkdirTemp("", "categraf-perm-test") + if err != nil { + t.Fatalf("Failed to create temp dir: %v", err) + } + defer os.RemoveAll(tmpDir) + + noPermDir := filepath.Join(tmpDir, "noperm") + err = os.MkdirAll(noPermDir, 0755) + if err != nil { + t.Fatalf("Failed to create dirs for %s: %v", noPermDir, err) + } + if err := os.WriteFile(filepath.Join(noPermDir, "hidden.log"), []byte("test"), 0644); err != nil { + t.Fatalf("Failed to write hidden.log: %v", err) + } + + allowedDir := filepath.Join(tmpDir, "allowed") + if err := os.MkdirAll(allowedDir, 0755); err != nil { + t.Fatalf("Failed to create dirs for %s: %v", allowedDir, err) + } + if err := os.WriteFile(filepath.Join(allowedDir, "visible.log"), []byte("test"), 0644); err != nil { + t.Fatalf("Failed to write visible.log: %v", err) + } + + // Remove permissions from noPermDir + if err := os.Chmod(noPermDir, 0000); err != nil { + t.Fatalf("Failed to chmod %s: %v", noPermDir, err) + } + defer os.Chmod(noPermDir, 0755) // restore for cleanup + + // Probe if chmod worked (e.g. root user might still be able to read) + if _, err := os.ReadDir(noPermDir); err == nil { + t.Skip("Skipping permission test because os.Chmod(0000) did not restrict read access (running as root?)") + } + + provider := NewProvider(100, 10000, 8) + pattern := filepath.Join(tmpDir, "**", "*.log") + + paths, err := provider.doublestarWalk(pattern) + if err != nil { + t.Fatalf("doublestarWalk returned error instead of ignoring permission: %v", err) + } + + if len(paths) != 1 { + t.Errorf("expected 1 file, got %d", len(paths)) + } +} diff --git a/logs/input/file/rotate_windows.go b/logs/input/file/rotate_windows.go index 109788376..9d98c0ea8 100644 --- a/logs/input/file/rotate_windows.go +++ b/logs/input/file/rotate_windows.go @@ -9,10 +9,74 @@ package file import ( "os" + "runtime" + + "golang.org/x/sys/windows" ) -// DidRotate is not implemented on windows, log rotations are handled by the -// tailer for now. +type windowsFileID struct { + volumeSerialNumber uint32 + fileIndexHigh uint32 + fileIndexLow uint32 +} + +// DidRotate returns true if the file has been log-rotated. +// When a log rotation occurs, the file can be either: +// - renamed and recreated +// - removed and recreated +// - truncated func DidRotate(file *os.File, lastReadOffset int64) (bool, error) { - return false, nil + if file == nil { + return false, nil + } + + currentFile, err := openFile(file.Name()) + if err != nil { + if os.IsNotExist(err) { + return false, nil + } + return false, err + } + defer currentFile.Close() + + currentInfo, err := currentFile.Stat() + if err != nil { + return false, err + } + + oldInfo, err := file.Stat() + if err != nil { + return true, nil + } + + truncated := currentInfo.Size() < lastReadOffset || oldInfo.Size() < lastReadOffset + if truncated { + return true, nil + } + + oldID, err := getWindowsFileID(file) + if err != nil { + return false, err + } + currentID, err := getWindowsFileID(currentFile) + if err != nil { + return false, err + } + + return oldID != currentID, nil +} + +func getWindowsFileID(file *os.File) (windowsFileID, error) { + var info windows.ByHandleFileInformation + err := windows.GetFileInformationByHandle(windows.Handle(file.Fd()), &info) + if err != nil { + return windowsFileID{}, err + } + runtime.KeepAlive(file) + + return windowsFileID{ + volumeSerialNumber: info.VolumeSerialNumber, + fileIndexHigh: info.FileIndexHigh, + fileIndexLow: info.FileIndexLow, + }, nil } diff --git a/logs/input/file/scanner.go b/logs/input/file/scanner.go index 4d858d6f9..44d4276c6 100644 --- a/logs/input/file/scanner.go +++ b/logs/input/file/scanner.go @@ -56,14 +56,14 @@ type Scanner struct { } // NewScanner returns a new scanner. -func NewScanner(sources *logsconfig.LogSources, tailingLimit int, pipelineProvider pipeline.Provider, registry auditor.Registry, +func NewScanner(sources *logsconfig.LogSources, tailingLimit int, maxTraverseLimit int, maxDepthLimit int, pipelineProvider pipeline.Provider, registry auditor.Registry, tailerSleepDuration time.Duration, validatePodContainerID bool, scanPeriod time.Duration) *Scanner { return &Scanner{ pipelineProvider: pipelineProvider, tailingLimit: tailingLimit, addedSources: sources.GetAddedForType(logsconfig.FileType), removedSources: sources.GetRemovedForType(logsconfig.FileType), - fileProvider: NewProvider(tailingLimit), + fileProvider: NewProvider(tailingLimit, maxTraverseLimit, maxDepthLimit), tailers: make(map[string]*Tailer), registry: registry, tailerSleepDuration: tailerSleepDuration, @@ -386,4 +386,4 @@ func (s *Scanner) createTailer(file *File, outputChan chan *message.Message) *Ta func (s *Scanner) createRotatedTailer(file *File, outputChan chan *message.Message, pattern *regexp.Regexp) *Tailer { return NewTailer(outputChan, file, s.tailerSleepDuration, NewDecoderFromSourceWithPattern(file.Source, pattern)) -} \ No newline at end of file +} diff --git a/logs/input/file/tailer.go b/logs/input/file/tailer.go index bafb77b45..d5edc7d5b 100644 --- a/logs/input/file/tailer.go +++ b/logs/input/file/tailer.go @@ -78,11 +78,10 @@ func NewDecoderFromSourceWithPattern(source *logsconfig.LogSource, multiLinePatt case logsconfig.KubernetesSourceType: if source.GetcontainerdFlg() == "Y" { lineParser = kubernetes.Parser - matcher = &decoder.NewLineMatcher{} } else { lineParser = kubernetes.JSONParser - matcher = &decoder.NewLineMatcher{} } + matcher = &decoder.NewLineMatcher{} // case logsconfig.DockerSourceType: // lineParser = docker.JSONParser // matcher = &decoder.NewLineMatcher{} @@ -236,7 +235,9 @@ func (t *Tailer) startStopTimer() { // onStop finishes to stop the tailer func (t *Tailer) onStop() { - t.osFile.Close() + if t.osFile != nil { + t.osFile.Close() + } t.decoder.Stop() log.Println("Closed", t.file.Path, "for tailer key", t.file.GetScanKey(), "read", t.bytesRead, "bytes and", t.decoder.GetLineCount(), "lines") } diff --git a/logs/input/file/tailer_windows.go b/logs/input/file/tailer_windows.go index 7f7f6a8bd..866b44148 100644 --- a/logs/input/file/tailer_windows.go +++ b/logs/input/file/tailer_windows.go @@ -13,6 +13,7 @@ import ( "log" "os" "path/filepath" + "sync/atomic" "flashcat.cloud/categraf/logs/decoder" ) @@ -33,9 +34,16 @@ func (t *Tailer) setup(offset int64, whence int) error { if err != nil { return err } - filePos, _ := f.Seek(offset, whence) - f.Close() + filePos, err := f.Seek(offset, whence) + if err != nil { + f.Close() + return err + } + // Keep this handle open for rotation detection and to drain the old file + // after it has been renamed. Normal reads open the path briefly so they + // keep following the current file. + t.osFile = f t.readOffset = filePos t.decodedOffset = filePos @@ -43,12 +51,20 @@ func (t *Tailer) setup(offset int64, whence int) error { } func (t *Tailer) readAvailable() (int, error) { + if atomic.LoadInt32(&t.didFileRotate) != 0 && t.osFile != nil { + return t.readFromFile(t.osFile, false) + } + f, err := openFile(t.fullpath) if err != nil { return 0, err } defer f.Close() + return t.readFromFile(f, true) +} + +func (t *Tailer) readFromFile(f *os.File, resetOffsetOnTruncate bool) (int, error) { st, err := f.Stat() if err != nil { log.Println("Error stat()ing file", err) @@ -57,16 +73,19 @@ func (t *Tailer) readAvailable() (int, error) { sz := st.Size() offset := t.GetReadOffset() - if sz == 0 { + if resetOffsetOnTruncate && sz == 0 { log.Println("File size now zero, resetting offset") t.SetReadOffset(0) t.SetDecodedOffset(0) - } else if sz < offset { + } else if resetOffsetOnTruncate && sz < offset { log.Println("Offset off end of file, resetting") t.SetReadOffset(0) t.SetDecodedOffset(0) } - f.Seek(t.GetReadOffset(), io.SeekStart) + if _, err := f.Seek(t.GetReadOffset(), io.SeekStart); err != nil { + log.Println("Error seeking file", err) + return 0, err + } bytes := 0 for { diff --git a/logs/util/date_format.go b/logs/util/date_format.go new file mode 100644 index 000000000..354d54b89 --- /dev/null +++ b/logs/util/date_format.go @@ -0,0 +1,116 @@ +//go:build !no_logs + +package util + +import ( + "fmt" + "regexp" + "strings" + "time" +) + +var ( + // 匹配 Logstash 风格的日期占位符 %{+yyyy-MM-dd} + datePatternRegex = regexp.MustCompile(`%\{\+([^}]+)\}`) + + // Logstash (Joda-Time) 到 Go 时间格式的映射 + // 按长度排序，确保长的模式先被替换 + dateFormatMappings = []struct { + joda string + go_ string + }{ + // 年份 + {"yyyy", "2006"}, + {"yy", "06"}, + + // 月份 + {"MM", "01"}, + {"M", "1"}, + + // 日期 + {"dd", "02"}, + {"d", "2"}, + + // 小时 + {"HH", "15"}, + {"H", "15"}, + + // 分钟 + // 12小时制 + {"hh", "03"}, + + // 分钟 + {"mm", "04"}, + {"m", "4"}, + + // 秒 + {"ss", "05"}, + {"s", "5"}, + + // 毫秒 + + // ISO8601 + {"ISO8601", "2006-01-02T15:04:05.000Z"}, + + // 容错：兼容部分用户习惯的大写年份 + {"YYYY", "2006"}, + {"YY", "06"}, + + // 常用的分隔符保持不变 + {"-", "-"}, + {".", "."}, + {"/", "/"}, + {"_", "_"}, + {" ", " "}, + {":", ":"}, + } +) + +// ContainsDatePattern returns true if the path contains a logstash date pattern +func ContainsDatePattern(path string) bool { + return datePatternRegex.MatchString(path) +} + +// ExpandDatePattern replaces the logstash date pattern in the path with the actual date +func ExpandDatePattern(path string, now time.Time) string { + return datePatternRegex.ReplaceAllStringFunc(path, func(match string) string { + // 提取占位符内容（去掉 %{+ 和 }） + pattern := match[3 : len(match)-1] + + currentNow := now + if strings.Contains(pattern, "ISO8601") { + currentNow = currentNow.UTC() + } + + // 毫秒级处理：Go 时间格式化会把 1, 2, 3 等数字当成占位符解析 + // 所以我们先将其替换为安全的无冲突占位符，等 Format 完成后再替换为真实毫秒 + pattern = strings.ReplaceAll(pattern, "SSS", "_XXX_") + + // 转换为 Go 时间格式 + goFormat := convertJodaToGoFormat(pattern) + + // 格式化时间 + formatted := currentNow.Format(goFormat) + + millis := fmt.Sprintf("%03d", currentNow.Nanosecond()/1e6) + return strings.ReplaceAll(formatted, "_XXX_", millis) + }) +} + +// convertJodaToGoFormat 将 Joda-Time 格式转换为 Go 时间格式 +func convertJodaToGoFormat(jodaFormat string) string { + result := jodaFormat + + // 按照映射表顺序替换（长的先替换，避免冲突） + for _, mapping := range dateFormatMappings { + result = strings.ReplaceAll(result, mapping.joda, mapping.go_) + } + + return result +} + +// parseDatePattern 解析并验证日期模式（用于调试和错误报告） +func parseDatePattern(path string) []string { + matches := datePatternRegex.FindAllString(path, -1) + return matches +} diff --git a/logs/util/date_format_test.go b/logs/util/date_format_test.go new file mode 100644 index 000000000..ace49eebd --- /dev/null +++ b/logs/util/date_format_test.go @@ -0,0 +1,61 @@ +//go:build !no_logs + +package util + +import ( + "testing" + "time" +) + +func TestExpandDatePattern(t *testing.T) { + now, _ := time.Parse(time.RFC3339, "2026-06-18T15:04:05.123Z") + + tests := []struct { + name string + path string + expected string + }{ + { + name: "YYYY.MM.dd", + path: "/var/log/app-%{+YYYY.MM.dd}.log", + expected: "/var/log/app-2026.06.18.log", + }, + { + name: "yyyy-MM-dd", + path: "/var/log/app-%{+yyyy-MM-dd}.log", + expected: "/var/log/app-2026-06-18.log", + }, + { + name: "ISO8601", + path: "/var/log/app-%{+ISO8601}.log", + expected: "/var/log/app-2026-06-18T15:04:05.123Z.log", + }, + { + name: "hh-mm-ss-SSS", + path: "/var/log/app-%{+hh-mm-ss-SSS}.log", + expected: "/var/log/app-03-04-05-123.log", + }, + { + name: "No pattern", + path: "/var/log/app.log", + expected: "/var/log/app.log", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := ExpandDatePattern(tt.path, now); got != tt.expected { + t.Errorf("ExpandDatePattern() = %v, want %v", got, tt.expected) + } + }) + } +} + +func TestContainsDatePattern(t *testing.T) { + if !ContainsDatePattern("/var/log/app-%{+yyyy-MM-dd}.log") { + t.Error("Expected true, got false") + } + if ContainsDatePattern("/var/log/app.log") { + t.Error("Expected false, got true") + } +} From 4c437af998c9b0a6daa190af740a61b867801557 Mon Sep 17 00:00:00 2001 From: kongfei605 Date: Thu, 18 Jun 2026 13:41:01 +0800 Subject: [PATCH 2/5] "ci: bump golangci-lint version to v1.61 --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index b2ff3209a..c3d5211cf 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -47,7 +47,7 @@ jobs: - name: Run golang-ci-lint uses: golangci/golangci-lint-action@v3.6.0 with: - version: v1.53 + version: v1.61 args: --timeout=10m --tests=false only-new-issues: true skip-pkg-cache: true From a50012777f77dea04cd9b8d761b38b473dbdae98 Mon Sep 17 00:00:00 2001 From: kongfei605 Date: Thu, 18 Jun 2026 13:53:50 +0800 Subject: [PATCH 3/5] ci: update golangci-lint workflow for Go 1.25 --- .github/workflows/ci.yml | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index c3d5211cf..63abfd3c8 100644 --- a/.github/workflows/ci.yml +++ b/.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.61 + 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 From cdda395b7bfad2d32b166896850dede93c44f451 Mon Sep 17 00:00:00 2001 From: kongfei605 Date: Thu, 18 Jun 2026 14:09:38 +0800 Subject: [PATCH 4/5] fix(logs): resolve golangci-lint warnings for errcheck and unused code --- logs/input/file/tailer.go | 2 +- logs/util/date_format.go | 6 ------ 2 files changed, 1 insertion(+), 7 deletions(-) diff --git a/logs/input/file/tailer.go b/logs/input/file/tailer.go index d5edc7d5b..7ab91ba0b 100644 --- a/logs/input/file/tailer.go +++ b/logs/input/file/tailer.go @@ -236,7 +236,7 @@ func (t *Tailer) startStopTimer() { // onStop finishes to stop the tailer func (t *Tailer) onStop() { if t.osFile != nil { - t.osFile.Close() + _ = t.osFile.Close() } t.decoder.Stop() log.Println("Closed", t.file.Path, "for tailer key", t.file.GetScanKey(), "read", t.bytesRead, "bytes and", t.decoder.GetLineCount(), "lines") diff --git a/logs/util/date_format.go b/logs/util/date_format.go index 354d54b89..bdae517bd 100644 --- a/logs/util/date_format.go +++ b/logs/util/date_format.go @@ -108,9 +108,3 @@ func convertJodaToGoFormat(jodaFormat string) string { return result } - -// parseDatePattern 解析并验证日期模式（用于调试和错误报告） -func parseDatePattern(path string) []string { - matches := datePatternRegex.FindAllString(path, -1) - return matches -} From eab3e142ac95f8bc6e57a750297561c4db8ccefc Mon Sep 17 00:00:00 2001 From: kongfei605 Date: Thu, 18 Jun 2026 15:26:16 +0800 Subject: [PATCH 5/5] fix(ci): resolve slog and elasticsearch go vet failures --- .../collector/data_stream_test.go | 4 +- inputs/elasticsearch/collector/ilm_test.go | 3 +- .../elasticsearch/collector/indices_test.go | 5 +- inputs/elasticsearch/collector/nodes_test.go | 2 +- inputs/elasticsearch/collector/slm.go | 26 ++++--- inputs/elasticsearch/collector/slm_test.go | 9 ++- .../elasticsearch/collector/snapshots_test.go | 13 ++-- .../pkg/clusterinfo/clusterinfo_test.go | 11 ++- prometheus/prometheus.go | 76 +++++++++---------- 9 files changed, 79 insertions(+), 70 deletions(-) diff --git a/inputs/elasticsearch/collector/data_stream_test.go b/inputs/elasticsearch/collector/data_stream_test.go index 6fdda5fe3..9b2f5e3b7 100644 --- a/inputs/elasticsearch/collector/data_stream_test.go +++ b/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) } }) diff --git a/inputs/elasticsearch/collector/ilm_test.go b/inputs/elasticsearch/collector/ilm_test.go index 53e42b5fc..5922590a9 100644 --- a/inputs/elasticsearch/collector/ilm_test.go +++ b/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) } diff --git a/inputs/elasticsearch/collector/indices_test.go b/inputs/elasticsearch/collector/indices_test.go index 2aa7b5091..c0f1d40b2 100644 --- a/inputs/elasticsearch/collector/indices_test.go +++ b/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) } diff --git a/inputs/elasticsearch/collector/nodes_test.go b/inputs/elasticsearch/collector/nodes_test.go index 983d736e3..d2fc9c77a 100644 --- a/inputs/elasticsearch/collector/nodes_test.go +++ b/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) diff --git a/inputs/elasticsearch/collector/slm.go b/inputs/elasticsearch/collector/slm.go index 1d7f37abb..da1575178 100644 --- a/inputs/elasticsearch/collector/slm.go +++ b/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 } diff --git a/inputs/elasticsearch/collector/slm_test.go b/inputs/elasticsearch/collector/slm_test.go index 957b1050b..8757a4e2b 100644 --- a/inputs/elasticsearch/collector/slm_test.go +++ b/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) } diff --git a/inputs/elasticsearch/collector/snapshots_test.go b/inputs/elasticsearch/collector/snapshots_test.go index 014121907..307b48abc 100644 --- a/inputs/elasticsearch/collector/snapshots_test.go +++ b/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) } }) diff --git a/inputs/elasticsearch/pkg/clusterinfo/clusterinfo_test.go b/inputs/elasticsearch/pkg/clusterinfo/clusterinfo_test.go index 129b2831c..2a6cfa775 100644 --- a/inputs/elasticsearch/pkg/clusterinfo/clusterinfo_test.go +++ b/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) diff --git a/prometheus/prometheus.go b/prometheus/prometheus.go index cf0fb8bcf..5258702b8 100644 --- a/prometheus/prometheus.go +++ b/prometheus/prometheus.go @@ -292,7 +292,7 @@ func (i *safePromQLNoStepSubqueryInterval) Get(int64) int64 { func reloadConfig(filename string, enableExemplarStorage bool, logger *slog.Logger, noStepSuqueryInterval *safePromQLNoStepSubqueryInterval, callback func(bool), rls ...reloader) (err error) { start := time.Now() timings := logger - logger.Info("msg", "Loading configuration file", "filename", filename) + logger.Info("Loading configuration file", "filename", filename) conf, err := config.LoadFile(filename, true, logger) if err != nil { @@ -309,17 +309,17 @@ func reloadConfig(filename string, enableExemplarStorage bool, logger *slog.Logg for _, rl := range rls { rstart := time.Now() if err := rl.reloader(conf); err != nil { - logger.Error("msg", "Failed to apply configuration", "err", err) + logger.Error("Failed to apply configuration", "err", err) failed = true } - timings.With((rl.name), time.Since(rstart)) + timings = timings.With(rl.name, time.Since(rstart)) } if failed { return fmt.Errorf("one or more errors occurred while applying the new configuration (--config.file=%q)", filename) } noStepSuqueryInterval.Set(conf.GlobalConfig.EvaluationInterval) - timings.Info("msg", "Completed loading of configuration file", "filename", filename, "totalDuration", time.Since(start)) + timings.Info("Completed loading of configuration file", "filename", filename, "totalDuration", time.Since(start)) return nil } @@ -366,33 +366,33 @@ func (c *flagConfig) setFeatureListOptions(logger *slog.Logger) error { switch o { case "expand-external-labels": c.enableExpandExternalLabels = true - logger.Info("msg", "Experimental expand-external-labels enabled") + logger.Info("Experimental expand-external-labels enabled") case "exemplar-storage": c.tsdb.EnableExemplarStorage = true - logger.Info("msg", "Experimental in-memory exemplar storage enabled") + logger.Info("Experimental in-memory exemplar storage enabled") case "memory-snapshot-on-shutdown": c.tsdb.EnableMemorySnapshotOnShutdown = true - logger.Info("msg", "Experimental memory snapshot on shutdown enabled") + logger.Info("Experimental memory snapshot on shutdown enabled") case "extra-scrape-metrics": c.scrape.ExtraMetrics = true - logger.Info("msg", "Experimental additional scrape metrics") + logger.Info("Experimental additional scrape metrics") case "new-service-discovery-manager": c.enableNewSDManager = true - logger.Info("msg", "Experimental service discovery manager") + logger.Info("Experimental service discovery manager") case "agent": - logger.Info("msg", "Experimental agent mode enabled.") + logger.Info("Experimental agent mode enabled.") case "promql-per-step-stats": c.enablePerStepStats = true - logger.Info("msg", "Experimental per-step statistics reporting") + logger.Info("Experimental per-step statistics reporting") case "auto-gomaxprocs": c.enableAutoGOMAXPROCS = true - logger.Info("msg", "Automatically set GOMAXPROCS to match Linux container CPU quota") + logger.Info("Automatically set GOMAXPROCS to match Linux container CPU quota") case "": continue case "promql-at-modifier", "promql-negative-offset": - logger.Warn("msg", "This option for --enable-feature is now permanently enabled and therefore a no-op.", "option", o) + logger.Warn("This option for --enable-feature is now permanently enabled and therefore a no-op.", "option", o) default: - logger.Info("msg", "Unknown option for --enable-feature", "option", o) + logger.Info("Unknown option for --enable-feature", "option", o) } } } @@ -561,7 +561,7 @@ func Start() { os.Exit(1) } - notifierManager := notifier.NewManager(&cfg.notifier, logger.With(logger, "component", "notifier")) + notifierManager := notifier.NewManager(&cfg.notifier, logger.With("component", "notifier")) ctxScrape, cancelScrape := context.WithCancel(context.Background()) ctxNotify, cancelNotify := context.WithCancel(context.Background()) @@ -648,7 +648,7 @@ func Start() { scrapeManager, err := scrape.NewManager( &cfg.scrape, - logger.With(logger, "component", "scrape manager"), + logger.With("component", "scrape manager"), logging.NewJSONFileLogger, fanoutStorage, prometheus.DefaultRegisterer, @@ -692,10 +692,10 @@ func Start() { cfg.web.LookbackDelta = time.Duration(cfg.lookbackDelta) cfg.web.IsAgent = true - webHandler := web.New(logger.With(logger, "component", "web"), &cfg.web) + webHandler := web.New(logger.With("component", "web"), &cfg.web) listener, err := webHandler.Listeners() if err != nil { - logger.Info("msg", "Unable to start web listener", "err", err) + logger.Info("Unable to start web listener", "err", err) os.Exit(1) } @@ -762,10 +762,10 @@ func Start() { // Don't forget to release the reloadReady channel so that waiting blocks can exit normally. select { case sig := <-term: - logger.Warn("msg", "Received "+sig.String()+" exiting gracefully...") + logger.Warn("Received " + sig.String() + " exiting gracefully...") reloadReady.Close() case <-webHandler.Quit(): - logger.Warn("msg", "Received termination request via web service, exiting gracefully...") + logger.Warn("Received termination request via web service, exiting gracefully...") case <-cancel: reloadReady.Close() case <-stop: @@ -786,11 +786,11 @@ func Start() { g.Add( func() error { err := discoveryManagerScrape.Run() - logger.Info("msg", "Scrape discovery manager stopped") + logger.Info("Scrape discovery manager stopped") return err }, func(err error) { - logger.Info("msg", "Stopping scrape discovery manager...") + logger.Info("Stopping scrape discovery manager...") cancelScrape() }, ) @@ -800,11 +800,11 @@ func Start() { g.Add( func() error { err := discoveryManagerNotify.Run() - logger.Info("msg", "Notify discovery manager stopped") + logger.Info("Notify discovery manager stopped") return err }, func(err error) { - logger.Info("msg", "Stopping notify discovery manager...") + logger.Info("Stopping notify discovery manager...") cancelNotify() }, ) @@ -820,13 +820,13 @@ func Start() { <-reloadReady.C err := scrapeManager.Run(discoveryManagerScrape.SyncCh()) - logger.Info("msg", "Scrape manager stopped") + logger.Info("Scrape manager stopped") return err }, func(err error) { // Scrape manager needs to be stopped before closing the local TSDB // so that it doesn't try to write samples to a closed storage. - logger.Info("msg", "Stopping scrape manager...") + logger.Info("Stopping scrape manager...") scrapeManager.Stop() }, ) @@ -858,11 +858,11 @@ func Start() { continue } if err := reloadConfig(cfg.configFile, cfg.tsdb.EnableExemplarStorage, logger, noStepSubqueryInterval, callback, reloaders...); err != nil { - logger.Error("msg", "Error reloading config", "err", err) + logger.Error("Error reloading config", "err", err) } case rc := <-webHandler.Reload(): if err := reloadConfig(cfg.configFile, cfg.tsdb.EnableExemplarStorage, logger, noStepSubqueryInterval, callback, reloaders...); err != nil { - logger.Error("msg", "Error reloading config", "err", err) + logger.Error("Error reloading config", "err", err) rc <- err } else { rc <- nil @@ -900,7 +900,7 @@ func Start() { webHandler.SetReady(web.Ready) notifs.DeleteNotification(notifications.StartingUp) - logger.Info("msg", "server is ready.") + logger.Info("server is ready.") <-cancel return nil }, @@ -915,7 +915,7 @@ func Start() { cancel := make(chan struct{}) g.Add( func() error { - logger.Info("msg", "Starting WAL storage ...") + logger.Info("Starting WAL storage ...") if cfg.agent.WALSegmentSize != 0 { if cfg.agent.WALSegmentSize < 10*1024*1024 || cfg.agent.WALSegmentSize > 256*1024*1024 { return errors.New("flag 'storage.agent.wal-segment-size' must be set between 10MB and 256MB") @@ -934,13 +934,13 @@ func Start() { switch fsType := prom_runtime.Statfs(localStoragePath); fsType { case "NFS_SUPER_MAGIC": - logger.Warn("fs_type", fsType, "msg", "This filesystem is not supported and may lead to data corruption and data loss. Please carefully read https://prometheus.io/docs/prometheus/latest/storage/ to learn more about supported filesystems.") + logger.Warn("This filesystem is not supported and may lead to data corruption and data loss. Please carefully read https://prometheus.io/docs/prometheus/latest/storage/ to learn more about supported filesystems.", "fs_type", fsType) default: - logger.Info("fs_type", fsType) + logger.Info("Supported filesystem", "fs_type", fsType) } - logger.Info("msg", "Agent WAL storage started") - logger.Info("msg", "Agent WAL storage options", + logger.Info("Agent WAL storage started") + logger.Info("Agent WAL storage options", "WALSegmentSize", cfg.agent.WALSegmentSize, "WALCompression", cfg.agent.WALCompression, "StripeSize", cfg.agent.StripeSize, @@ -956,7 +956,7 @@ func Start() { }, func(e error) { if err := fanoutStorage.Close(); err != nil { - logger.Error("msg", "Error stopping storage", "err", err) + logger.Error("Error stopping storage", "err", err) } close(cancel) }, @@ -990,7 +990,7 @@ func Start() { <-reloadReady.C notifierManager.Run(discoveryManagerNotify.SyncCh()) - logger.Info("msg", "Notifier manager stopped") + logger.Info("Notifier manager stopped") return nil }, func(err error) { @@ -1000,10 +1000,10 @@ func Start() { } atomic.StoreInt32(&isRunning, 1) if err := g.Run(); err != nil { - logger.Error("err", err) + logger.Error("run error", "err", err) os.Exit(1) } - logger.Info("msg", "See you next time!") + logger.Info("See you next time!") } func Stop() {