[8.15](backport #40400) [metricbeat] - Allow metricsets to report their status via v2 protocol

CHANGELOG.next.asciidoc

@@ -33,6 +33,10 @@ https:/elastic/beats/compare/v8.8.1\...main[Check the HEAD diff]
 *Metricbeat*
 
 - Setting period for counter cache for Prometheus remote_write at least to 60sec {pull}38553[38553]
+- Add support of Graphite series 1.1.0+ tagging extension for statsd module. {pull}39619[39619]
+- Allow metricsets to report their status via control v2 protocol. {pull}40025[40025]
+- Remove fallback to the node limit for the `kubernetes.pod.cpu.usage.limit.pct` and `kubernetes.pod.memory.usage.limit.pct` metrics calculation
+- Add support for Kibana status metricset in v8 format {pull}40275[40275]
 
 *Osquerybeat*
 

NOTICE.txt

@@ -13180,11 +13180,19 @@ Contents of probable licence file $GOMODCACHE/github.com/elastic/elastic-agent-l
 
 --------------------------------------------------------------------------------
 Dependency : github.com/elastic/elastic-agent-system-metrics
+<<<<<<< HEAD
 Version: v0.10.3
 Licence type (autodetected): Apache-2.0
 --------------------------------------------------------------------------------
 
 Contents of probable licence file $GOMODCACHE/github.com/elastic/[email protected]/LICENSE.txt:
+=======
+Version: v0.11.0
+Licence type (autodetected): Apache-2.0
+--------------------------------------------------------------------------------
+
+Contents of probable licence file $GOMODCACHE/github.com/elastic/[email protected]/LICENSE.txt:
+>>>>>>> 2060383cc2 ([metricbeat] - Allow metricsets to report their status via v2 protocol (#40400))
 
  Apache License
  Version 2.0, January 2004

go.mod

@@ -198,8 +198,13 @@ require (
  github.com/elastic/bayeux v1.0.5
  github.com/elastic/ebpfevents v0.6.0
  github.com/elastic/elastic-agent-autodiscover v0.8.1
+<<<<<<< HEAD
  github.com/elastic/elastic-agent-libs v0.9.13
  github.com/elastic/elastic-agent-system-metrics v0.10.3
+=======
+ github.com/elastic/elastic-agent-libs v0.9.15
+ github.com/elastic/elastic-agent-system-metrics v0.11.0
+>>>>>>> 2060383cc2 ([metricbeat] - Allow metricsets to report their status via v2 protocol (#40400))
  github.com/elastic/go-elasticsearch/v8 v8.14.0
  github.com/elastic/go-sfdc v0.0.0-20240621062639-bcc8456508ff
  github.com/elastic/mito v1.13.1

go.sum

@@ -554,12 +554,21 @@ github.com/elastic/ebpfevents v0.6.0 h1:BrL3m7JFK7U6h2jkbk3xAWWs//IZnugCHEDds5u2
 github.com/elastic/ebpfevents v0.6.0/go.mod h1:ESG9gw7N+n5yCCMgdg1IIJENKWSmX7+X0Fi9GUs9nvU=
 github.com/elastic/elastic-agent-autodiscover v0.8.1 h1:u6TWqh7wfevu6S4GUq4SIxYBRo4b/P5RZmx/rSvT10A=
 github.com/elastic/elastic-agent-autodiscover v0.8.1/go.mod h1:0gzGsaDCAqBfUZjuCqqWsSI60eaZ778A5tQZV72rPV0=
+<<<<<<< HEAD
 github.com/elastic/elastic-agent-client/v7 v7.13.0 h1:ENCfV5XIMmjWo9/0J7t//5N7xgm43Ktg0SyIomupRcA=
 github.com/elastic/elastic-agent-client/v7 v7.13.0/go.mod h1:h2yJHN8Q5rhfi9i6FfyPufh+StFN+UD9PYGv8blXKbE=
 github.com/elastic/elastic-agent-libs v0.9.13 h1:D1rh1s67zlkDWmixWQaNWzn+qy6DafIDPTQnLpBNBUA=
 github.com/elastic/elastic-agent-libs v0.9.13/go.mod h1:G9ljFvDE+muOOOQBf2eRituF0fE4suGkv25rfjTwY+c=
 github.com/elastic/elastic-agent-system-metrics v0.10.3 h1:8pWdj8DeY8PBG/BA0DJalRpJWruDoP5QrIP/YKug5dE=
 github.com/elastic/elastic-agent-system-metrics v0.10.3/go.mod h1:3JwPa3zZJjmBYN87xwdLcFpHrUkWpR863jiYdg39sSc=
+=======
+github.com/elastic/elastic-agent-client/v7 v7.15.0 h1:nDB7v8TBoNuD6IIzC3z7Q0y+7bMgXoT2DsHfolO2CHE=
+github.com/elastic/elastic-agent-client/v7 v7.15.0/go.mod h1:6h+f9QdIr3GO2ODC0Y8+aEXRwzbA5W4eV4dd/67z7nI=
+github.com/elastic/elastic-agent-libs v0.9.15 h1:WCLtuErafUxczT/rXJa4Vr6mxwC8dgtqMbEq+qWGD4M=
+github.com/elastic/elastic-agent-libs v0.9.15/go.mod h1:2VgYxHaeM+cCDBjiS2wbmTvzPGbnlXAtYrlcLefheS8=
+github.com/elastic/elastic-agent-system-metrics v0.11.0 h1:/bWrgTsHZWLUhdT7WPNuQDFkrSfm+A4qf6QDQnZo9d8=
+github.com/elastic/elastic-agent-system-metrics v0.11.0/go.mod h1:3QiMu9wTKJFvpCN+5klgGqasTMNKJbgY3xcoN1KQXJk=
+>>>>>>> 2060383cc2 ([metricbeat] - Allow metricsets to report their status via v2 protocol (#40400))
 github.com/elastic/elastic-transport-go/v8 v8.6.0 h1:Y2S/FBjx1LlCv5m6pWAF2kDJAHoSjSRSJCApolgfthA=
 github.com/elastic/elastic-transport-go/v8 v8.6.0/go.mod h1:YLHer5cj0csTzNFXoNQ8qhtGY1GTvSqPnKWKaqQE3Hk=
 github.com/elastic/fsevents v0.0.0-20181029231046-e1d381a4d270 h1:cWPqxlPtir4RoQVCpGSRXmLqjEHpJKbR60rxh1nQZY4=

metricbeat/helper/http_test.go

@@ -19,7 +19,7 @@ package helper
 
 import (
  "fmt"
- "io/ioutil"
+ "io"
  "net"
  "net/http"
  "net/http/httptest"
@@ -31,6 +31,7 @@ import (
  "github.com/stretchr/testify/assert"
  "github.com/stretchr/testify/require"
 
+ "github.com/elastic/beats/v7/libbeat/management/status"
  "github.com/elastic/beats/v7/metricbeat/helper/dialer"
  "github.com/elastic/beats/v7/metricbeat/mb"
  "github.com/elastic/beats/v7/metricbeat/mb/parse"
@@ -55,7 +56,7 @@ func TestGetAuthHeaderFromToken(t *testing.T) {
  for _, test := range tests {
  t.Run(test.Name, func(t *testing.T) {
  content := []byte(test.Content)
- tmpfile, err := ioutil.TempFile("", "token")
+ tmpfile, err := os.CreateTemp("", "token")
  if err != nil {
  t.Fatal(err)
  }
@@ -236,14 +237,14 @@ func TestOverUnixSocket(t *testing.T) {
  fmt.Fprintf(w, "ehlo!")
  })
 
- go http.Serve(l, mux)
+ go http.Serve(l, mux) //nolint:errcheck,gosec // Ignore the error, it's a test file
 
  return l
  }
 
  for title, c := range cases {
  t.Run(title, func(t *testing.T) {
- tmpDir, err := ioutil.TempDir("", "testsocket")
+ tmpDir, err := os.MkdirTemp("", "testsocket")
  require.NoError(t, err)
  defer os.RemoveAll(tmpDir)
 
@@ -262,7 +263,7 @@ func TestOverUnixSocket(t *testing.T) {
  r, err := h.FetchResponse()
  require.NoError(t, err)
  defer r.Body.Close()
- content, err := ioutil.ReadAll(r.Body)
+ content, err := io.ReadAll(r.Body)
  require.NoError(t, err)
  assert.Equal(t, []byte("ehlo!"), content)
  })
@@ -327,3 +328,5 @@ func (*dummyModule) Config() mb.ModuleConfig {
 func (*dummyModule) UnpackConfig(interface{}) error {
  return nil
 }
+func (dummyModule) UpdateStatus(_ status.Status, _ string) {}
+func (dummyModule) SetStatusReporter(_ status.StatusReporter) {}

metricbeat/mb/mb.go

@@ -27,6 +27,7 @@ import (
  "net/url"
  "time"
 
+ "github.com/elastic/beats/v7/libbeat/management/status"
  "github.com/elastic/beats/v7/metricbeat/helper/dialer"
  conf "github.com/elastic/elastic-agent-libs/config"
  "github.com/elastic/elastic-agent-libs/logp"
@@ -62,9 +63,11 @@ const (
 
 // Module is the common interface for all Module implementations.
 type Module interface {
- Name() string // Name returns the name of the Module.
- Config() ModuleConfig // Config returns the ModuleConfig used to create the Module.
- UnpackConfig(to interface{}) error // UnpackConfig unpacks the raw module config to the given object.
+ Name() string // Name returns the name of the Module.
+ Config() ModuleConfig // Config returns the ModuleConfig used to create the Module.
+ UnpackConfig(to interface{}) error // UnpackConfig unpacks the raw module config to the given object.
+ UpdateStatus(status status.Status, msg string) // UpdateStatus updates the status of the module. Reflected on elastic-agent.
+ SetStatusReporter(statusReporter status.StatusReporter) // SetStatusReporter updates the status reporter for the given module.
 }
 
 // BaseModule implements the Module interface.
@@ -73,9 +76,10 @@ type Module interface {
 // MetricSets, it can embed this type into another struct to satisfy the
 // Module interface requirements.
 type BaseModule struct {
- name string
- config ModuleConfig
- rawConfig *conf.C
+ name string
+ config ModuleConfig
+ rawConfig *conf.C
+ statusReporter status.StatusReporter
 }
 
 func (m *BaseModule) String() string {
@@ -95,6 +99,18 @@ func (m *BaseModule) UnpackConfig(to interface{}) error {
  return m.rawConfig.Unpack(to)
 }
 
+// UpdateStatus updates the status of the module. Reflected on elastic-agent.
+func (m *BaseModule) UpdateStatus(status status.Status, msg string) {
+ if m.statusReporter != nil {
+ m.statusReporter.UpdateStatus(status, msg)
+ }
+}
+
+// SetStatusReporter sets the status repoter of the module.
+func (m *BaseModule) SetStatusReporter(statusReporter status.StatusReporter) {
+ m.statusReporter = statusReporter
+}
+
 // WithConfig re-configures the module with the given raw configuration and returns a
 // copy of the module.
 // Intended to be called from module factories. Note that if metricsets are specified

metricbeat/mb/module/runner.go

@@ -25,6 +25,7 @@ import (
  "github.com/elastic/beats/v7/libbeat/beat"
  "github.com/elastic/beats/v7/libbeat/cfgfile"
  "github.com/elastic/beats/v7/libbeat/common/diagnostics"
+ "github.com/elastic/beats/v7/libbeat/management/status"
  "github.com/elastic/elastic-agent-libs/monitoring"
 )
 
@@ -123,3 +124,7 @@ func (mr *runner) Diagnostics() []diagnostics.DiagnosticSetup {
 func (mr *runner) String() string {
  return fmt.Sprintf("%s [metricsets=%d]", mr.mod.Name(), len(mr.mod.metricSets))
 }
+
+func (mr *runner) SetStatusReporter(reporter status.StatusReporter) {
+ mr.mod.SetStatusReporter(reporter)
+}

metricbeat/mb/module/runner_group.go

@@ -23,6 +23,7 @@ import (
 
  "github.com/elastic/beats/v7/libbeat/cfgfile"
  "github.com/elastic/beats/v7/libbeat/common/diagnostics"
+ "github.com/elastic/beats/v7/libbeat/management/status"
 )
 
 type runnerGroup struct {
@@ -40,6 +41,14 @@ func newRunnerGroup(runners []cfgfile.Runner) cfgfile.Runner {
  }
 }
 
+func (rg *runnerGroup) SetStatusReporter(reporter status.StatusReporter) {
+ for _, runner := range rg.runners {
+ if runnerWithStatus, ok := runner.(status.WithStatusReporter); ok {
+ runnerWithStatus.SetStatusReporter(reporter)
+ }
+ }
+}
+
 func (rg *runnerGroup) Start() {
  rg.startOnce.Do(func() {
  for _, runner := range rg.runners {

metricbeat/mb/module/wrapper.go

@@ -26,6 +26,7 @@ import (
 
  "github.com/elastic/beats/v7/libbeat/beat"
  "github.com/elastic/beats/v7/libbeat/common"
+ "github.com/elastic/beats/v7/libbeat/management/status"
  "github.com/elastic/beats/v7/metricbeat/mb"
  conf "github.com/elastic/elastic-agent-libs/config"
  "github.com/elastic/elastic-agent-libs/logp"
@@ -146,6 +147,7 @@ func (mw *Wrapper) Start(done <-chan struct{}) <-chan beat.Event {
  registry.Add(metricsPath, msw.Metrics(), monitoring.Full)
  monitoring.NewString(msw.Metrics(), "starttime").Set(common.Time(time.Now()).String())
 
+ msw.module.UpdateStatus(status.Starting, fmt.Sprintf("%s/%s is starting", msw.module.Name(), msw.Name()))
  msw.run(done, out)
  }(msw)
  }
@@ -253,14 +255,20 @@ func (msw *metricSetWrapper) fetch(ctx context.Context, reporter reporter) {
  err := fetcher.Fetch(reporter.V2())
  if err != nil {
  reporter.V2().Error(err)
+ msw.module.UpdateStatus(status.Degraded, fmt.Sprintf("Error fetching data for metricset %s.%s: %v", msw.module.Name(), msw.MetricSet.Name(), err))
  logp.Err("Error fetching data for metricset %s.%s: %s", msw.module.Name(), msw.Name(), err)
+ } else {
+ msw.module.UpdateStatus(status.Running, "")
  }
  case mb.ReportingMetricSetV2WithContext:
  reporter.StartFetchTimer()
  err := fetcher.Fetch(ctx, reporter.V2())
  if err != nil {
  reporter.V2().Error(err)
+ msw.module.UpdateStatus(status.Degraded, fmt.Sprintf("Error fetching data for metricset %s.%s: %v", msw.module.Name(), msw.MetricSet.Name(), err))
  logp.Err("Error fetching data for metricset %s.%s: %s", msw.module.Name(), msw.Name(), err)
+ } else {
+ msw.module.UpdateStatus(status.Running, "")
  }
  default:
  panic(fmt.Sprintf("unexpected fetcher type for %v", msw))

metricbeat/mb/testing/modules.go

@@ -53,13 +53,15 @@ that Metricbeat does it and with the same validations.
  }
  }
 */
+
 package testing
 
 import (
  "context"
  "testing"
  "time"
 
+ "github.com/elastic/beats/v7/libbeat/management/status"
  "github.com/elastic/go-concert/timed"
 
  "github.com/elastic/beats/v7/metricbeat/mb"
@@ -72,9 +74,11 @@ type TestModule struct {
  RawConfig *conf.C
 }
 
-func (m *TestModule) Name() string { return m.ModName }
-func (m *TestModule) Config() mb.ModuleConfig { return m.ModConfig }
-func (m *TestModule) UnpackConfig(to interface{}) error { return m.RawConfig.Unpack(to) }
+func (m *TestModule) Name() string { return m.ModName }
+func (m *TestModule) Config() mb.ModuleConfig { return m.ModConfig }
+func (m *TestModule) UnpackConfig(to interface{}) error { return m.RawConfig.Unpack(to) }
+func (m *TestModule) UpdateStatus(_ status.Status, _ string) {}
+func (m *TestModule) SetStatusReporter(_ status.StatusReporter) {}
 
 func NewTestModule(t testing.TB, config interface{}) *TestModule {
  c, err := conf.NewConfigFrom(config)

metricbeat/module/elasticsearch/node_stats/data_test.go

@@ -22,6 +22,7 @@ package node_stats
 import (
  "testing"
 
+ "github.com/elastic/beats/v7/libbeat/management/status"
  "github.com/elastic/beats/v7/metricbeat/mb"
  "github.com/elastic/beats/v7/metricbeat/module/elasticsearch"
 )
@@ -60,3 +61,6 @@ func (m mockModule) Config() mb.ModuleConfig {
 func (m mockModule) UnpackConfig(to interface{}) error {
  return nil
 }
+
+func (m mockModule) UpdateStatus(_ status.Status, _ string) {}
+func (m mockModule) SetStatusReporter(_ status.StatusReporter) {}

metricbeat/module/system/process/process.go

@@ -20,6 +20,7 @@
 package process
 
 import (
+ "errors"
  "fmt"
  "os"
  "runtime"
@@ -111,7 +112,8 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) error {
  // monitor either a single PID, or the configured set of processes.
  if m.setpid == 0 {
  procs, roots, err := m.stats.Get()
- if err != nil {
+ if err != nil && !errors.Is(err, process.NonFatalErr{}) {
+ // return only if the error is fatal in nature
  return fmt.Errorf("process stats: %w", err)
  }
 
@@ -121,9 +123,10 @@ func (m *MetricSet) Fetch(r mb.ReporterV2) error {
  RootFields: roots[evtI],
  })
  if !isOpen {
- return nil
+ return err
  }
  }
+ return err
  } else {
  proc, root, err := m.stats.GetOneRootEvent(m.setpid)
  if err != nil {

metricbeat/module/system/process/process_test.go

@@ -37,13 +37,17 @@ func TestFetch(t *testing.T) {
 
  f := mbtest.NewReportingMetricSetV2Error(t, getConfig())
  events, errs := mbtest.ReportingFetchV2Error(f)
- assert.Empty(t, errs)
+ for _, err := range errs {
+ assert.ErrorIsf(t, err, process.NonFatalErr{}, "Expected non-fatal error, got %v", err)
+ }
  assert.NotEmpty(t, events)
 
  time.Sleep(2 * time.Second)
 
  events, errs = mbtest.ReportingFetchV2Error(f)
- assert.Empty(t, errs)
+ for _, err := range errs {
+ assert.ErrorIsf(t, err, process.NonFatalErr{}, "Expected non-fatal error, got %v", err)
+ }
  assert.NotEmpty(t, events)
 
  t.Logf("fetched %d events, showing events[0]:", len(events))