[8.2](backport #31785) Feature/cache expiration

CHANGELOG.next.asciidoc

@@ -48,6 +48,9 @@ https://github.com/elastic/beats/compare/v8.2.0\...main[Check the HEAD diff]
 
 *Metricbeat*
 
+- make `system/filesystem` code sensitive to `hostfs` and migrate libraries to `elastic-agent-opts` {pull}31001[31001]
+- Fix kubernetes module's internal cache expiration issue. This avoid metrics like `kubernetes.container.cpu.usage.limit.pct` from not being populated. {pull}31785[31785]
+- add missing HealthyHostCount and UnHealthyHostCount for application ELB. {pull}31853[31853]
 
 *Packetbeat*
 

libbeat/common/cache.go

@@ -255,7 +255,9 @@ func (c *Cache) StartJanitor(interval time.Duration) {
 
 // StopJanitor stops the goroutine created by StartJanitor.
 func (c *Cache) StopJanitor() {
-	close(c.janitorQuit)
+	if c.janitorQuit != nil {
+		close(c.janitorQuit)
+	}
 }
 
 // get returns the non-expired values from the cache.

metricbeat/module/kubernetes/container/container.go

@@ -75,7 +75,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return &MetricSet{
 		BaseMetricSet: base,
 		http:          http,
-		enricher:      util.NewContainerMetadataEnricher(base, true),
+		enricher:      util.NewContainerMetadataEnricher(base, mod.GetPerfMetricsCache(), true),
 		mod:           mod,
 	}, nil
 }
@@ -93,7 +93,11 @@ func (m *MetricSet) Fetch(reporter mb.ReporterV2) {
 		return
 	}
 
+<<<<<<< HEAD
 	events, err := eventMapping(body, util.PerfMetrics)
+=======
+	events, err := eventMapping(body, m.mod.GetPerfMetricsCache(), m.Logger())
+>>>>>>> 4b625fc411 (Feature/cache expiration (#31785))
 	if err != nil {
 		m.Logger().Error(err)
 		reporter.Error(err)

metricbeat/module/kubernetes/container/container_test.go

@@ -24,6 +24,7 @@ import (
 	"io/ioutil"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -40,7 +41,7 @@ func TestEventMapping(t *testing.T) {
 	body, err := ioutil.ReadAll(f)
 	assert.NoError(t, err, "cannot read test file "+testFile)
 
-	cache := util.NewPerfMetricsCache()
+	cache := util.NewPerfMetricsCache(120 * time.Second)
 	cache.NodeCoresAllocatable.Set("gke-beats-default-pool-a5b33e2e-hdww", 2)
 	cache.NodeMemAllocatable.Set("gke-beats-default-pool-a5b33e2e-hdww", 146227200)
 	cache.ContainerMemLimit.Set(util.ContainerUID("default", "nginx-deployment-2303442956-pcqfc", "nginx"), 14622720)

metricbeat/module/kubernetes/kubernetes.go

@@ -28,6 +28,7 @@ import (
 	"github.com/elastic/beats/v7/metricbeat/helper"
 	p "github.com/elastic/beats/v7/metricbeat/helper/prometheus"
 	"github.com/elastic/beats/v7/metricbeat/mb"
+	"github.com/elastic/beats/v7/metricbeat/module/kubernetes/util"
 )
 
 func init() {
@@ -41,6 +42,7 @@ type Module interface {
 	mb.Module
 	GetStateMetricsFamilies(prometheus p.Prometheus) ([]*dto.MetricFamily, error)
 	GetKubeletStats(http *helper.HTTP) ([]byte, error)
+	GetPerfMetricsCache() *util.PerfMetricsCache
 }
 
 type familiesCache struct {
@@ -84,6 +86,7 @@ type module struct {
 
 	kubeStateMetricsCache *kubeStateMetricsCache
 	kubeletStatsCache     *kubeletStatsCache
+	perfMetrics           *util.PerfMetricsCache
 	cacheHash             uint64
 }
 
@@ -94,15 +97,25 @@ func ModuleBuilder() func(base mb.BaseModule) (mb.Module, error) {
 	kubeletStatsCache := &kubeletStatsCache{
 		cacheMap: make(map[uint64]*statsCache),
 	}
+	perfMetrics := util.NewPerfMetricsCache(0)
 	return func(base mb.BaseModule) (mb.Module, error) {
 		hash, err := generateCacheHash(base.Config().Hosts)
 		if err != nil {
 			return nil, errors.Wrap(err, "error generating cache hash for kubeStateMetricsCache")
 		}
+
+		// NOTE: `Period * 2` is an arbitrary value to make the cache NEVER to expire before the next scraping run
+		// if different metricsets have different periods, we will effectively set (timeout = max(Period) * 2)
+		minCacheExpirationTime := base.Config().Period * 2
+		if perfMetrics.GetTimeout() < minCacheExpirationTime {
+			perfMetrics.SetOrUpdateTimeout(minCacheExpirationTime)
+		}
+
 		m := module{
 			BaseModule:            base,
 			kubeStateMetricsCache: kubeStateMetricsCache,
 			kubeletStatsCache:     kubeletStatsCache,
+			perfMetrics:           perfMetrics,
 			cacheHash:             hash,
 		}
 		return &m, nil
@@ -153,3 +166,7 @@ func generateCacheHash(host []string) (uint64, error) {
 	}
 	return id, nil
 }
+
+func (m *module) GetPerfMetricsCache() *util.PerfMetricsCache {
+	return m.perfMetrics
+}

metricbeat/module/kubernetes/node/node.go

@@ -79,7 +79,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return &MetricSet{
 		BaseMetricSet: base,
 		http:          http,
-		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.Node{}, false),
+		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.Node{}, mod.GetPerfMetricsCache(), false),
 		mod:           mod,
 	}, nil
 }

metricbeat/module/kubernetes/pod/pod.go

@@ -76,7 +76,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return &MetricSet{
 		BaseMetricSet: base,
 		http:          http,
-		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.Pod{}, true),
+		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.Pod{}, mod.GetPerfMetricsCache(), true),
 		mod:           mod,
 	}, nil
 }
@@ -94,7 +94,11 @@ func (m *MetricSet) Fetch(reporter mb.ReporterV2) {
 		return
 	}
 
+<<<<<<< HEAD
 	events, err := eventMapping(body, util.PerfMetrics)
+=======
+	events, err := eventMapping(body, m.mod.GetPerfMetricsCache(), m.Logger())
+>>>>>>> 4b625fc411 (Feature/cache expiration (#31785))
 	if err != nil {
 		m.Logger().Error(err)
 		reporter.Error(err)

metricbeat/module/kubernetes/pod/pod_test.go

@@ -24,6 +24,7 @@ import (
 	"io/ioutil"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -40,7 +41,7 @@ func TestEventMapping(t *testing.T) {
 	body, err := ioutil.ReadAll(f)
 	assert.NoError(t, err, "cannot read test file "+testFile)
 
-	cache := util.NewPerfMetricsCache()
+	cache := util.NewPerfMetricsCache(120 * time.Second)
 	cache.NodeCoresAllocatable.Set("gke-beats-default-pool-a5b33e2e-hdww", 2)
 	cache.NodeMemAllocatable.Set("gke-beats-default-pool-a5b33e2e-hdww", 146227200)
 	cache.ContainerMemLimit.Set(util.ContainerUID("default", "nginx-deployment-2303442956-pcqfc", "nginx"), 14622720)

metricbeat/module/kubernetes/state_container/state_container.go

@@ -122,7 +122,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return &MetricSet{
 		BaseMetricSet: base,
 		prometheus:    prometheus,
-		enricher:      util.NewContainerMetadataEnricher(base, false),
+		enricher:      util.NewContainerMetadataEnricher(base, mod.GetPerfMetricsCache(), false),
 		mod:           mod,
 	}, nil
 }

metricbeat/module/kubernetes/state_cronjob/state_cronjob.go

@@ -63,7 +63,7 @@ func NewCronJobMetricSet(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		BaseMetricSet: base,
 		prometheus:    prometheus,
 		mod:           mod,
-		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.CronJob{}, false),
+		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.CronJob{}, mod.GetPerfMetricsCache(), false),
 		mapping: &p.MetricsMapping{
 			Metrics: map[string]p.MetricMap{
 				"kube_cronjob_info":                           p.InfoMetric(),
@@ -122,8 +122,6 @@ func (m *CronJobMetricSet) Fetch(reporter mb.ReporterV2) {
 			return
 		}
 	}
-
-	return
 }
 
 // Close stops this metricset

metricbeat/module/kubernetes/state_daemonset/state_daemonset.go

@@ -89,7 +89,11 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return &MetricSet{
 		BaseMetricSet: base,
 		prometheus:    prometheus,
+<<<<<<< HEAD
 		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.ReplicaSet{}, false),
+=======
+		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.DaemonSet{}, mod.GetPerfMetricsCache(), false),
+>>>>>>> 4b625fc411 (Feature/cache expiration (#31785))
 		mod:           mod,
 	}, nil
 }

metricbeat/module/kubernetes/state_deployment/state_deployment.go

@@ -90,7 +90,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return &MetricSet{
 		BaseMetricSet: base,
 		prometheus:    prometheus,
-		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.Deployment{}, false),
+		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.Deployment{}, mod.GetPerfMetricsCache(), false),
 		mod:           mod,
 	}, nil
 }
@@ -128,8 +128,6 @@ func (m *MetricSet) Fetch(reporter mb.ReporterV2) {
 			return
 		}
 	}
-
-	return
 }
 
 // Close stops this metricset

metricbeat/module/kubernetes/state_job/state_job.go

@@ -108,7 +108,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return &MetricSet{
 		BaseMetricSet: base,
 		prometheus:    prometheus,
-		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.Job{}, false),
+		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.Job{}, mod.GetPerfMetricsCache(), false),
 		mod:           mod,
 	}, nil
 }
@@ -145,8 +145,6 @@ func (m *MetricSet) Fetch(reporter mb.ReporterV2) {
 			return
 		}
 	}
-
-	return
 }
 
 // Close stops this metricset

metricbeat/module/kubernetes/state_node/state_node.go

@@ -114,7 +114,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return &MetricSet{
 		BaseMetricSet: base,
 		prometheus:    prometheus,
-		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.Node{}, false),
+		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.Node{}, mod.GetPerfMetricsCache(), false),
 		mod:           mod,
 	}, nil
 }
@@ -151,8 +151,6 @@ func (m *MetricSet) Fetch(reporter mb.ReporterV2) {
 			return
 		}
 	}
-
-	return
 }
 
 // Close stops this metricset

metricbeat/module/kubernetes/state_persistentvolume/state_persistentvolume.go

@@ -54,6 +54,10 @@ func NewPersistentVolumeMetricSet(base mb.BaseMetricSet) (mb.MetricSet, error) {
 		BaseMetricSet: base,
 		prometheus:    prometheus,
 		mod:           mod,
+<<<<<<< HEAD
+=======
+		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.PersistentVolume{}, mod.GetPerfMetricsCache(), false),
+>>>>>>> 4b625fc411 (Feature/cache expiration (#31785))
 		mapping: &p.MetricsMapping{
 			Metrics: map[string]p.MetricMap{
 				"kube_persistentvolume_capacity_bytes": p.Metric("capacity.bytes"),

metricbeat/module/kubernetes/state_persistentvolumeclaim/state_persistentvolumeclaim.go

@@ -54,6 +54,10 @@ func NewpersistentvolumeclaimMetricSet(base mb.BaseMetricSet) (mb.MetricSet, err
 		BaseMetricSet: base,
 		prometheus:    prometheus,
 		mod:           mod,
+<<<<<<< HEAD
+=======
+		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.PersistentVolumeClaim{}, mod.GetPerfMetricsCache(), false),
+>>>>>>> 4b625fc411 (Feature/cache expiration (#31785))
 		mapping: &p.MetricsMapping{
 			Metrics: map[string]p.MetricMap{
 

metricbeat/module/kubernetes/state_pod/state_pod.go

@@ -92,7 +92,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return &MetricSet{
 		BaseMetricSet: base,
 		prometheus:    prometheus,
-		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.Pod{}, false),
+		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.Pod{}, mod.GetPerfMetricsCache(), false),
 		mod:           mod,
 	}, nil
 }
@@ -130,8 +130,6 @@ func (m *MetricSet) Fetch(reporter mb.ReporterV2) {
 			return
 		}
 	}
-
-	return
 }
 
 // Close stops this metricset

metricbeat/module/kubernetes/state_replicaset/state_replicaset.go

@@ -90,7 +90,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return &MetricSet{
 		BaseMetricSet: base,
 		prometheus:    prometheus,
-		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.ReplicaSet{}, false),
+		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.ReplicaSet{}, mod.GetPerfMetricsCache(), false),
 		mod:           mod,
 	}, nil
 }
@@ -127,8 +127,6 @@ func (m *MetricSet) Fetch(reporter mb.ReporterV2) {
 			return
 		}
 	}
-
-	return
 }
 
 // Close stops this metricset

metricbeat/module/kubernetes/state_resourcequota/state_resourcequota.go

@@ -97,5 +97,4 @@ func (m *ResourceQuotaMetricSet) Fetch(reporter mb.ReporterV2) {
 			return
 		}
 	}
-	return
 }

metricbeat/module/kubernetes/state_service/state_service.go

@@ -86,7 +86,7 @@ func NewServiceMetricSet(base mb.BaseMetricSet) (mb.MetricSet, error) {
 				"hostname":         p.Label("ingress_hostname"),
 			},
 		},
-		enricher: util.NewResourceMetadataEnricher(base, &kubernetes.Service{}, false),
+		enricher: util.NewResourceMetadataEnricher(base, &kubernetes.Service{}, mod.GetPerfMetricsCache(), false),
 	}, nil
 }
 
@@ -122,7 +122,6 @@ func (m *ServiceMetricSet) Fetch(reporter mb.ReporterV2) {
 			return
 		}
 	}
-	return
 }
 
 // Close stops this metricset

metricbeat/module/kubernetes/state_statefulset/state_statefulset.go

@@ -90,7 +90,7 @@ func New(base mb.BaseMetricSet) (mb.MetricSet, error) {
 	return &MetricSet{
 		BaseMetricSet: base,
 		prometheus:    prometheus,
-		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.StatefulSet{}, false),
+		enricher:      util.NewResourceMetadataEnricher(base, &kubernetes.StatefulSet{}, mod.GetPerfMetricsCache(), false),
 		mod:           mod,
 	}, nil
 }
@@ -127,8 +127,6 @@ func (m *MetricSet) Fetch(reporter mb.ReporterV2) {
 			return
 		}
 	}
-
-	return
 }
 
 // Close stops this metricset

metricbeat/module/kubernetes/state_storageclass/state_storageclass.go

@@ -99,5 +99,4 @@ func (m *StorageClassMetricSet) Fetch(reporter mb.ReporterV2) {
 			return
 		}
 	}
-	return
 }