From 3d62805035c4b0d577476f81a4d7eb6449a4afdb Mon Sep 17 00:00:00 2001 From: Chris Grindstaff Date: Tue, 2 Dec 2025 11:47:05 -0500 Subject: [PATCH 1/2] feat: disk cache --- cmd/exporters/prometheus/cache.go | 166 ++++++++++++++++++- cmd/exporters/prometheus/cache_test.go | 38 +++++ cmd/exporters/prometheus/disk_cache.go | 47 +++++- cmd/exporters/prometheus/httpd.go | 173 +++----------------- cmd/exporters/prometheus/prometheus.go | 110 ++++--------- cmd/exporters/prometheus/prometheus_test.go | 37 +---- 6 files changed, 301 insertions(+), 270 deletions(-) create mode 100644 cmd/exporters/prometheus/cache_test.go diff --git a/cmd/exporters/prometheus/cache.go b/cmd/exporters/prometheus/cache.go index bccceb91a..9f5da5379 100644 --- a/cmd/exporters/prometheus/cache.go +++ b/cmd/exporters/prometheus/cache.go @@ -5,36 +5,188 @@ package prometheus import ( + "bytes" "github.com/netapp/harvest/v2/pkg/set" + "github.com/netapp/harvest/v2/pkg/slogx" + "io" + "log/slog" + "net/http" + "strings" "sync" "time" ) -type cache struct { - *sync.Mutex +type cacher interface { + getOverview() (*CacheStats, error) + exportMetrics(key string, data [][]byte, names *set.Set) + streamMetrics(w http.ResponseWriter, seen map[string]struct{}, metrics [][]byte) (int, error) + isValid() bool +} + +type memCache struct { + mu *sync.Mutex + logger *slog.Logger data map[string][][]byte timers map[string]time.Time expire time.Duration } -func newCache(d time.Duration) *cache { - c := cache{Mutex: &sync.Mutex{}, expire: d} +func (c *memCache) isValid() bool { + return true +} + +func (c *memCache) getOverview() (*CacheStats, error) { + c.mu.Lock() + cacheData := make(map[string][][]byte) + for key, data := range c.Get() { + cacheData[key] = make([][]byte, len(data)) + copy(cacheData[key], data) + } + c.mu.Unlock() + + stats := &CacheStats{ + UniqueData: make(map[string]map[string][]string), + } + + for key, data := range cacheData { + var collector, object string + + if keys := strings.Split(key, "."); len(keys) == 3 { + collector = keys[0] + object = keys[1] + } else { + continue + } + + // skip metadata + if strings.HasPrefix(object, "metadata_") { + continue + } + + metricNames := set.New() + for _, m := range data { + if x := strings.Split(string(m), "{"); len(x) >= 2 && x[0] != "" { + metricNames.Add(x[0]) + } + } + stats.NumMetrics += metricNames.Size() + + if _, exists := stats.UniqueData[collector]; !exists { + stats.UniqueData[collector] = make(map[string][]string) + stats.NumCollectors++ + } + if _, exists := stats.UniqueData[collector][object]; !exists { + stats.NumObjects++ + } + stats.UniqueData[collector][object] = metricNames.Values() + } + + return stats, nil +} + +func (c *memCache) exportMetrics(key string, data [][]byte, metricNames *set.Set) { + c.Put(key, data, metricNames) +} + +func (c *memCache) streamMetrics(w http.ResponseWriter, tagsSeen map[string]struct{}, metrics [][]byte) (int, error) { + c.mu.Lock() + var count int + if metrics == nil { + // stream all cached metrics + for _, metrics := range c.Get() { + count += c.writeMetrics(w, metrics, tagsSeen) + } + } else { + // stream only provided metrics + count += c.writeMetrics(w, metrics, tagsSeen) + } + + c.mu.Unlock() + + return count, nil +} + +// writeMetrics writes metrics to the writer, skipping duplicates. +// Normally Render() only adds one TYPE/HELP for each metric type. +// Some metric types (e.g., metadata_collector_metrics) are submitted from multiple collectors. +// That causes duplicates that are suppressed in this function. +// The seen map is used to keep track of which metrics have been added. +func (c *memCache) writeMetrics(w io.Writer, metrics [][]byte, tagsSeen map[string]struct{}) int { + + var count int + + for i := 0; i < len(metrics); i++ { + metric := metrics[i] + if bytes.HasPrefix(metric, []byte("# ")) { + + // Find the metric name and check if it has been seen before + var ( + spacesSeen int + space2Index int + ) + + for j := range metric { + if metric[j] == ' ' { + spacesSeen++ + if spacesSeen == 2 { + space2Index = j + } else if spacesSeen == 3 { + name := string(metric[space2Index+1 : j]) + if _, ok := tagsSeen[name]; !ok { + tagsSeen[name] = struct{}{} + c.writeMetric(w, metric) + count++ + if i+1 < len(metrics) { + c.writeMetric(w, metrics[i+1]) + count++ + i++ + } + } + break + } + } + } + } else { + c.writeMetric(w, metric) + count++ + } + } + + return count +} + +func (c *memCache) writeMetric(w io.Writer, data []byte) { + _, err := w.Write(data) + if err != nil { + c.logger.Error("write metrics", slogx.Err(err)) + } + _, err = w.Write([]byte("\n")) + if err != nil { + c.logger.Error("write newline", slogx.Err(err)) + } +} + +func newMemCache(l *slog.Logger, d time.Duration) *memCache { + c := memCache{mu: &sync.Mutex{}, expire: d, logger: l} c.data = make(map[string][][]byte) c.timers = make(map[string]time.Time) return &c } -func (c *cache) Get() map[string][][]byte { +func (c *memCache) Get() map[string][][]byte { c.Clean() return c.data } -func (c *cache) Put(key string, data [][]byte, _ *set.Set) { +func (c *memCache) Put(key string, data [][]byte, _ *set.Set) { + c.mu.Lock() + defer c.mu.Unlock() + c.data[key] = data c.timers[key] = time.Now() } -func (c *cache) Clean() { +func (c *memCache) Clean() { for k, t := range c.timers { if time.Since(t) > c.expire { delete(c.timers, k) diff --git a/cmd/exporters/prometheus/cache_test.go b/cmd/exporters/prometheus/cache_test.go new file mode 100644 index 000000000..02033bf6f --- /dev/null +++ b/cmd/exporters/prometheus/cache_test.go @@ -0,0 +1,38 @@ +package prometheus + +import ( + "github.com/google/go-cmp/cmp" + "github.com/netapp/harvest/v2/assert" + "strings" + "testing" +) + +func Test_memcache_streamMetrics(t *testing.T) { + example := [][]byte{ + []byte(`# HELP some_metric help text`), + []byte(`# TYPE some_metric type`), + []byte(`some_metric{node="node_1"} 0.0`), + []byte(`# HELP some_other_metric help text`), + []byte(`# TYPE some_other_metric type`), + []byte(`some_other_metric{node="node_2"} 0.0`), + []byte(`# HELP some_other_metric DUPLICATE help text`), + []byte(`# TYPE some_other_metric type`), + []byte(`some_other_metric{node="node_3"} 0.0`), + } + + expected := `# HELP some_metric help text +# TYPE some_metric type +some_metric{node="node_1"} 0.0 +# HELP some_other_metric help text +# TYPE some_other_metric type +some_other_metric{node="node_2"} 0.0 +some_other_metric{node="node_3"} 0.0 +` + m := memCache{} + seen := make(map[string]struct{}) + var w strings.Builder + _ = m.writeMetrics(&w, example, seen) + + diff := cmp.Diff(w.String(), expected) + assert.Equal(t, diff, "") +} diff --git a/cmd/exporters/prometheus/disk_cache.go b/cmd/exporters/prometheus/disk_cache.go index b08b9a849..fd0da5eff 100644 --- a/cmd/exporters/prometheus/disk_cache.go +++ b/cmd/exporters/prometheus/disk_cache.go @@ -7,6 +7,7 @@ import ( "github.com/netapp/harvest/v2/pkg/slogx" "io" "log/slog" + "net/http" "os" "path/filepath" "strings" @@ -23,7 +24,7 @@ type CacheStats struct { } type diskCache struct { - *sync.Mutex + mu *sync.Mutex files map[string]string // key -> filepath timers map[string]time.Time // key -> timestamp metricNames map[string]*set.Set // key -> metric names @@ -38,6 +39,37 @@ type diskCache struct { keyReplacer *strings.Replacer } +func (dc *diskCache) isValid() bool { + return dc != nil && dc.baseDir != "" +} + +func (dc *diskCache) getOverview() (*CacheStats, error) { + dc.mu.Lock() + stats, err := dc.GetStats() + dc.mu.Unlock() + if err != nil { + return nil, err + } + + return stats, nil +} + +func (dc *diskCache) exportMetrics(key string, data [][]byte, metricNames *set.Set) { + dc.Put(key, data, metricNames) +} + +func (dc *diskCache) streamMetrics(w http.ResponseWriter, _ map[string]struct{}, metrics [][]byte) (int, error) { + // since the disk cache streams all cached metrics including metadata, we ignore streaming when metrics is not nil + if metrics != nil { + return 0, nil + } + err := dc.streamToWriter(w) + if err != nil { + return 0, err + } + return dc.GetMetricCount(), nil +} + func newDiskCache(d time.Duration, baseDir string, logger *slog.Logger) *diskCache { if d <= 0 { logger.Warn("invalid expire duration, using default 5 minutes", slog.Duration("provided", d)) @@ -56,7 +88,7 @@ func newDiskCache(d time.Duration, baseDir string, logger *slog.Logger) *diskCac ctx, cancel := context.WithCancel(context.Background()) dc := &diskCache{ - Mutex: &sync.Mutex{}, + mu: &sync.Mutex{}, files: make(map[string]string), timers: make(map[string]time.Time), metricNames: make(map[string]*set.Set), @@ -151,6 +183,9 @@ func (dc *diskCache) GetMetricCount() int { // Put stores metrics to disk and updates cache metadata. func (dc *diskCache) Put(key string, data [][]byte, metricNames *set.Set) { + dc.mu.Lock() + defer dc.mu.Unlock() + filePath := dc.generateFilepath(key) if err := dc.writeToDisk(filePath, data); err != nil { @@ -176,8 +211,8 @@ func (dc *diskCache) Put(key string, data [][]byte, metricNames *set.Set) { slog.Int("metrics_count", len(data))) } -// StreamToWriter streams all non-expired cache files to the writer. -func (dc *diskCache) StreamToWriter(w io.Writer) error { +// streamToWriter streams all non-expired cache files to the writer. +func (dc *diskCache) streamToWriter(w io.Writer) error { var resultErr error errorCount := 0 totalCount := 0 @@ -240,8 +275,8 @@ func (dc *diskCache) streamFile(filePath string, w io.Writer) error { } func (dc *diskCache) Clean() { - dc.Lock() - defer dc.Unlock() + dc.mu.Lock() + defer dc.mu.Unlock() for key, timestamp := range dc.timers { if time.Since(timestamp) <= dc.expire { diff --git a/cmd/exporters/prometheus/httpd.go b/cmd/exporters/prometheus/httpd.go index 4b6e62e6e..f8f82e451 100644 --- a/cmd/exporters/prometheus/httpd.go +++ b/cmd/exporters/prometheus/httpd.go @@ -7,10 +7,8 @@ package prometheus import ( - "bytes" "errors" "fmt" - "io" "log/slog" "net" "net/http" @@ -21,7 +19,6 @@ import ( "strings" "time" - "github.com/netapp/harvest/v2/pkg/set" "github.com/netapp/harvest/v2/pkg/slogx" ) @@ -144,30 +141,22 @@ func (p *Prometheus) ServeMetrics(w http.ResponseWriter, r *http.Request) { tagsSeen := make(map[string]struct{}) - if p.useDiskCache { - p.diskCache.Lock() - count = p.diskCache.GetMetricCount() - err := p.diskCache.StreamToWriter(w) - p.diskCache.Unlock() - if err != nil { - p.Logger.Error("failed to stream metrics from disk cache", slogx.Err(err)) - } - } else { - p.memoryCache.Lock() - for _, metrics := range p.memoryCache.Get() { - count += p.writeMetrics(w, metrics, tagsSeen) - } - p.memoryCache.Unlock() + _, err := p.aCache.streamMetrics(w, tagsSeen, nil) + if err != nil { + p.Logger.Error("failed to stream metrics", slogx.Err(err)) } // serve our own metadata // notice that some values are always taken from previous session - md, _ := p.render(p.Metadata) - count += p.writeMetrics(w, md, tagsSeen) + md, _, _ := p.render(p.Metadata) + _, err = p.aCache.streamMetrics(w, tagsSeen, md) + if err != nil { + p.Logger.Error("failed to stream metadata metrics", slogx.Err(err)) + } // update metadata p.Metadata.Reset() - err := p.Metadata.LazySetValueInt64("time", "http", time.Since(start).Microseconds()) + err = p.Metadata.LazySetValueInt64("time", "http", time.Since(start).Microseconds()) if err != nil { p.Logger.Error("metadata time", slogx.Err(err)) } @@ -177,72 +166,18 @@ func (p *Prometheus) ServeMetrics(w http.ResponseWriter, r *http.Request) { } } -// writeMetrics writes metrics to the writer, skipping duplicates. -// Normally Render() only adds one TYPE/HELP for each metric type. -// Some metric types (e.g., metadata_collector_metrics) are submitted from multiple collectors. -// That causes duplicates that are suppressed in this function. -// The seen map is used to keep track of which metrics have been added. -func (p *Prometheus) writeMetrics(w io.Writer, metrics [][]byte, tagsSeen map[string]struct{}) int { - - var count int - - for i := 0; i < len(metrics); i++ { - metric := metrics[i] - if bytes.HasPrefix(metric, []byte("# ")) { - - // Find the metric name and check if it has been seen before - var ( - spacesSeen int - space2Index int - ) - - for j := range metric { - if metric[j] == ' ' { - spacesSeen++ - if spacesSeen == 2 { - space2Index = j - } else if spacesSeen == 3 { - name := string(metric[space2Index+1 : j]) - if _, ok := tagsSeen[name]; !ok { - tagsSeen[name] = struct{}{} - p.writeMetric(w, metric) - count++ - if i+1 < len(metrics) { - p.writeMetric(w, metrics[i+1]) - count++ - i++ - } - } - break - } - } - } - } else { - p.writeMetric(w, metric) - count++ - } - } - - return count -} - -func (p *Prometheus) writeMetric(w io.Writer, data []byte) { - _, err := w.Write(data) - if err != nil { - p.Logger.Error("write metrics", slogx.Err(err)) - return - } - _, err = w.Write([]byte("\n")) - if err != nil { - p.Logger.Error("write newline", slogx.Err(err)) - return - } -} - // ServeInfo provides a human-friendly overview of metric types and source collectors // this is done in a very inefficient way, by "reverse engineering" the metrics. // That's probably ok, since we don't expect this to be called often. func (p *Prometheus) ServeInfo(w http.ResponseWriter, r *http.Request) { + + var ( + numCollectors int + numObjects int + numMetrics int + uniqueData map[string]map[string][]string + ) + start := time.Now() if !p.checkAddr(r.RemoteAddr) { @@ -254,70 +189,16 @@ func (p *Prometheus) ServeInfo(w http.ResponseWriter, r *http.Request) { body := make([]string, 0) - numCollectors := 0 - numObjects := 0 - numMetrics := 0 - - uniqueData := map[string]map[string][]string{} - - if p.useDiskCache { - p.diskCache.Lock() - stats, err := p.diskCache.GetStats() - p.diskCache.Unlock() - if err != nil { - p.Logger.Error("failed to get cache statistics", slogx.Err(err)) - http.Error(w, "Failed to collect cache statistics", http.StatusInternalServerError) - return - } - - numCollectors = stats.NumCollectors - numObjects = stats.NumObjects - numMetrics = stats.NumMetrics - uniqueData = stats.UniqueData - } else { - p.memoryCache.Lock() - cacheData := make(map[string][][]byte) - for key, data := range p.memoryCache.Get() { - cacheData[key] = make([][]byte, len(data)) - copy(cacheData[key], data) - } - p.memoryCache.Unlock() - - p.Logger.Debug("fetching cached elements", slog.Int("count", len(cacheData))) - - for key, data := range cacheData { - var collector, object string - - if keys := strings.Split(key, "."); len(keys) == 3 { - collector = keys[0] - object = keys[1] - } else { - continue - } - - // skip metadata - if strings.HasPrefix(object, "metadata_") { - continue - } - - metricNames := set.New() - for _, m := range data { - if x := strings.Split(string(m), "{"); len(x) >= 2 && x[0] != "" { - metricNames.Add(x[0]) - } - } - numMetrics += metricNames.Size() - - if _, exists := uniqueData[collector]; !exists { - uniqueData[collector] = make(map[string][]string) - numCollectors++ - } - if _, exists := uniqueData[collector][object]; !exists { - numObjects++ - } - uniqueData[collector][object] = metricNames.Values() - } + overview, err := p.aCache.getOverview() + if err != nil { + p.Logger.Error("failed to get cache statistics", slogx.Err(err)) + http.Error(w, "Failed to collect cache statistics", http.StatusInternalServerError) + return } + numCollectors = overview.NumCollectors + numObjects = overview.NumObjects + numMetrics = overview.NumMetrics + uniqueData = overview.UniqueData for col, perObject := range uniqueData { objects := make([]string, 0) @@ -339,7 +220,7 @@ func (p *Prometheus) ServeInfo(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) w.Header().Set("Content-Type", "text/html") - _, err := w.Write([]byte(bodyFlat)) + _, err = w.Write([]byte(bodyFlat)) if err != nil { p.Logger.Error("write info", slogx.Err(err)) } diff --git a/cmd/exporters/prometheus/prometheus.go b/cmd/exporters/prometheus/prometheus.go index 73fa275fa..fe1818c19 100644 --- a/cmd/exporters/prometheus/prometheus.go +++ b/cmd/exporters/prometheus/prometheus.go @@ -49,8 +49,7 @@ const ( type Prometheus struct { *exporter.AbstractExporter - memoryCache *cache - diskCache *diskCache + aCache cacher allowAddrs []string allowAddrsRegex []*regexp.Regexp cacheAddrs map[string]bool @@ -58,37 +57,26 @@ type Prometheus struct { addMetaTags bool globalPrefix string replacer *strings.Replacer - useDiskCache bool } func New(abc *exporter.AbstractExporter) exporter.Exporter { return &Prometheus{AbstractExporter: abc} } -func (p *Prometheus) createCache(d time.Duration) { - if p.useDiskCache { - // Path is mandatory when disk cache is enabled - if p.Params.DiskCache == nil || p.Params.DiskCache.Path == "" { - p.Logger.Error("disk cache enabled but path is not specified") - return - } +func (p *Prometheus) createCacher(dur time.Duration) cacher { + if p.Params.DiskCache != nil && p.Params.DiskCache.Path != "" { + p.Logger.Debug("disk cache enabled - will use disk-based caching for RSS optimization", + slog.String("path", p.Params.DiskCache.Path)) cacheDir := p.Params.DiskCache.Path - // Include poller name in cache directory to avoid collisions between multiple pollers if p.Options.Poller != "" { cacheDir = filepath.Join(cacheDir, p.Options.Poller) } - - p.diskCache = newDiskCache(d, cacheDir, p.Logger) - - if p.diskCache != nil { - p.Logger.Debug("disk cache configured", - slog.String("cacheDir", cacheDir)) - } - } else { - p.memoryCache = newCache(d) + return newDiskCache(dur, cacheDir, p.Logger) } + + return newMemCache(p.Logger, dur) } func (p *Prometheus) Init() error { @@ -128,36 +116,25 @@ func (p *Prometheus) Init() error { p.addMetaTags = true } - // Check if disk cache is enabled (path is mandatory) - if p.Params.DiskCache != nil && p.Params.DiskCache.Path != "" { - p.useDiskCache = true - p.Logger.Debug("disk cache enabled - will use disk-based caching for RSS optimization", - slog.String("path", p.Params.DiskCache.Path)) - } else { - p.useDiskCache = false - p.Logger.Debug("disk cache disabled - using memory-based caching") - } - - // all other parameters are only relevant to the HTTP daemon + maxKeep := cacheMaxKeep + var maxKeepDur time.Duration if x := p.Params.CacheMaxKeep; x != nil { - if d, err := time.ParseDuration(*x); err == nil { - p.Logger.Debug("using custom cache_max_keep", slog.String("cacheMaxKeep", *x)) - p.createCache(d) - } else { - p.Logger.Error("cache_max_keep", slogx.Err(err), slog.String("x", *x)) - } + maxKeep = *x + p.Logger.Debug("using custom cache_max_keep", slog.String("cacheMaxKeep", maxKeep)) } - - if p.memoryCache == nil && p.diskCache == nil { - p.Logger.Debug("using default cache_max_keep", slog.String("cacheMaxKeep", cacheMaxKeep)) - if d, err := time.ParseDuration(cacheMaxKeep); err == nil { - p.createCache(d) - } else { - return err - } + d, err := time.ParseDuration(maxKeep) + if err != nil { + p.Logger.Error("failed to use cache_max_keep duration. Using default", slogx.Err(err), + slog.String("maxKeep", maxKeep), + slog.String("default", cacheMaxKeep), + ) + maxKeepDur, _ = time.ParseDuration(cacheMaxKeep) + } else { + maxKeepDur = d } - if p.memoryCache == nil && p.diskCache == nil { + p.aCache = p.createCacher(maxKeepDur) + if !p.aCache.isValid() { return errs.New(errs.ErrInvalidParam, "cache initialization failed") } @@ -250,9 +227,10 @@ func newReplacer() *strings.Replacer { func (p *Prometheus) Export(data *matrix.Matrix) (exporter.Stats, error) { var ( - metrics [][]byte - stats exporter.Stats - err error + metrics [][]byte + stats exporter.Stats + err error + metricNames *set.Set ) // lock the exporter, to prevent other collectors from writing to us @@ -261,39 +239,15 @@ func (p *Prometheus) Export(data *matrix.Matrix) (exporter.Stats, error) { // render metrics into Prometheus format start := time.Now() - metrics, stats = p.render(data) + metrics, stats, metricNames = p.render(data) // fix render time for metadata d := time.Since(start) - // Extract metric names from matrix for cache statistics - var prefix string - if data.Object == "" { - prefix = strings.TrimSuffix(p.globalPrefix, "_") - } else { - prefix = p.globalPrefix + data.Object - } - - metricNames := set.New() - for _, metric := range data.GetMetrics() { - if metric.IsExportable() { - metricNames.Add(prefix + "_" + metric.GetName()) - } - } - // store metrics in cache key := data.UUID + "." + data.Object + "." + data.Identifier - // lock cache, to prevent HTTPd reading while we are mutating it - if p.useDiskCache { - p.diskCache.Lock() - p.diskCache.Put(key, metrics, metricNames) - p.diskCache.Unlock() - } else { - p.memoryCache.Lock() - p.memoryCache.Put(key, metrics, metricNames) - p.memoryCache.Unlock() - } + p.aCache.exportMetrics(key, metrics, metricNames) // update metadata p.AddExportCount(uint64(len(metrics))) @@ -325,7 +279,7 @@ func (p *Prometheus) Export(data *matrix.Matrix) (exporter.Stats, error) { // volume_read_ops{node="my-node",vol="some_vol"} 2523 // fcp_lif_read_ops{vserver="nas_svm",port_id="e02"} 771 -func (p *Prometheus) render(data *matrix.Matrix) ([][]byte, exporter.Stats) { +func (p *Prometheus) render(data *matrix.Matrix) ([][]byte, exporter.Stats, *set.Set) { var ( rendered [][]byte tagged *set.Set @@ -345,6 +299,7 @@ func (p *Prometheus) render(data *matrix.Matrix) ([][]byte, exporter.Stats) { buf.Grow(4096) globalLabels := make([]string, 0, len(data.GetGlobalLabels())) normalizedLabels = make(map[string][]string) + metricNames := set.New() if p.addMetaTags { tagged = set.New() @@ -402,6 +357,7 @@ func (p *Prometheus) render(data *matrix.Matrix) ([][]byte, exporter.Stats) { if !metric.IsExportable() { continue } + metricNames.Add(prefix + "_" + metric.GetName()) exportableMetrics++ } @@ -735,7 +691,7 @@ func (p *Prometheus) render(data *matrix.Matrix) ([][]byte, exporter.Stats) { RenderedBytes: renderedBytes, } - return rendered, stats + return rendered, stats, metricNames } var numAndUnitRe = regexp.MustCompile(`(\d+)\s*(\w+)`) diff --git a/cmd/exporters/prometheus/prometheus_test.go b/cmd/exporters/prometheus/prometheus_test.go index a38dcb244..1e2f6f75b 100644 --- a/cmd/exporters/prometheus/prometheus_test.go +++ b/cmd/exporters/prometheus/prometheus_test.go @@ -17,37 +17,6 @@ import ( "github.com/netapp/harvest/v2/pkg/matrix" ) -func TestFilterMetaTags(t *testing.T) { - - example := [][]byte{ - []byte(`# HELP some_metric help text`), - []byte(`# TYPE some_metric type`), - []byte(`some_metric{node="node_1"} 0.0`), - []byte(`# HELP some_other_metric help text`), - []byte(`# TYPE some_other_metric type`), - []byte(`some_other_metric{node="node_2"} 0.0`), - []byte(`# HELP some_other_metric DUPLICATE help text`), - []byte(`# TYPE some_other_metric type`), - []byte(`some_other_metric{node="node_3"} 0.0`), - } - - expected := `# HELP some_metric help text -# TYPE some_metric type -some_metric{node="node_1"} 0.0 -# HELP some_other_metric help text -# TYPE some_other_metric type -some_other_metric{node="node_2"} 0.0 -some_other_metric{node="node_3"} 0.0 -` - p := Prometheus{} - seen := make(map[string]struct{}) - var w strings.Builder - _ = p.writeMetrics(&w, example, seen) - - diff := cmp.Diff(w.String(), expected) - assert.Equal(t, diff, "") -} - func TestEscape(t *testing.T) { replacer := newReplacer() @@ -149,7 +118,7 @@ net_app_bike_max_speed{} 3`, "bike"}, prom := p.(*Prometheus) var lines []string - for _, metrics := range prom.memoryCache.Get() { + for _, metrics := range prom.aCache.(*memCache).Get() { for _, metric := range metrics { lines = append(lines, string(metric)) } @@ -187,7 +156,7 @@ netapp_change_log{category="metric",cluster="umeng-aff300-01-02",object="volume" prom := p.(*Prometheus) var lines []string - for _, metrics := range prom.memoryCache.Get() { + for _, metrics := range prom.aCache.(*memCache).Get() { for _, metric := range metrics { lines = append(lines, string(metric)) } @@ -262,7 +231,7 @@ func TestRenderHistogramExample(t *testing.T) { prom := p.(*Prometheus) var lines []string - for _, metrics := range prom.memoryCache.Get() { + for _, metrics := range prom.aCache.(*memCache).Get() { for _, metricLine := range metrics { sline := string(metricLine) if !strings.HasPrefix(sline, "#") { From 79643fa37ab83832acd6a7d7de2bed7086422233 Mon Sep 17 00:00:00 2001 From: Chris Grindstaff Date: Wed, 3 Dec 2025 08:07:12 -0500 Subject: [PATCH 2/2] feat: disk cache --- cmd/exporters/prometheus/disk_cache.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/cmd/exporters/prometheus/disk_cache.go b/cmd/exporters/prometheus/disk_cache.go index fd0da5eff..394e9751c 100644 --- a/cmd/exporters/prometheus/disk_cache.go +++ b/cmd/exporters/prometheus/disk_cache.go @@ -63,6 +63,9 @@ func (dc *diskCache) streamMetrics(w http.ResponseWriter, _ map[string]struct{}, if metrics != nil { return 0, nil } + dc.mu.Lock() + defer dc.mu.Unlock() + err := dc.streamToWriter(w) if err != nil { return 0, err