Refactor state_* metricsets to share response from endpoint

metricbeat/helper/prometheus/prometheus.go

@@ -44,6 +44,8 @@ type Prometheus interface {
 
 	GetProcessedMetrics(mapping *MetricsMapping) ([]common.MapStr, error)
 
+	ProcessMetrics(families []*dto.MetricFamily, mapping *MetricsMapping) ([]common.MapStr, error)
+
 	ReportProcessedMetrics(mapping *MetricsMapping, r mb.ReporterV2) error
 }
 
@@ -139,11 +141,7 @@ type MetricsMapping struct {
 	ExtraFields map[string]string
 }
 
-func (p *prometheus) GetProcessedMetrics(mapping *MetricsMapping) ([]common.MapStr, error) {
-	families, err := p.GetFamilies()
-	if err != nil {
-		return nil, err
-	}
+func (p *prometheus) ProcessMetrics(families []*dto.MetricFamily, mapping *MetricsMapping) ([]common.MapStr, error) {
 
 	eventsMap := map[string]common.MapStr{}
 	infoMetrics := []*infoMetricData{}
@@ -260,6 +258,14 @@ func (p *prometheus) GetProcessedMetrics(mapping *MetricsMapping) ([]common.MapS
 	return events, nil
 }
 
+func (p *prometheus) GetProcessedMetrics(mapping *MetricsMapping) ([]common.MapStr, error) {
+	families, err := p.GetFamilies()
+	if err != nil {
+		return nil, err
+	}
+	return p.ProcessMetrics(families, mapping)
+}
+
 // infoMetricData keeps data about an infoMetric
 type infoMetricData struct {
 	Labels common.MapStr

metricbeat/module/kubernetes/kubernetes.go

@@ -0,0 +1,98 @@
+// Licensed to Elasticsearch B.V. under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Elasticsearch B.V. licenses this file to you 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.
+
+package kubernetes
+
+import (
+	"github.com/elastic/beats/v7/libbeat/logp"
+	"sync"
+	"time"
+	"fmt"
+
+	dto "github.com/prometheus/client_model/go"
+
+	p "github.com/elastic/beats/v7/metricbeat/helper/prometheus"
+	"github.com/elastic/beats/v7/metricbeat/mb"
+)
+
+func init() {
+	// Register the ModuleFactory function for the "kubernetes" module.
+	if err := mb.Registry.AddModule("kubernetes", ModuleBuilder()); err != nil {
+		panic(err)
+	}
+}
+
+type Module interface {
+	mb.Module
+	GetSharedFamilies(prometheus p.Prometheus, ms string) ([]*dto.MetricFamily, error)
+}
+
+type familiesCache struct {
+	sharedFamilies     []*dto.MetricFamily
+	lastFetchErr       error
+	lastFetchTimestamp time.Time
+	setter string
+}
+
+type cacheMap map[string]*familiesCache
+
+type module struct {
+	mb.BaseModule
+	lock sync.Mutex
+
+	fCache cacheMap
+	logger  *logp.Logger
+}
+
+func ModuleBuilder() func(base mb.BaseModule) (mb.Module, error) {
+	sharedFamiliesCache := make(cacheMap)
+	return func(base mb.BaseModule) (mb.Module, error) {
+		hash := fmt.Sprintf("%s%s", base.Config().Period, base.Config().Hosts)
+		sharedFamiliesCache[hash] = &familiesCache{}
+		m := module{
+			BaseModule: base,
+			logger : logp.NewLogger(fmt.Sprintf("debug (%s)", hash)),
+			fCache: sharedFamiliesCache,
+		}
+		m.logger.Warn("Building module now with  ", base.Config())
+		return &m, nil
+	}
+}
+
+func (m *module) GetSharedFamilies(prometheus p.Prometheus, ms string) ([]*dto.MetricFamily, error) {
+	m.lock.Lock()
+	defer m.lock.Unlock()
+
+	now := time.Now()
+	hash := fmt.Sprintf("%s%s", m.BaseModule.Config().Period, m.BaseModule.Config().Hosts)
+	fCache := m.fCache[hash]
+
+	if ms != fCache.setter {
+		m.logger.Warn("DIFF[ms!=cacheSetter]: ", ms, " != ", fCache.setter)
+	}
+
+	if fCache.lastFetchTimestamp.IsZero() || now.Sub(fCache.lastFetchTimestamp) > m.Config().Period {
+		m.logger.Warn("FETCH families for ms: ", ms, ". Last setter was ", fCache.setter)
+		fCache.sharedFamilies, fCache.lastFetchErr = prometheus.GetFamilies()
+		fCache.lastFetchTimestamp = now
+		fCache.setter = ms
+	} else {
+		m.logger.Warn("REUSE families for ms: ", ms, ". Last setter was ", fCache.setter)
+	}
+
+	return fCache.sharedFamilies, fCache.lastFetchErr
+}

metricbeat/module/kubernetes/state_container/state_container.go

@@ -18,6 +18,7 @@
 package state_container
 
 import (
+	"fmt"
 	"strings"
 
 	"github.com/pkg/errors"
@@ -26,6 +27,7 @@ import (
 	p "github.com/elastic/beats/v7/metricbeat/helper/prometheus"
 	"github.com/elastic/beats/v7/metricbeat/mb"
 	"github.com/elastic/beats/v7/metricbeat/mb/parse"
+	k8smod "github.com/elastic/beats/v7/metricbeat/module/kubernetes"
 	"github.com/elastic/beats/v7/metricbeat/module/kubernetes/util"
 )
 
@@ -89,20 +91,26 @@ type MetricSet struct {
 	mb.BaseMetricSet
 	prometheus p.Prometheus
 	enricher   util.Enricher
+	mod        k8smod.Module
 }
 
 // New create a new instance of the MetricSet
-// Part of new is also setting up the configuration by processing additional
+// Part of newF is also setting up the configuration by processing additional
 // configuration entries if needed.
 func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	prometheus, err := p.NewPrometheusClient(base)
 	if err != nil {
 		return nil, err
 	}
+	mod, ok := base.Module().(k8smod.Module)
+	if !ok {
+		return nil, fmt.Errorf("must be child of kubernetes module")
+	}
 	return &MetricSet{
 		BaseMetricSet: base,
 		prometheus:    prometheus,
 		enricher:      util.NewContainerMetadataEnricher(base, false),
+		mod:           mod,
 	}, nil
 }
 
@@ -112,7 +120,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 func (m *MetricSet) Fetch(reporter mb.ReporterV2) error {
 	m.enricher.Start()
 
-	events, err := m.prometheus.GetProcessedMetrics(mapping)
+	families, err := m.mod.GetSharedFamilies(m.prometheus, "state_container")
+	if err != nil {
+		return errors.Wrap(err, "error getting families")
+	}
+	events, err := m.prometheus.ProcessMetrics(families, mapping)
 	if err != nil {
 		return errors.Wrap(err, "error getting event")
 	}

metricbeat/module/kubernetes/state_pod/state_pod.go

@@ -18,11 +18,13 @@
 package state_pod
 
 import (
+	"fmt"
 	"github.com/elastic/beats/v7/libbeat/common"
 	"github.com/elastic/beats/v7/libbeat/common/kubernetes"
 	p "github.com/elastic/beats/v7/metricbeat/helper/prometheus"
 	"github.com/elastic/beats/v7/metricbeat/mb"
 	"github.com/elastic/beats/v7/metricbeat/mb/parse"
+	k8smod "github.com/elastic/beats/v7/metricbeat/module/kubernetes"
 	"github.com/elastic/beats/v7/metricbeat/module/kubernetes/util"
 )
 
@@ -72,6 +74,7 @@ type MetricSet struct {
 	mb.BaseMetricSet
 	prometheus p.Prometheus
 	enricher   util.Enricher
+	mod        k8smod.Module
 }
 
 // New create a new instance of the MetricSet
@@ -82,11 +85,15 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	if err != nil {
 		return nil, err
 	}
-
+	mod, ok := base.Module().(k8smod.Module)
+	if !ok {
+		return nil, fmt.Errorf("must be child of kubernetes module")
+	}
 	return &MetricSet{
 		BaseMetricSet: base,
 		prometheus:    prometheus,
 		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.Pod{}, false),
+		mod:           mod,
 	}, nil
 }
 
@@ -96,7 +103,13 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 func (m *MetricSet) Fetch(reporter mb.ReporterV2) {
 	m.enricher.Start()
 
-	events, err := m.prometheus.GetProcessedMetrics(mapping)
+	families, err := m.mod.GetSharedFamilies(m.prometheus, "state_pod")
+	if err != nil {
+		m.Logger().Error(err)
+		reporter.Error(err)
+		return
+	}
+	events, err := m.prometheus.ProcessMetrics(families, mapping)
 	if err != nil {
 		m.Logger().Error(err)
 		reporter.Error(err)