-
Notifications
You must be signed in to change notification settings - Fork 4.9k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Move filebeat to new publisher pipeline
- remove filebeat/spooler and filebeat/publisher package -> all spooling and reporting published events is moved to the publisher pipeline. Difference between spooler/publisher to new pipeline is: The new publisher pipeline operates fully asynchronous - have filebeat register an eventACKer with the publisher pipeline. The eventACKer will forward state updates of ACKed events to the registrar - filebeat uses beat.Event for events - update util.Data to use beat.Event: - store state in event.Private field for consumption by registry - changes to filebeat/channels package: - introduce OutletFactory: connect to publisher pipeline, applying common prospector settings - remove Outleter.SetSignal and Outleter.OnEventSignal - add Outleter.Close - introduce SubOutlet (separate closing): - when a suboutlet is closed, the original outlet is still active - if underlying outlet is closed, the suboutlet becomes closed as well (can not forward anymore) - introduce CloseOnSignal: close Outlet once a 'done' channel is closed - most functionality from harvester.Forwarder is moved into the outlet/publisher pipeline client - fix: ensure client events listener properly installed Note: Outlet shutdown with prospectors and harvesters is somewhat delicate. There are 3 shutdown signals to take into account: - filebeat.done - harvester.done - outDone (signal used to unblock prospectors from registrar on shutdown). An outlet is shared between all harvesters of a prospector and the prospector itself. If outDone is closed, all outlets will be closed, unblocking potentially waiting harvesters and prosepectors on filebeat shutdown. The prospector uses a sub-outlet for sending state updates (being closed on filebeat.done). The harvesters sub-outlet is closed when harveser.done is closed. The signals are only required to unblock an harvester/prospector on exit. On normal shutdown, the outlets are closed after all workers have been finished.
- Loading branch information
urso
committed
Jul 11, 2017
1 parent
37f4abb
commit 8e66d95
Showing
43 changed files
with
643 additions
and
1,089 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,41 @@ | ||
package beater | ||
|
||
import ( | ||
"github.com/elastic/beats/filebeat/util" | ||
"github.com/elastic/beats/libbeat/publisher/beat" | ||
) | ||
|
||
// eventAcker handles publisher pipeline ACKs and forwards | ||
// them to the registrar. | ||
type eventACKer struct { | ||
out successLogger | ||
} | ||
|
||
type successLogger interface { | ||
Published(events []*util.Data) bool | ||
} | ||
|
||
func newEventACKer(out successLogger) *eventACKer { | ||
return &eventACKer{out: out} | ||
} | ||
|
||
func (a *eventACKer) ackEvents(events []beat.Event) { | ||
data := make([]*util.Data, 0, len(events)) | ||
for _, event := range events { | ||
p := event.Private | ||
if p == nil { | ||
continue | ||
} | ||
|
||
datum, ok := p.(*util.Data) | ||
if !ok || !datum.HasState() { | ||
continue | ||
} | ||
|
||
data = append(data, datum) | ||
} | ||
|
||
if len(data) > 0 { | ||
a.out.Published(data) | ||
} | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,128 @@ | ||
package channel | ||
|
||
import ( | ||
"sync" | ||
|
||
"github.com/elastic/beats/libbeat/common" | ||
"github.com/elastic/beats/libbeat/processors" | ||
"github.com/elastic/beats/libbeat/publisher/bc/publisher" | ||
"github.com/elastic/beats/libbeat/publisher/beat" | ||
) | ||
|
||
type OutletFactory struct { | ||
done <-chan struct{} | ||
pipeline publisher.Publisher | ||
|
||
eventer beat.ClientEventer | ||
wgEvents *sync.WaitGroup | ||
} | ||
|
||
// clientEventer adjusts wgEvents if events are dropped during shutdown. | ||
type clientEventer struct { | ||
wgEvents *sync.WaitGroup | ||
} | ||
|
||
// prospectorOutletConfig defines common prospector settings | ||
// for the publisher pipline. | ||
type prospectorOutletConfig struct { | ||
// event processing | ||
common.EventMetadata `config:",inline"` // Fields and tags to add to events. | ||
Processors processors.PluginConfig `config:"processors"` | ||
|
||
// implicit event fields | ||
Type string `config:"type"` // prospector.type | ||
|
||
// hidden filebeat modules settings | ||
Module string `config:"_module_name"` // hidden setting | ||
Fileset string `config:"_fileset_name"` // hidden setting | ||
|
||
// Output meta data settings | ||
Pipeline string `config:"pipeline"` // ES Ingest pipeline name | ||
|
||
} | ||
|
||
// NewOutletFactory creates a new outlet factory for | ||
// connecting a prospector to the publisher pipeline. | ||
func NewOutletFactory( | ||
done <-chan struct{}, | ||
pipeline publisher.Publisher, | ||
wgEvents *sync.WaitGroup, | ||
) *OutletFactory { | ||
o := &OutletFactory{ | ||
done: done, | ||
pipeline: pipeline, | ||
wgEvents: wgEvents, | ||
} | ||
|
||
if wgEvents != nil { | ||
o.eventer = &clientEventer{wgEvents} | ||
} | ||
|
||
return o | ||
} | ||
|
||
// Create builds a new Outleter, while applying common prospector settings. | ||
// Prospectors and all harvesters use the same pipeline client instance. | ||
// This guarantees ordering between events as required by the registrar for | ||
// file.State updates | ||
func (f *OutletFactory) Create(cfg *common.Config) (Outleter, error) { | ||
config := prospectorOutletConfig{} | ||
if err := cfg.Unpack(&config); err != nil { | ||
return nil, err | ||
} | ||
|
||
processors, err := processors.New(config.Processors) | ||
if err != nil { | ||
return nil, err | ||
} | ||
|
||
setMeta := func(to common.MapStr, key, value string) { | ||
if value != "" { | ||
to[key] = value | ||
} | ||
} | ||
|
||
meta := common.MapStr{} | ||
setMeta(meta, "pipeline", config.Pipeline) | ||
|
||
fields := common.MapStr{} | ||
setMeta(fields, "module", config.Module) | ||
setMeta(fields, "name", config.Fileset) | ||
if len(fields) > 0 { | ||
fields = common.MapStr{ | ||
"fileset": fields, | ||
} | ||
} | ||
if config.Type != "" { | ||
fields["prospector"] = common.MapStr{ | ||
"type": config.Type, | ||
} | ||
} | ||
|
||
client, err := f.pipeline.ConnectX(beat.ClientConfig{ | ||
PublishMode: beat.GuaranteedSend, | ||
EventMetadata: config.EventMetadata, | ||
Meta: meta, | ||
Fields: fields, | ||
Processor: processors, | ||
Events: f.eventer, | ||
}) | ||
if err != nil { | ||
return nil, err | ||
} | ||
|
||
outlet := newOutlet(client, f.wgEvents) | ||
if f.done != nil { | ||
return CloseOnSignal(outlet, f.done), nil | ||
} | ||
return outlet, nil | ||
} | ||
|
||
func (*clientEventer) Closing() {} | ||
func (*clientEventer) Closed() {} | ||
func (*clientEventer) Published() {} | ||
|
||
func (c *clientEventer) FilteredOut(_ beat.Event) {} | ||
func (c *clientEventer) DroppedOnPublish(_ beat.Event) { | ||
c.wgEvents.Done() | ||
} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,11 +1,15 @@ | ||
package channel | ||
|
||
import "github.com/elastic/beats/filebeat/util" | ||
import ( | ||
"github.com/elastic/beats/filebeat/util" | ||
"github.com/elastic/beats/libbeat/common" | ||
) | ||
|
||
// OutletFactory is used to create a new Outlet instance | ||
type OutleterFactory func(*common.Config) (Outleter, error) | ||
|
||
// Outleter is the outlet for a prospector | ||
type Outleter interface { | ||
SetSignal(signal <-chan struct{}) | ||
OnEventSignal(data *util.Data) bool | ||
Close() error | ||
OnEvent(data *util.Data) bool | ||
Copy() Outleter | ||
} |
Oops, something went wrong.