Refactor state_* metricsets to share response from endpoint

CHANGELOG.next.asciidoc

@@ -987,6 +987,7 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Apache: convert status.total_kbytes to status.total_bytes in fleet mode. {pull}23022[23022]
 - Release MSSQL as GA {pull}23146[23146]
 - Add support for SASL/SCRAM authentication to the Kafka module. {pull}24810[24810]
+- Refactor state_* metricsets to share response from endpoint. {pull}25640[25640]
 - Add server id to zookeeper events. {pull}25550[25550]
 - Add additional network metrics to docker/network {pull}25354[25354]
 

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,93 @@
+// 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 (
+	"fmt"
+	"sync"
+	"time"
+
+	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) ([]*dto.MetricFamily, error)
+}
+
+type familiesCache struct {
+	sharedFamilies     []*dto.MetricFamily
+	lastFetchErr       error
+	lastFetchTimestamp time.Time
+	lock               sync.Mutex
+}
+
+type cacheMap map[string]*familiesCache
+
+type module struct {
+	mb.BaseModule
+
+	fCache cacheMap
+}
+
+func ModuleBuilder() func(base mb.BaseModule) (mb.Module, error) {
+	sharedFamiliesCache := make(cacheMap)
+	return func(base mb.BaseModule) (mb.Module, error) {
+		hash := generateCacheHash(base.Config().Hosts)
+		// NOTE: These entries will be never removed, this can be a leak if
+		// metricbeat is used to monitor clusters dynamically created.
+		// (https://github.com/elastic/beats/pull/25640#discussion_r633395213)
+		sharedFamiliesCache[hash] = &familiesCache{}
+		m := module{
+			BaseModule: base,
+			fCache:     sharedFamiliesCache,
+		}
+		return &m, nil
+	}
+}
+
+func (m *module) GetSharedFamilies(prometheus p.Prometheus) ([]*dto.MetricFamily, error) {
+	now := time.Now()
+
+	hash := generateCacheHash(m.Config().Hosts)
+	fCache := m.fCache[hash]
+
+	fCache.lock.Lock()
+	defer fCache.lock.Unlock()
+
+	if fCache.lastFetchTimestamp.IsZero() || now.Sub(fCache.lastFetchTimestamp) > m.Config().Period {
+		fCache.sharedFamilies, fCache.lastFetchErr = prometheus.GetFamilies()
+		fCache.lastFetchTimestamp = now
+	}
+
+	return fCache.sharedFamilies, fCache.lastFetchErr
+}
+
+func generateCacheHash(host []string) string {
+	return fmt.Sprintf("%s", host)
+}

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,6 +91,7 @@ type MetricSet struct {
 	mb.BaseMetricSet
 	prometheus p.Prometheus
 	enricher   util.Enricher
+	mod        k8smod.Module
 }
 
 // New create a new instance of the MetricSet
@@ -99,10 +102,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.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)
+	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_cronjob/state_cronjob.go

@@ -18,11 +18,14 @@
 package state_cronjob
 
 import (
+	"fmt"
+
 	"github.com/pkg/errors"
 
 	"github.com/elastic/beats/v7/libbeat/common"
 	p "github.com/elastic/beats/v7/metricbeat/helper/prometheus"
 	"github.com/elastic/beats/v7/metricbeat/mb"
+	k8smod "github.com/elastic/beats/v7/metricbeat/module/kubernetes"
 )
 
 func init() {
@@ -40,6 +43,7 @@ type CronJobMetricSet struct {
 	mb.BaseMetricSet
 	prometheus p.Prometheus
 	mapping    *p.MetricsMapping
+	mod        k8smod.Module
 }
 
 // NewCronJobMetricSet returns a prometheus based metricset for CronJobs
@@ -49,9 +53,15 @@ func NewCronJobMetricSet(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		return nil, err
 	}
 
+	mod, ok := base.Module().(k8smod.Module)
+	if !ok {
+		return nil, fmt.Errorf("must be child of kubernetes module")
+	}
+
 	return &CronJobMetricSet{
 		BaseMetricSet: base,
 		prometheus:    prometheus,
+		mod:           mod,
 		mapping: &p.MetricsMapping{
 			Metrics: map[string]p.MetricMap{
 				"kube_cronjob_info":                           p.InfoMetric(),
@@ -77,7 +87,11 @@ func NewCronJobMetricSet(base mb.BaseMetricSet) (mb.MetricSet, error) {
 //
 // Copied from other kube state metrics.
 func (m *CronJobMetricSet) Fetch(reporter mb.ReporterV2) error {
-	events, err := m.prometheus.GetProcessedMetrics(m.mapping)
+	families, err := m.mod.GetSharedFamilies(m.prometheus)
+	if err != nil {
+		return errors.Wrap(err, "error getting family metrics")
+	}
+	events, err := m.prometheus.ProcessMetrics(families, m.mapping)
 	if err != nil {
 		return errors.Wrap(err, "error getting metrics")
 	}

metricbeat/module/kubernetes/state_daemonset/state_daemonset.go

@@ -18,11 +18,14 @@
 package state_daemonset
 
 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"
 )
 
@@ -69,6 +72,7 @@ type MetricSet struct {
 	mb.BaseMetricSet
 	prometheus p.Prometheus
 	enricher   util.Enricher
+	mod        k8smod.Module
 }
 
 // New create a new instance of the MetricSet
@@ -79,10 +83,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.ReplicaSet{}, false),
+		mod:           mod,
 	}, nil
 }
 
@@ -92,7 +101,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)
+	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)

metricbeat/module/kubernetes/state_deployment/state_deployment.go

@@ -18,11 +18,14 @@
 package state_deployment
 
 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"
 )
 
@@ -70,6 +73,7 @@ type MetricSet struct {
 	mb.BaseMetricSet
 	prometheus p.Prometheus
 	enricher   util.Enricher
+	mod        k8smod.Module
 }
 
 // New create a new instance of the MetricSet
@@ -80,10 +84,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.Deployment{}, false),
+		mod:           mod,
 	}, nil
 }
 
@@ -93,7 +102,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)
+	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)