1 Star 0 Fork 0

zhangjungang/beats

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
processor.go 1.69 KB
一键复制 编辑 原始数据 按行查看 历史
package processors
import (
"fmt"
"strings"
"github.com/elastic/beats/libbeat/common"
"github.com/elastic/beats/libbeat/logp"
)
type Processors struct {
list []Processor
}
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)
}
// Applies a sequence of processing rules and returns the filtered event
func (procs *Processors) Run(event common.MapStr) common.MapStr {
// Check if processors are set, just return event if not
if len(procs.list) == 0 {
return event
}
// clone the event at first, before starting filtering
filtered := event.Clone()
var err error
for _, p := range procs.list {
filtered, err = p.Run(filtered)
if err != nil {
logp.Debug("filter", "fail to apply processor %s: %s", p, err)
}
if filtered == nil {
// drop event
return nil
}
}
return filtered
}
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/zhangjungang/beats.git
git@gitee.com:zhangjungang/beats.git
zhangjungang
beats
beats
v6.0.0-alpha2

搜索帮助