Asynchronous Promtail stages

[cyriltovena]

This PR essentially changes promtail stages interface to uses channel, the idea is from @jeschkies based on https://blog.golang.org/pipelines .

The old interface was synchronous:

type Stage interface {
	Process(labels model.LabelSet, extracted map[string]interface{}, time *time.Time, entry *string)
	Name() string
}

Which means it was difficult to buffer/skip/duplicate entries before passing them down to another stages. For instance we were using a drop label that was interpreted by the pipeline to skip entries.

Now this is much simpler, the new interface uses two channels; one inbound and one outbound:

type Stage interface {
	Name() string
	Run(chan Entry) chan Entry
}

The pipeline will chain all channels together, in->out -> in -> out -> .... etc...

Unfortunately once you start going async, you have to go async bottom-up in the whole stack, this means promtail client interface has also changed from:

type Client interface {
  Handle(labels model.LabelSet,time time.Time,line string) err
  Stop()
}

to

type Client interface {
  Chan() chan<- api.Entry
  Stop()
}

Now as you can imagine, we can wrap the Client chanel (sink) with the new pipeline to introduce mutation/buffering/deletion and so on before sending them.

This means we will be able to move forward with #74 much more easily without hacking around the synchronous limitation, @jeschkies has a PR already in good shape that will be following.

Special notes for your reviewer:

Lot of channel and goroutine usage !

All tests are still the same, this is a Noop from the user point of view, I've backported all stages.

Checklist

• Documentation added
• Tests updated

[codecov-io]

Codecov Report

Merging #2996 (8da40aa) into master (854930e) will increase coverage by 0.13%.
The diff coverage is 85.31%.

@@            Coverage Diff             @@
##           master    #2996      +/-   ##
==========================================
+ Coverage   62.39%   62.53%   +0.13%     
==========================================
  Files         185      185              
  Lines       15626    15731     +105     
==========================================
+ Hits         9750     9837      +87     
- Misses       4949     4975      +26     
+ Partials      927      919       -8     

Impacted Files	Coverage Δ
cmd/docker-driver/loki.go	0.00% <0.00%> (ø)
cmd/fluent-bit/dque.go	0.00% <0.00%> (ø)
pkg/promtail/promtail.go	63.63% <ø> (ø)
pkg/promtail/targets/file/filetargetmanager.go	0.00% <0.00%> (ø)
pkg/promtail/targets/lokipush/pushtarget.go	59.74% <88.88%> (+5.68%)	⬆️
pkg/promtail/client/client.go	77.69% <93.33%> (-0.63%)	⬇️
pkg/promtail/client/logger.go	79.31% <93.75%> (+4.31%)	⬆️
pkg/logentry/stages/match.go	90.90% <94.91%> (+0.43%)	⬆️
pkg/logentry/stages/drop.go	91.58% <96.96%> (+3.23%)	⬆️
cmd/fluent-bit/loki.go	82.22% <100.00%> (+1.73%)	⬆️
... and 34 more

[jeschkies]

Thanks for this great effort.

Did you mean The old interface was sync: instead of The old interface was async: in the PR message?

[cyriltovena]

Thanks for this great effort.

Did you mean The old interface was sync: instead of The old interface was async: in the PR message?

Yep sorry :)

[owen-d]

I gave it a quick review but spent some extra time in the api and stage packages. I like this new approach and think it looks good. I'm curious to see how well the runtime handles the new work (more goroutines, etc) but I expect it'll be much better.

That being said, I don't spend a ton of time in these packages usually.

[cyriltovena]

I gave it a quick review but spent some extra time in the api and stage packages. I like this new approach and think it looks good. I'm curious to see how well the runtime handles the new work (more goroutines, etc) but I expect it'll be much better.

That being said, I don't spend a ton of time in these packages usually.

Thanks @owen-d I still need to address Karsten feedback before merging.

I also want to take it for a spin in dev.

[cyriltovena]

Everything seems to run smoothly and CPU is lower than before 👍 Getting test to pass and I'll merge this.