1 Star 0 Fork 0

sqos/beats

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
processor.go 2.66 KB
一键复制 编辑 原始数据 按行查看 历史
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, ", ")
}
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/sqos/beats.git
git@gitee.com:sqos/beats.git
sqos
beats
beats
v6.1.3

搜索帮助