1 Star 0 Fork 0

zhangjungang/beats

加入 Gitee
与超过 1200万 开发者一起发现、参与优秀开源项目,私有仓库也完全免费 :)
免费加入
克隆/下载
eventreader_fsnotify.go 2.34 KB
一键复制 编辑 原始数据 按行查看 历史
// +build linux freebsd openbsd netbsd windows
package file_integrity
import (
"syscall"
"time"
"github.com/fsnotify/fsnotify"
"github.com/pkg/errors"
"github.com/elastic/beats/auditbeat/module/file_integrity/monitor"
"github.com/elastic/beats/libbeat/logp"
)
type reader struct {
watcher monitor.Watcher
config Config
eventC chan Event
log *logp.Logger
}
// NewEventReader creates a new EventProducer backed by fsnotify.
func NewEventReader(c Config) (EventProducer, error) {
watcher, err := monitor.New(c.Recursive)
if err != nil {
return nil, err
}
return &reader{
watcher: watcher,
config: c,
eventC: make(chan Event, 1),
log: logp.NewLogger(moduleName),
}, nil
}
func (r *reader) Start(done <-chan struct{}) (<-chan Event, error) {
for _, p := range r.config.Paths {
if err := r.watcher.Add(p); err != nil {
if err == syscall.EMFILE {
r.log.Warnw("Failed to add watch (check the max number of "+
"open files allowed with 'ulimit -a')",
"file_path", p, "error", err)
} else {
r.log.Warnw("Failed to add watch", "file_path", p, "error", err)
}
}
}
if err := r.watcher.Start(); err != nil {
return nil, errors.Wrap(err, "unable to start watcher")
}
go r.consumeEvents(done)
r.log.Infow("Started fsnotify watcher",
"file_path", r.config.Paths,
"recursive", r.config.Recursive)
return r.eventC, nil
}
func (r *reader) consumeEvents(done <-chan struct{}) {
defer close(r.eventC)
defer r.watcher.Close()
for {
select {
case <-done:
r.log.Debug("fsnotify reader terminated")
return
case event := <-r.watcher.EventChannel():
if event.Name == "" || r.config.IsExcludedPath(event.Name) {
continue
}
r.log.Debugw("Received fsnotify event",
"file_path", event.Name,
"event_flags", event.Op)
start := time.Now()
e := NewEvent(event.Name, opToAction(event.Op), SourceFSNotify,
r.config.MaxFileSizeBytes, r.config.HashTypes)
e.rtt = time.Since(start)
r.eventC <- e
case err := <-r.watcher.ErrorChannel():
r.log.Warnw("fsnotify watcher error", "error", err)
}
}
}
func opToAction(op fsnotify.Op) Action {
switch op {
case fsnotify.Create:
return Created
case fsnotify.Write:
return Updated
case fsnotify.Remove:
return Deleted
case fsnotify.Rename:
return Moved
case fsnotify.Chmod:
return AttributesModified
default:
return 0
}
}
马建仓 AI 助手
尝试更多
代码解读
代码找茬
代码优化
1
https://gitee.com/zhangjungang/beats.git
git@gitee.com:zhangjungang/beats.git
zhangjungang
beats
beats
v6.2.2

搜索帮助