textfile: report duplicates and preserve valid metrics (#2590)

Signed-off-by: Jan-Otto Kröpke <mail@jkroepke.de>

internal/collector/textfile/duplicates_collect_test.go

@@ -0,0 +1,77 @@
+// SPDX-License-Identifier: Apache-2.0
+//
+// Copyright The Prometheus Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//go:build windows
+
+package textfile
+
+import (
+	"log/slog"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDuplicateAcrossFilesPreservesOtherMetrics(t *testing.T) {
+	t.Parallel()
+
+	directory := t.TempDir()
+	require.NoError(t, os.WriteFile(filepath.Join(directory, "a.prom"), []byte("# HELP probe Shared probe.\nprobe{source=\"same\"} 1\n"), 0o600))
+	require.NoError(t, os.WriteFile(filepath.Join(directory, "b.prom"), []byte("# HELP probe Shared probe.\nprobe{source=\"same\"} 2\nother 9\n"), 0o600))
+	c := New(&Config{TextFileDirectories: []string{directory}})
+	require.NoError(t, c.Build(slog.New(slog.DiscardHandler), nil))
+	wrapper := &duplicateTestCollector{collector: c}
+	registry := prometheus.NewRegistry()
+	registry.MustRegister(wrapper)
+	families, err := registry.Gather()
+	require.NoError(t, err)
+	require.ErrorContains(t, wrapper.err, "duplicate metric")
+	require.ErrorContains(t, wrapper.err, "probe")
+	require.ErrorContains(t, wrapper.err, "a.prom")
+	require.ErrorContains(t, wrapper.err, "b.prom")
+
+	seen := make(map[string]bool)
+	for _, family := range families {
+		seen[family.GetName()] = true
+		switch family.GetName() {
+		case "other":
+			require.InDelta(t, 9, family.GetMetric()[0].GetUntyped().GetValue(), 1e-9)
+		case "probe":
+			require.Len(t, family.GetMetric(), 1)
+			require.InDelta(t, 1, family.GetMetric()[0].GetUntyped().GetValue(), 1e-9)
+		case "windows_textfile_mtime_seconds":
+			require.Len(t, family.GetMetric(), 1)
+			require.Equal(t, "a.prom", family.GetMetric()[0].GetLabel()[0].GetValue())
+		}
+	}
+
+	require.True(t, seen["other"])
+	require.True(t, seen["probe"])
+	require.True(t, seen["windows_textfile_mtime_seconds"])
+}
+
+type duplicateTestCollector struct {
+	collector *Collector
+	err       error
+}
+
+func (c *duplicateTestCollector) Describe(_ chan<- *prometheus.Desc) {}
+
+func (c *duplicateTestCollector) Collect(ch chan<- prometheus.Metric) {
+	c.err = c.collector.Collect(ch, 0)
+}

internal/collector/textfile/textfile.go

@@ -324,11 +324,10 @@ func (cr carriageReturnFilteringReader) Read(p []byte) (int, error) {
 func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error {
 	mTimes := map[string]time.Time{}
 
-	// Create empty metricFamily slice here and append parsedFamilies to it inside the loop.
-	// Once loop is complete, raise error if any duplicates are present.
-	// This will ensure that duplicate metrics are correctly detected between multiple .prom files.
 	var metricFamilies []*dto.MetricFamily
 
+	seriesFiles := make(map[string]string)
+
 	errs := make([]error, 0)
 
 	// Iterate over files and accumulate their metrics.
@@ -338,32 +337,58 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error
 				return fmt.Errorf("error reading directory: %w", err)
 			}
 
-			if !dirEntry.IsDir() && strings.HasSuffix(dirEntry.Name(), ".prom") {
-				c.logger.Debug("Processing file: " + path)
+			if dirEntry.IsDir() || !strings.HasSuffix(dirEntry.Name(), ".prom") {
+				return nil
+			}
 
-				families_array, err := scrapeFile(path, c.logger)
-				if err != nil {
-					errs = append(errs, fmt.Errorf("error scraping file %q: %w", path, err))
+			c.logger.Debug("Processing file: " + path)
 
-					return nil
+			families_array, err := scrapeFile(path, c.logger)
+			if err != nil {
+				errs = append(errs, fmt.Errorf("error scraping file %q: %w", path, err))
+
+				return nil
+			}
+
+			fileInfo, err := os.Stat(path)
+			if err != nil {
+				errs = append(errs, fmt.Errorf("error reading file info %q: %w", path, err))
+
+				return nil
+			}
+
+			if _, hasName := mTimes[fileInfo.Name()]; hasName {
+				errs = append(errs, fmt.Errorf("duplicate filename detected: %q", path))
+
+				return nil
+			}
+
+			fileHasDuplicates := false
+
+			for _, family := range families_array {
+				metrics := family.GetMetric()[:0]
+				for _, metric := range family.GetMetric() {
+					key := seriesKey(family.GetName(), metric.GetLabel())
+					if previousFile, exists := seriesFiles[key]; exists {
+						errs = append(errs, fmt.Errorf("duplicate metric %q with labels %v in %q and %q", family.GetName(), metric.GetLabel(), previousFile, path))
+						fileHasDuplicates = true
+
+						continue
+					}
+
+					seriesFiles[key] = path
+
+					metrics = append(metrics, metric)
 				}
 
-				fileInfo, err := os.Stat(path)
-				if err != nil {
-					errs = append(errs, fmt.Errorf("error reading file info %q: %w", path, err))
-
-					return nil
-				}
-
-				if _, hasName := mTimes[fileInfo.Name()]; hasName {
-					errs = append(errs, fmt.Errorf("duplicate filename detected: %q", path))
-
-					return nil
+				family.Metric = metrics
+				if len(metrics) != 0 {
+					metricFamilies = append(metricFamilies, family)
 				}
+			}
 
+			if !fileHasDuplicates {
 				mTimes[fileInfo.Name()] = fileInfo.ModTime()
-
-				metricFamilies = append(metricFamilies, families_array...)
 			}
 
 			return nil
@@ -375,13 +400,8 @@ func (c *Collector) Collect(ch chan<- prometheus.Metric, _ time.Duration) error
 
 	c.exportMTimes(mTimes, ch)
 
-	// If duplicates are detected across *multiple* files, return error.
-	if duplicateMetricEntry(metricFamilies) {
-		c.logger.Warn("duplicate metrics detected across multiple files")
-	} else {
-		for _, mf := range metricFamilies {
-			c.convertMetricFamily(c.logger, mf, ch)
-		}
+	for _, mf := range metricFamilies {
+		c.convertMetricFamily(c.logger, mf, ch)
 	}
 
 	return errors.Join(errs...)