Add Fields and Tag to monitor config

CHANGELOG.asciidoc

@@ -479,6 +479,9 @@ https://github.com/elastic/beats/compare/v5.2.2...v5.3.0[View commits]
 - Improve handling of different path variants on Windows. {pull}3781[3781]
 - Add multiline.flush_pattern option, for specifying the 'end' of a multiline pattern {pull}4019[4019]
 
+*Heartbeat*
+
+- Add `tags`, `fields` and `fields_under_root` in monitors configuration. {pull}3623[3623]
 
 *Metricbeat*
 

heartbeat/_meta/beat.full.yml

@@ -46,6 +46,22 @@ heartbeat.monitors:
   # Waiting duration until another ICMP Echo Request is emitted.
   wait: 1s
 
+  # The tags of the monitors are included in their own field with each
+  # transaction published. Tags make it easy to group servers by different
+  # logical properties.
+  #tags: ["service-X", "web-tier"]
+
+  # Optional fields that you can specify to add additional information to the
+  # monitor output. Fields can be scalar values, arrays, dictionaries, or any nested
+  # combination of these.
+  #fields:
+  #  env: staging
+
+  # If this option is set to true, the custom fields are stored as top-level
+  # fields in the output document instead of being grouped under a fields
+  # sub-dictionary. Default is false.
+  #fields_under_root: false
+
 - type: tcp # monitor type `tcp`. Connect via TCP and optionally verify endpoint
             # by sending/receiving a custom payload
 
@@ -64,8 +80,8 @@ heartbeat.monitors:
   #       Requires ports configs to be checked. If ssl is configured,
   #       a SSL/TLS based connection will be established. Otherwise plain tcp connection
   #       will be established
-  #   - hostname + port like `localhost:12345`:  
-  #       Connect to port on given host. If ssl is configured, 
+  #   - hostname + port like `localhost:12345`:
+  #       Connect to port on given host. If ssl is configured,
   #       a SSL/TLS based connection will be established. Otherwise plain tcp connection
   #       will be established
   #   - full url syntax. `scheme://<host>:[port]`. The `<scheme>` can be one of

heartbeat/beater/manager.go

@@ -32,15 +32,22 @@ type Monitor struct {
 	factory    monitors.Factory
 	config     *common.Config
 
-	active map[string]MonitorTask
+	active map[string]monitorTask
 }
 
-type MonitorTask struct {
-	name, typ string
+type monitorTask struct {
+	job    monitors.Job
+	cancel JobCanceller
 
-	job      monitors.Job
-	schedule scheduler.Schedule
-	cancel   JobCanceller
+	config monitorTaskConfig
+}
+
+type monitorTaskConfig struct {
+	Name     string             `config:"name"`
+	Type     string             `config:"type"`
+	Schedule *schedule.Schedule `config:"schedule" validate:"required"`
+	// Fields and tags to add to monitor.
+	EventMetadata common.EventMetadata `config:",inline"`
 }
 
 type JobControl interface {
@@ -109,7 +116,7 @@ func newMonitorManager(
 			name:    info.Name,
 			factory: factory,
 			config:  config,
-			active:  map[string]MonitorTask{},
+			active:  map[string]monitorTask{},
 		})
 	}
 
@@ -146,19 +153,15 @@ func newMonitorManager(
 }
 
 func (m *Monitor) Update(configs []*common.Config) error {
-	all := map[string]MonitorTask{}
+	all := map[string]monitorTask{}
 	for i, upd := range configs {
 		config, err := common.MergeConfigs(m.config, upd)
 		if err != nil {
 			logp.Err("Failed merging monitor config with updates: %v", err)
 			return err
 		}
 
-		shared := struct {
-			Name     string             `config:"name"`
-			Type     string             `config:"type"`
-			Schedule *schedule.Schedule `config:"schedule" validate:"required"`
-		}{}
+		shared := monitorTaskConfig{}
 		if err := config.Unpack(&shared); err != nil {
 			logp.Err("Failed parsing job schedule: ", err)
 			return err
@@ -170,17 +173,10 @@ func (m *Monitor) Update(configs []*common.Config) error {
 			return err
 		}
 
-		name := shared.Name
-		if name == "" {
-			name = shared.Type
-		}
-
 		for _, job := range jobs {
-			all[job.Name()] = MonitorTask{
-				name:     name,
-				typ:      shared.Type,
-				job:      job,
-				schedule: shared.Schedule,
+			all[job.Name()] = monitorTask{
+				job:    job,
+				config: shared,
 			}
 		}
 	}
@@ -189,12 +185,12 @@ func (m *Monitor) Update(configs []*common.Config) error {
 	for _, job := range m.active {
 		job.cancel()
 	}
-	m.active = map[string]MonitorTask{}
+	m.active = map[string]monitorTask{}
 
 	// start new and reconfigured tasks
 	for id, t := range all {
-		job := createJob(m.manager.client, t.job, t.name, t.typ)
-		t.cancel = m.manager.jobControl.Add(t.schedule, id, job)
+		job := createJob(m.manager.client, t)
+		t.cancel = m.manager.jobControl.Add(t.config.Schedule, id, job)
 		m.active[id] = t
 	}
 
@@ -234,13 +230,17 @@ func createWatchUpdater(monitor *Monitor) func(content []byte) {
 	}
 }
 
-func createJob(client publisher.Client, r monitors.Job, name, typ string) scheduler.TaskFunc {
-	return createJobTask(client, r, name, typ)
+func createJob(client publisher.Client, task monitorTask) scheduler.TaskFunc {
+	return task.prepareSchedulerJob(client, task.job)
 }
 
-func createJobTask(client publisher.Client, r monitors.TaskRunner, name, typ string) scheduler.TaskFunc {
+func (m *monitorTask) prepareSchedulerJob(client publisher.Client, r monitors.TaskRunner) scheduler.TaskFunc {
+	name := m.config.Name
+	if name == "" {
+		name = m.config.Name
+	}
 	return func() []scheduler.TaskFunc {
-		event, next, err := r.Run()
+		event, next, err := m.job.Run()
 		if err != nil {
 			logp.Err("Job %v failed with: ", err)
 		}
@@ -249,13 +249,15 @@ func createJobTask(client publisher.Client, r monitors.TaskRunner, name, typ str
 			event.DeepUpdate(common.MapStr{
 				"monitor": common.MapStr{
 					"name": name,
-					"type": typ,
+					"type": m.config.Type,
 				},
 			})
 
 			if _, exists := event["type"]; !exists {
 				event["type"] = defaultEventType
 			}
+			common.MergeFields(event, m.config.EventMetadata.Fields, m.config.EventMetadata.FieldsUnderRoot)
+			common.AddTags(event, m.config.EventMetadata.Tags)
 			client.PublishEvent(event)
 		}
 
@@ -265,7 +267,7 @@ func createJobTask(client publisher.Client, r monitors.TaskRunner, name, typ str
 
 		cont := make([]scheduler.TaskFunc, len(next))
 		for i, n := range next {
-			cont[i] = createJobTask(client, n, name, typ)
+			cont[i] = m.prepareSchedulerJob(client, n)
 		}
 		return cont
 	}

heartbeat/heartbeat.full.yml

@@ -46,6 +46,22 @@ heartbeat.monitors:
   # Waiting duration until another ICMP Echo Request is emitted.
   wait: 1s
 
+  # The tags of the monitors are included in their own field with each
+  # transaction published. Tags make it easy to group servers by different
+  # logical properties.
+  #tags: ["service-X", "web-tier"]
+
+  # Optional fields that you can specify to add additional information to the
+  # monitor output. Fields can be scalar values, arrays, dictionaries, or any nested
+  # combination of these.
+  #fields:
+  #  env: staging
+
+  # If this option is set to true, the custom fields are stored as top-level
+  # fields in the output document instead of being grouped under a fields
+  # sub-dictionary. Default is false.
+  #fields_under_root: false
+
 - type: tcp # monitor type `tcp`. Connect via TCP and optionally verify endpoint
             # by sending/receiving a custom payload
 
@@ -64,8 +80,8 @@ heartbeat.monitors:
   #       Requires ports configs to be checked. If ssl is configured,
   #       a SSL/TLS based connection will be established. Otherwise plain tcp connection
   #       will be established
-  #   - hostname + port like `localhost:12345`:  
-  #       Connect to port on given host. If ssl is configured, 
+  #   - hostname + port like `localhost:12345`:
+  #       Connect to port on given host. If ssl is configured,
   #       a SSL/TLS based connection will be established. Otherwise plain tcp connection
   #       will be established
   #   - full url syntax. `scheme://<host>:[port]`. The `<scheme>` can be one of

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

@@ -1,7 +1,11 @@
 heartbeat.monitors:
-- type: icmp
-  hosts: ["localhost"]
-  schedule: '@every 10s'
+- type: http
+  urls: ["http://localhost:9200"]
+  schedule: '@every 1s'
+  tags: ["http_monitor_tags"]
+  fields_under_root: true
+  fields:
+    hello: world
 
 output.file:
   path: {{ output_file_path|default(beat.working_dir + "/output") }}

heartbeat/tests/system/test_base.py

@@ -16,3 +16,20 @@ def test_base(self):
         heartbeat_proc = self.start_beat()
         self.wait_until(lambda: self.log_contains("heartbeat is running"))
         heartbeat_proc.check_kill_and_wait()
+
+    def test_monitor_config(self):
+        """
+        Basic test with fields and tags in monitor
+        """
+        self.render_config_template(
+            path=os.path.abspath(self.working_dir) + "/*"
+        )
+
+        heartbeat_proc = self.start_beat()
+        self.wait_until(lambda: self.output_lines() > 0)
+        heartbeat_proc.check_kill_and_wait()
+
+        doc = self.read_output()[0]
+        assert doc["hello"] == "world"
+        assert doc["tags"] == ["http_monitor_tags"]
+        assert "fields" not in doc