Convert heartbeat to new publisher pipeline #4591
Conversation
- install EventMetadata processing with beat.Client on connect
- split in return types between Task and Job:
- a Task can optionally return event fields of type common.MapStr
- a Job returns/reports full events by wrapping the Tasks return into a
beat.Event type with additional annotations
- unexport some 'local' types
| @@ -66,11 +62,12 @@ func (bt *Heartbeat) Run(b *beat.Beat) error { | |||
|
|
|||
| <-bt.done | |||
|
|
|||
| bt.manager.Stop() | |||
There was a problem hiding this comment.
could also be done as a defer statement on line 46 I think? Is there a reason you use bt.manager instead of manager`
| t.cancel = m.manager.jobControl.Add(t.config.Schedule, id, job) | ||
| processors, err := processors.New(t.config.Processors) | ||
| if err != nil { | ||
| logp.Critical("Fail to load monitor processors: %v", err) |
There was a problem hiding this comment.
Not for this PR but thinking what we should do in this case? Alternatively we could return an error and move the creating of the jobs into a second for loop. Means all jobs only get created if all tasks work. Not sure what the user expects.
There was a problem hiding this comment.
This loop is about creating tasks, after merging the tasks settings (via json file) with the actual monitor settings. That is, this loop also drives monitors config reloading and is to be replaced is metricbeat/filebeat config reloading. In the filebeat/metricbeat case, the reloading is less flexible in handline settings, allowing us to load the configs once and not per task.
| func (f funcTask) Run() (common.MapStr, []TaskRunner, error) { return f.run() } | ||
|
|
||
| /* |
|
@urso I'm removing the needs_docs label here because you've already added the processors docs. If additional doc work is required, please open an issue against the docs. |
Requires: #4554