代码拉取完成,页面将自动刷新
package processors
import (
"fmt"
"strings"
"github.com/elastic/beats/libbeat/beat"
"github.com/elastic/beats/libbeat/common"
"github.com/elastic/beats/libbeat/logp"
)
type Processors struct {
List []Processor
}
type Processor interface {
Run(event *beat.Event) (*beat.Event, error)
String() string
}
func New(config PluginConfig) (*Processors, error) {
procs := Processors{}
for _, processor := range config {
if len(processor) != 1 {
return nil, fmt.Errorf("each processor needs to have exactly one action, but found %d actions",
len(processor))
}
for processorName, cfg := range processor {
gen, exists := registry.reg[processorName]
if !exists {
return nil, fmt.Errorf("the processor %s doesn't exist", processorName)
}
cfg.PrintDebugf("Configure processor '%v' with:", processorName)
constructor := gen.Plugin()
plugin, err := constructor(cfg)
if err != nil {
return nil, err
}
procs.add(plugin)
}
}
logp.Debug("processors", "Processors: %v", procs)
return &procs, nil
}
func (procs *Processors) add(p Processor) {
procs.List = append(procs.List, p)
}
// RunBC (run backwards-compatible) applies the processors, by providing the
// old interface based on common.MapStr.
// The event us temporarily converted to beat.Event. By this 'conversion' the
// '@timestamp' field can not be accessed by processors.
// Note: this method will be removed, when the publisher pipeline BC-API is to
// be removed.
func (procs *Processors) RunBC(event common.MapStr) common.MapStr {
ret := procs.Run(&beat.Event{Fields: event})
if ret == nil {
return nil
}
return ret.Fields
}
func (procs *Processors) All() []beat.Processor {
if procs == nil || len(procs.List) == 0 {
return nil
}
ret := make([]beat.Processor, len(procs.List))
for i, p := range procs.List {
ret[i] = p
}
return ret
}
// Applies a sequence of processing rules and returns the filtered event
func (procs *Processors) Run(event *beat.Event) *beat.Event {
// Check if processors are set, just return event if not
if len(procs.List) == 0 {
return event
}
for _, p := range procs.List {
var err error
event, err = p.Run(event)
if err != nil {
// XXX: We don't drop the event, but continue filtering here iff the most
// recent processor did return an event.
// We want processors having this kind of implicit behavior
// on errors?
logp.Debug("filter", "fail to apply processor %s: %s", p, err)
}
if event == nil {
// drop event
return nil
}
}
return event
}
func (procs Processors) String() string {
var s []string
for _, p := range procs.List {
s = append(s, p.String())
}
return strings.Join(s, ", ")
}
此处可能存在不合适展示的内容,页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容,可点击提交进行申诉,我们将尽快为您处理。