[Heartbeat] Fix bug where enabled: false is ignored. (#27615)

Fixes #27599

This properly handles skipped monitors, and also fixes the broken test
for this.

Additionally, this switches to the go1.3+ way of handling error
hierarchies in `heartbeat.go` using `errors.Is`. It also cleans up the code by no longer
making a disabled config an `error` when parsing stdfields. It's now
only an error when detected in `monitor.go` which is cleaner.

heartbeat/beater/heartbeat.go

@@ -18,15 +18,13 @@
 package beater
 
 import (
+	"errors"
 	"fmt"
 	"time"
 
-	"github.com/pkg/errors"
-
 	"github.com/elastic/beats/v7/heartbeat/config"
 	"github.com/elastic/beats/v7/heartbeat/hbregistry"
 	"github.com/elastic/beats/v7/heartbeat/monitors"
-	"github.com/elastic/beats/v7/heartbeat/monitors/stdfields"
 	"github.com/elastic/beats/v7/heartbeat/scheduler"
 	"github.com/elastic/beats/v7/libbeat/autodiscover"
 	"github.com/elastic/beats/v7/libbeat/beat"
@@ -132,11 +130,12 @@ func (bt *Heartbeat) RunStaticMonitors(b *beat.Beat) (stop func(), err error) {
 	for _, cfg := range bt.config.Monitors {
 		created, err := factory.Create(b.Publisher, cfg)
 		if err != nil {
-			if err == stdfields.ErrPluginDisabled {
+			if errors.Is(err, monitors.ErrMonitorDisabled) {
+				logp.Info("skipping disabled monitor: %s", err)
 				continue // don't stop loading monitors just because they're disabled
 			}
 
-			return nil, errors.Wrap(err, "could not create monitor")
+			return nil, fmt.Errorf("could not create monitor: %w", err)
 		}
 
 		created.Start()
@@ -163,7 +162,7 @@ func (bt *Heartbeat) RunCentralMgmtMonitors(b *beat.Beat) {
 func (bt *Heartbeat) RunReloadableMonitors(b *beat.Beat) (err error) {
 	// Check monitor configs
 	if err := bt.monitorReloader.Check(bt.dynamicFactory); err != nil {
-		logp.Error(errors.Wrap(err, "error loading reloadable monitors"))
+		logp.Error(fmt.Errorf("error loading reloadable monitors: %w", err))
 	}
 
 	// Execute the monitor

heartbeat/monitors/monitor.go

@@ -34,6 +34,9 @@ import (
 	"github.com/elastic/beats/v7/libbeat/logp"
 )
 
+// ErrMonitorDisabled is returned when the monitor plugin is marked as disabled.
+var ErrMonitorDisabled = errors.New("monitor not loaded, plugin is disabled")
+
 // Monitor represents a configured recurring monitoring configuredJob loaded from a config file. Starting it
 // will cause it to run with the given scheduler until Stop() is called.
 type Monitor struct {
@@ -114,6 +117,10 @@ func newMonitorUnsafe(
 		return nil, err
 	}
 
+	if !config.Enabled() {
+		return nil, fmt.Errorf("monitor '%s' with id '%s' skipped: %w", standardFields.Name, standardFields.ID, ErrMonitorDisabled)
+	}
+
 	pluginFactory, found := registrar.Get(standardFields.Type)
 	if !found {
 		return nil, fmt.Errorf("monitor type %v does not exist, valid types are %v", standardFields.Type, registrar.MonitorNames())

heartbeat/monitors/stdfields/stdfields.go

@@ -26,9 +26,6 @@ import (
 	"github.com/elastic/beats/v7/libbeat/common"
 )
 
-// ErrPluginDisabled is returned when the monitor plugin is marked as disabled.
-var ErrPluginDisabled = errors.New("monitor not loaded, plugin is disabled")
-
 type ServiceFields struct {
 	Name string `config:"name"`
 }
@@ -60,9 +57,5 @@ func ConfigToStdMonitorFields(config *common.Config) (StdMonitorFields, error) {
 		}
 	}
 
-	if !mpi.Enabled {
-		return mpi, ErrPluginDisabled
-	}
-
 	return mpi, nil
 }

heartbeat/tests/system/config/heartbeat.yml.j2

@@ -8,6 +8,10 @@ heartbeat.monitors:
   timeout: {{monitor.timeout}}
   {% endif -%}
 
+  {%- if monitor.enabled is defined %}
+  enabled: {{monitor.enabled}}
+  {% endif -%}
+
   {%- if monitor.tags is defined %}
   tags:
     {% for tag in monitor.tags -%}

heartbeat/tests/system/test_base.py

@@ -5,6 +5,7 @@
 from elasticsearch import Elasticsearch
 from beat.beat import INTEGRATION_TESTS
 from beat import common_tests
+from time import sleep
 
 
 class Test(BaseTest, common_tests.TestExportsMixin):
@@ -53,7 +54,7 @@ def test_disabled(self):
         )
 
         heartbeat_proc = self.start_beat()
-        self.wait_until(lambda: self.log_contains("heartbeat is running"))
+        self.wait_until(lambda: self.log_contains("skipping disabled monitor"))
         heartbeat_proc.check_kill_and_wait()
 
     def test_fields_under_root(self):