[Heartbeat] Refactor config system

CHANGELOG.next.asciidoc

@@ -105,6 +105,8 @@ https://github.com/elastic/beats/compare/v7.0.0-alpha2...master[Check the HEAD d
 - Add `nodes` to filebeat-kubernetes.yaml ClusterRole. {issue}24051[24051] {pull}24052[24052]
 
 *Heartbeat*
+- Adds negative body match. {pull}20728[20728]
+- Refactor synthetics configuration to new syntax. {pull}23467[23467]
 
 *Journalbeat*
 

heartbeat/beater/heartbeat.go

@@ -18,7 +18,6 @@
 package beater
 
 import (
-	"context"
 	"fmt"
 	"time"
 
@@ -83,10 +82,11 @@ func New(b *beat.Beat, rawConfig *common.Config) (beat.Beater, error) {
 func (bt *Heartbeat) Run(b *beat.Beat) error {
 	logp.Info("heartbeat is running! Hit CTRL-C to stop it.")
 
-	err := bt.RunStaticMonitors(b)
+	stopStaticMonitors, err := bt.RunStaticMonitors(b)
 	if err != nil {
 		return err
 	}
+	defer stopStaticMonitors()
 
 	if b.Manager.Enabled() {
 		bt.RunCentralMgmtMonitors(b)
@@ -102,13 +102,6 @@ func (bt *Heartbeat) Run(b *beat.Beat) error {
 		}
 	}
 
-	if len(bt.config.SyntheticSuites) > 0 {
-		err := bt.RunSyntheticSuiteMonitors(b)
-		if err != nil {
-			return err
-		}
-	}
-
 	if bt.config.Autodiscover != nil {
 		bt.autodiscover, err = bt.makeAutodiscover(b)
 		if err != nil {
@@ -131,22 +124,30 @@ func (bt *Heartbeat) Run(b *beat.Beat) error {
 }
 
 // RunStaticMonitors runs the `heartbeat.monitors` portion of the yaml config if present.
-func (bt *Heartbeat) RunStaticMonitors(b *beat.Beat) error {
+func (bt *Heartbeat) RunStaticMonitors(b *beat.Beat) (stop func(), err error) {
 	factory := monitors.NewFactory(b.Info, bt.scheduler, true)
 
+	var runners []cfgfile.Runner
 	for _, cfg := range bt.config.Monitors {
 		created, err := factory.Create(b.Publisher, cfg)
 		if err != nil {
 			if err == stdfields.ErrPluginDisabled {
 				continue // don't stop loading monitors just because they're disabled
 			}
 
-			return errors.Wrap(err, "could not create monitor")
+			return nil, errors.Wrap(err, "could not create monitor")
 		}
 
 		created.Start()
+		runners = append(runners, created)
 	}
-	return nil
+
+	stop = func() {
+		for _, runner := range runners {
+			runner.Stop()
+		}
+	}
+	return stop, nil
 }
 
 // RunCentralMgmtMonitors loads any central management configured configs.
@@ -170,50 +171,6 @@ func (bt *Heartbeat) RunReloadableMonitors(b *beat.Beat) (err error) {
 	return nil
 }
 
-// Provide hook to define journey list discovery from x-pack
-type JourneyLister func(ctx context.Context, suiteFile string, params common.MapStr) ([]string, error)
-
-var mainJourneyLister JourneyLister
-
-func RegisterJourneyLister(jl JourneyLister) {
-	mainJourneyLister = jl
-}
-
-func (bt *Heartbeat) RunSyntheticSuiteMonitors(b *beat.Beat) error {
-	// If we are running without XPack this will be nil
-	if mainJourneyLister == nil {
-		return nil
-	}
-	for _, suite := range bt.config.SyntheticSuites {
-		logp.Info("Listing suite %s", suite.Path)
-		journeyNames, err := mainJourneyLister(context.TODO(), suite.Path, suite.Params)
-		if err != nil {
-			return err
-		}
-		factory := monitors.NewFactory(b.Info, bt.scheduler, false)
-		for _, name := range journeyNames {
-			cfg, err := common.NewConfigFrom(map[string]interface{}{
-				"type":         "browser",
-				"path":         suite.Path,
-				"schedule":     suite.Schedule,
-				"params":       suite.Params,
-				"journey_name": name,
-				"name":         name,
-				"id":           name,
-			})
-			if err != nil {
-				return err
-			}
-			created, err := factory.Create(b.Publisher, cfg)
-			if err != nil {
-				return errors.Wrap(err, "could not create monitor")
-			}
-			created.Start()
-		}
-	}
-	return nil
-}
-
 // makeAutodiscover creates an autodiscover object ready to be started.
 func (bt *Heartbeat) makeAutodiscover(b *beat.Beat) (*autodiscover.Autodiscover, error) {
 	autodiscover, err := autodiscover.NewAutodiscover(

heartbeat/config/config.go

@@ -32,7 +32,7 @@ type Config struct {
 	ConfigMonitors  *common.Config       `config:"config.monitors"`
 	Scheduler       Scheduler            `config:"scheduler"`
 	Autodiscover    *autodiscover.Config `config:"autodiscover"`
-	SyntheticSuites []*SyntheticSuite    `config:"synthetic_suites"`
+	SyntheticSuites []*common.Config     `config:"synthetic_suites"`
 }
 
 // Scheduler defines the syntax of a heartbeat.yml scheduler block.
@@ -41,12 +41,5 @@ type Scheduler struct {
 	Location string `config:"location"`
 }
 
-type SyntheticSuite struct {
-	Path     string                 `config:"path"`
-	Name     string                 `config:"id_prefix"`
-	Schedule string                 `config:"schedule"`
-	Params   map[string]interface{} `config:"params"`
-}
-
 // DefaultConfig is the canonical instantiation of Config.
 var DefaultConfig = Config{}

heartbeat/monitors/active/http/http.go

@@ -22,7 +22,8 @@ import (
 	"net/http"
 	"net/url"
 
-	"github.com/elastic/beats/v7/heartbeat/monitors"
+	"github.com/elastic/beats/v7/heartbeat/monitors/plugin"
+
 	"github.com/elastic/beats/v7/heartbeat/monitors/jobs"
 	"github.com/elastic/beats/v7/heartbeat/monitors/wrappers"
 	"github.com/elastic/beats/v7/libbeat/common"
@@ -32,8 +33,7 @@ import (
 )
 
 func init() {
-	monitors.RegisterActive("http", create)
-	monitors.RegisterActive("synthetics/http", create)
+	plugin.Register("http", create, "synthetics/http")
 }
 
 var debugf = logp.MakeDebug("http")
@@ -42,15 +42,15 @@ var debugf = logp.MakeDebug("http")
 func create(
 	name string,
 	cfg *common.Config,
-) (js []jobs.Job, endpoints int, err error) {
+) (p plugin.Plugin, err error) {
 	config := defaultConfig
 	if err := cfg.Unpack(&config); err != nil {
-		return nil, 0, err
+		return plugin.Plugin{}, err
 	}
 
 	tls, err := tlscommon.LoadTLSConfig(config.TLS)
 	if err != nil {
-		return nil, 0, err
+		return plugin.Plugin{}, err
 	}
 
 	var body []byte
@@ -61,21 +61,21 @@ func create(
 		compression := config.Check.Request.Compression
 		enc, err = getContentEncoder(compression.Type, compression.Level)
 		if err != nil {
-			return nil, 0, err
+			return plugin.Plugin{}, err
 		}
 
 		buf := bytes.NewBuffer(nil)
 		err = enc.Encode(buf, bytes.NewBufferString(config.Check.Request.SendBody))
 		if err != nil {
-			return nil, 0, err
+			return plugin.Plugin{}, err
 		}
 
 		body = buf.Bytes()
 	}
 
 	validator, err := makeValidateResponse(&config.Check.Response)
 	if err != nil {
-		return nil, 0, err
+		return plugin.Plugin{}, err
 	}
 
 	// Determine whether we're using a proxy or not and then use that to figure out how to
@@ -87,7 +87,7 @@ func create(
 	if config.ProxyURL != "" || config.MaxRedirects > 0 {
 		transport, err := newRoundTripper(&config, tls)
 		if err != nil {
-			return nil, 0, err
+			return plugin.Plugin{}, err
 		}
 
 		makeJob = func(urlStr string) (jobs.Job, error) {
@@ -99,24 +99,24 @@ func create(
 		}
 	}
 
-	js = make([]jobs.Job, len(config.Hosts))
+	js := make([]jobs.Job, len(config.Hosts))
 	for i, urlStr := range config.Hosts {
 		u, _ := url.Parse(urlStr)
 		if err != nil {
-			return nil, 0, err
+			return plugin.Plugin{}, err
 		}
 
 		job, err := makeJob(urlStr)
 		if err != nil {
-			return nil, 0, err
+			return plugin.Plugin{}, err
 		}
 
 		// Assign any execution errors to the error field and
 		// assign the url field
 		js[i] = wrappers.WithURLField(u, job)
 	}
 
-	return js, len(config.Hosts), nil
+	return plugin.Plugin{Jobs: js, Close: nil, Endpoints: len(config.Hosts)}, nil
 }
 
 func newRoundTripper(config *Config, tls *tlscommon.TLSConfig) (*http.Transport, error) {

heartbeat/monitors/active/http/http_test.go

@@ -78,17 +78,17 @@ func sendTLSRequest(t *testing.T, testURL string, useUrls bool, extraConfig map[
 	config, err := common.NewConfigFrom(configSrc)
 	require.NoError(t, err)
 
-	jobs, endpoints, err := create("tls", config)
+	p, err := create("tls", config)
 	require.NoError(t, err)
 
 	sched := schedule.MustParse("@every 1s")
-	job := wrappers.WrapCommon(jobs, stdfields.StdMonitorFields{ID: "tls", Type: "http", Schedule: sched, Timeout: 1})[0]
+	job := wrappers.WrapCommon(p.Jobs, stdfields.StdMonitorFields{ID: "tls", Type: "http", Schedule: sched, Timeout: 1})[0]
 
 	event := &beat.Event{}
 	_, err = job(event)
 	require.NoError(t, err)
 
-	require.Equal(t, 1, endpoints)
+	require.Equal(t, 1, p.Endpoints)
 
 	return event
 }
@@ -318,11 +318,11 @@ func TestLargeResponse(t *testing.T) {
 	config, err := common.NewConfigFrom(configSrc)
 	require.NoError(t, err)
 
-	jobs, _, err := create("largeresp", config)
+	p, err := create("largeresp", config)
 	require.NoError(t, err)
 
 	sched, _ := schedule.Parse("@every 1s")
-	job := wrappers.WrapCommon(jobs, stdfields.StdMonitorFields{ID: "test", Type: "http", Schedule: sched, Timeout: 1})[0]
+	job := wrappers.WrapCommon(p.Jobs, stdfields.StdMonitorFields{ID: "test", Type: "http", Schedule: sched, Timeout: 1})[0]
 
 	event := &beat.Event{}
 	_, err = job(event)
@@ -532,11 +532,11 @@ func TestRedirect(t *testing.T) {
 	config, err := common.NewConfigFrom(configSrc)
 	require.NoError(t, err)
 
-	jobs, _, err := create("redirect", config)
+	p, err := create("redirect", config)
 	require.NoError(t, err)
 
 	sched, _ := schedule.Parse("@every 1s")
-	job := wrappers.WrapCommon(jobs, stdfields.StdMonitorFields{ID: "test", Type: "http", Schedule: sched, Timeout: 1})[0]
+	job := wrappers.WrapCommon(p.Jobs, stdfields.StdMonitorFields{ID: "test", Type: "http", Schedule: sched, Timeout: 1})[0]
 
 	// Run this test multiple times since in the past we had an issue where the redirects
 	// list was added onto by each request. See https://github.com/elastic/beats/pull/15944
@@ -579,11 +579,11 @@ func TestNoHeaders(t *testing.T) {
 	config, err := common.NewConfigFrom(configSrc)
 	require.NoError(t, err)
 
-	jobs, _, err := create("http", config)
+	p, err := create("http", config)
 	require.NoError(t, err)
 
 	sched, _ := schedule.Parse("@every 1s")
-	job := wrappers.WrapCommon(jobs, stdfields.StdMonitorFields{ID: "test", Type: "http", Schedule: sched, Timeout: 1})[0]
+	job := wrappers.WrapCommon(p.Jobs, stdfields.StdMonitorFields{ID: "test", Type: "http", Schedule: sched, Timeout: 1})[0]
 
 	event := &beat.Event{}
 	_, err = job(event)

heartbeat/monitors/active/icmp/icmp.go

@@ -22,6 +22,8 @@ import (
 	"net"
 	"net/url"
 
+	"github.com/elastic/beats/v7/heartbeat/monitors/plugin"
+
 	"github.com/elastic/beats/v7/heartbeat/eventext"
 	"github.com/elastic/beats/v7/heartbeat/look"
 	"github.com/elastic/beats/v7/heartbeat/monitors"
@@ -35,30 +37,29 @@ import (
 var debugf = logp.MakeDebug("icmp")
 
 func init() {
-	monitors.RegisterActive("icmp", create)
-	monitors.RegisterActive("synthetics/icmp", create)
+	plugin.Register("icmp", create, "synthetics/icmp")
 }
 
 func create(
 	name string,
 	commonConfig *common.Config,
-) (jobs []jobs.Job, endpoints int, err error) {
+) (p plugin.Plugin, err error) {
 	loop, err := getStdLoop()
 	if err != nil {
 		logp.Warn("Failed to initialize ICMP loop %v", err)
-		return nil, 0, err
+		return plugin.Plugin{}, err
 	}
 
 	config := DefaultConfig
 	if err := commonConfig.Unpack(&config); err != nil {
-		return nil, 0, err
+		return plugin.Plugin{}, err
 	}
 
 	jf, err := newJobFactory(config, monitors.NewStdResolver(), loop)
 	if err != nil {
-		return nil, 0, err
+		return plugin.Plugin{}, err
 	}
-	return jf.makeJobs()
+	return jf.makePlugin()
 
 }
 
@@ -89,29 +90,30 @@ func (jf *jobFactory) checkConfig() error {
 	return nil
 }
 
-func (jf *jobFactory) makeJobs() (j []jobs.Job, endpoints int, err error) {
+func (jf *jobFactory) makePlugin() (plugin2 plugin.Plugin, err error) {
 	if err := jf.loop.checkNetworkMode(jf.ipVersion); err != nil {
-		return nil, 0, err
+		return plugin.Plugin{}, err
 	}
 
 	pingFactory := jf.pingIPFactory(&jf.config)
 
+	var j []jobs.Job
 	for _, host := range jf.config.Hosts {
 		job, err := monitors.MakeByHostJob(host, jf.config.Mode, monitors.NewStdResolver(), pingFactory)
 
 		if err != nil {
-			return nil, 0, err
+			return plugin.Plugin{}, err
 		}
 
 		u, err := url.Parse(fmt.Sprintf("icmp://%s", host))
 		if err != nil {
-			return nil, 0, err
+			return plugin.Plugin{}, err
 		}
 
 		j = append(j, wrappers.WithURLField(u, job))
 	}
 
-	return j, len(jf.config.Hosts), nil
+	return plugin.Plugin{Jobs: j, Close: nil, Endpoints: len(jf.config.Hosts)}, nil
 }
 
 func (jf *jobFactory) pingIPFactory(config *Config) func(*net.IPAddr) jobs.Job {