Move processors into new pipeline #4554

urso · 2017-06-24T23:27:32Z

Update all processors to operate in *beat.Event
Update processors factory to use *common.Config, not common.Config. common.Config works, but go-ucfg interfaces/API assume to operate on pointers (passing by value might break functionality in future?)
Update most processors interface to pass the processors type/context by pointer instead of by value (due to use of interfaces, the processors have been allocated on heap anyways).
Install list of processors in new publisher pipeline. This allows publisher pipeline to account for events being dropped by filters, when reporting ACKs. From beats point of view, there should be no difference between dropped events (by filters) and ACKed events (by outputs). Interface mandates all events being properly ACKed by the publisher pipeline.
Tags, Fields and beat meta data settings are now implicitely converted to Processors as well
-> change ensures, all metadata (local and global) being applied before the
actual processors (installed with the pipeline) are executed. Ensures all
processors will see the very same events.
introduce (*processors.Processors).RunBC(common.MapStr) common.MapStr. Will be removed, once the beats themselves configure publisher pipeline to run processors (filebeat/metricbeat are executing processors on their own).
add Private field to beat.Event for event-based ACK callbacks. Use Private field to store additional event-metadata required for post-processing in the ACK callback (as processor pipeline can change event at will)
Prepare moving beats to new pipeline, but introducing ConnectX to BC API, for beats to connect to shared global pipeline (Note: will be removed when BC layer is removed).
Processors are still executed by the go-routine calling (beat.Client).Publish.
New Processor Execution Order (C=client based processor, P=pipeline based processor)
1. (P) extract EventMetadataKey fields + tags (to be removed in favor of 4)
2. (P) generalize/normalize event
3. (P) add beats metadata (name, hostname, version)
4. (P) add pipeline fields + tags
5. (C) add client fields + tags
6. (P/C) apply EventMetadataKey fields + tags (to be removed in favor of 4)
7. (C) client processors list
8. (P) pipeline processors list
9. (P) (if publish/debug enabled) log event
10. (P) (if output disabled) dropEvent

For docs: event processing execution order would be nice to have.

ruflin · 2017-06-27T07:35:21Z

libbeat/processors/actions/drop_event.go

 	// return event=nil to delete the entire event
 	return nil, nil
 }

-func (f dropEvent) String() string { return "drop_event" }
+func (*dropEvent) String() string { return "drop_event" }


Didn't know this works, thought _ is required.

ruflin

LGTM. Few questions:

There are quite a few changes on the JSON part. I wasn't sure how this is related to the other changes?
If I understand it right, config wise nothing changes for fields / tags, it's just the internal handling as processor. Also these cannot be configured by a user a processor as they are not registered processors.

ruflin · 2017-07-01T18:12:41Z

libbeat/processors/actions/decode_json_fields_test.go

-	actual, err := p.Run(input)
-
-	return actual
+	actual, _ := p.Run(&beat.Event{Fields: input})


No error check?

the error has never been checked for. In this PR I'm trying to adjust interfaces, not fix/improve/change behavior in tests or other pieces.

ruflin · 2017-07-01T18:18:21Z

libbeat/processors/processor.go

 		if err != nil {
+			// XXX: We don't drop the event, but continue filtering here iff the most


Could we add an "error" to the event in case a processor did not work so it can be seen in the event, that something went wrong?

Same here. Old/Current logic does ignore errors and at most prints a debug message. I don't intend to change any processing logic.

ruflin · 2017-07-01T18:22:39Z

libbeat/publisher/pipeline/processor.go

+//
+// Pipeline (C=client, P=pipeline)
+//
+// 1. (P) add EventMetadataKey fields + tags (to be removed in favor of 4)


We probably should have that somewhere in the docs

Yeah. Adding tag.

urso · 2017-07-02T17:37:48Z

There are quite a few changes on the JSON part. I wasn't sure how this is related to the other changes?

There's some shared json parsing logic treating @timestamp and such a little different. In processors we're using beat.Event, but the reader in filebeat is returning common.MapStr. As filebeat is still using the old publisher API, I had to convert to temporary beat.Event type. There is no change in logic.

If I understand it right, config wise nothing changes for fields / tags, it's just the internal handling as processor. Also these cannot be configured by a user a processor as they are not registered processors.

Yep. The pipeline/processor.go defines some internal processors, not globally registered/configurable.

- Update all processors to operate in *beat.Event - Update processors factory to use `*common.Config`, not `common.Config`. `common.Config` works, but go-ucfg interfaces/API assume to operate on pointers (passing by value might break functionality in future?) - Update most processors interface to pass the processors type/context by pointer instead of by value (due to use of interfaces, the processors have been allocated on heap anyways). - Install list of processors in new publisher pipeline. This allows publisher pipeline to account for events being dropped by filters, when reporting ACKs. From beats point of view, there should be no difference between dropped events (by filters) and ACKed events (by outputs). Interface mandates all events being properly ACKed by the publisher pipeline. - Tags, Fields and beat meta data settings are now implicitely converted to Processors as well - -> change ensures, all metadata (local and global) being applied before the actual processors (installed with the pipeline) are executed. Ensures all processors will see the very same events. - introduce `(*processors.Processors).RunBC(common.MapStr) common.MapStr`. Will be removed, once the beats themselves configure publisher pipeline to run processors (filebeat/metricbeat are executing processors on their own). - add `Private` field to `beat.Event` for event-based ACK callbacks. Use `Private` field to store additional event-metadata required for post-processing in the ACK callback (as processor pipeline can change event at will) - Prepare moving beats to new pipeline, but introducing `ConnectX` to BC API, for beats to connect to shared global pipeline (Note: will be removed when BC layer is removed). - Processors are still executed by the go-routine calling `(beat.Client).Publish`. - New Processor Execution Order (C=client based processor, P=pipeline based processor) 1. (P) extract EventMetadataKey fields + tags (to be removed in favor of 4) 2. (P) generalize/normalize event 3. (P) add beats metadata (name, hostname, version) 4. (P) add pipeline fields + tags 5. (C) add client fields + tags 6. (P/C) apply EventMetadataKey fields + tags (to be removed in favor of 4) 7. (C) client processors list 8. (P) pipeline processors list 9. (P) (if publish/debug enabled) log event 10. (P) (if output disabled) dropEvent client processors higher prio

dedemorton · 2018-01-22T18:22:56Z

I'm removing the needs_docs label because any outstanding doc work required here is tracked by #4598

urso added the in progress Pull request is currently in progress. label Jun 24, 2017

urso force-pushed the enh/pipeline-processors branch from 789983d to fe75d24 Compare June 26, 2017 12:12

ruflin reviewed Jun 27, 2017

View reviewed changes

urso force-pushed the enh/pipeline-processors branch from fe75d24 to bbe6888 Compare June 30, 2017 12:33

urso added review and removed in progress Pull request is currently in progress. labels Jun 30, 2017

urso changed the title ~~[WIP] Move processors into new pipeline~~ Move processors into new pipeline Jun 30, 2017

urso added the refactoring label Jun 30, 2017

urso force-pushed the enh/pipeline-processors branch from 5a8b240 to 526cf2e Compare July 1, 2017 15:36

ruflin approved these changes Jul 1, 2017

View reviewed changes

urso added the needs_docs label Jul 2, 2017

urso force-pushed the enh/pipeline-processors branch from 526cf2e to 96eca70 Compare July 2, 2017 13:26

urso force-pushed the enh/pipeline-processors branch from 96eca70 to aab3cb1 Compare July 2, 2017 18:19

urso mentioned this pull request Jul 2, 2017

Convert heartbeat to new publisher pipeline #4591

Merged

urso force-pushed the enh/pipeline-processors branch from aab3cb1 to 0a0bb10 Compare July 2, 2017 22:46

urso force-pushed the enh/pipeline-processors branch from 0a0bb10 to 6f47a11 Compare July 3, 2017 05:56

ruflin merged commit 5ddd685 into elastic:master Jul 3, 2017

monicasarbu added the libbeat label Jul 3, 2017

urso mentioned this pull request Jul 3, 2017

Publisher Pipeline #4598

Closed

22 tasks

dedemorton mentioned this pull request Jul 25, 2017

Beats doc updates in 6.0 #4540

Closed

42 tasks

dedemorton mentioned this pull request Dec 14, 2017

Beats doc updates in 6.x #5632

Closed

37 tasks

dedemorton removed the needs_docs label Jan 22, 2018

ewgRa mentioned this pull request Mar 8, 2018

json.overwrite_keys causes filebeat to panic #6381

Closed

urso deleted the enh/pipeline-processors branch February 19, 2019 18:47

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Move processors into new pipeline #4554

Move processors into new pipeline #4554

urso commented Jun 24, 2017 •

edited

Loading

ruflin Jun 27, 2017

ruflin left a comment

ruflin Jul 1, 2017

urso Jul 2, 2017

ruflin Jul 1, 2017

urso Jul 2, 2017

ruflin Jul 1, 2017

urso Jul 2, 2017

urso commented Jul 2, 2017

dedemorton commented Jan 22, 2018 •

edited

Loading

		if err != nil {
		// XXX: We don't drop the event, but continue filtering here iff the most

Move processors into new pipeline #4554

Move processors into new pipeline #4554

Conversation

urso commented Jun 24, 2017 • edited Loading

ruflin Jun 27, 2017

Choose a reason for hiding this comment

ruflin left a comment

Choose a reason for hiding this comment

ruflin Jul 1, 2017

Choose a reason for hiding this comment

urso Jul 2, 2017

Choose a reason for hiding this comment

ruflin Jul 1, 2017

Choose a reason for hiding this comment

urso Jul 2, 2017

Choose a reason for hiding this comment

ruflin Jul 1, 2017

Choose a reason for hiding this comment

urso Jul 2, 2017

Choose a reason for hiding this comment

urso commented Jul 2, 2017

dedemorton commented Jan 22, 2018 • edited Loading

urso commented Jun 24, 2017 •

edited

Loading

dedemorton commented Jan 22, 2018 •

edited

Loading