Mirrors/sftpgo
1
0
Fork 0
mirror of https://github.com/drakkan/sftpgo.git synced 2025-12-07 06:40:54 +03:00
Code Issues Packages Projects Releases Wiki Activity

notifiers plugin: add settings to retry unhandled events

Browse Source
This commit is contained in:
Nicola Murino
2021-07-20 12:51:21 +02:00
parent 13183a9f76
commit c900cde8e4
12 changed files with 202 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

149
sdk/plugin/notifier.go
View File

@@ -4,19 +4,25 @@ import (
"crypto/sha256"
"fmt"
"os/exec"
"sync"
"time"
"github.com/hashicorp/go-hclog"
"github.com/hashicorp/go-plugin"
"google.golang.org/protobuf/types/known/timestamppb"
"github.com/drakkan/sftpgo/v2/logger"
"github.com/drakkan/sftpgo/v2/sdk/plugin/notifier"
"github.com/drakkan/sftpgo/v2/sdk/plugin/notifier/proto"
"github.com/drakkan/sftpgo/v2/util"
)
// NotifierConfig defines configuration parameters for notifiers plugins
type NotifierConfig struct {
FsEvents []string `json:"fs_events" mapstructure:"fs_events"`
UserEvents []string `json:"user_events" mapstructure:"user_events"`
FsEvents []string `json:"fs_events" mapstructure:"fs_events"`
UserEvents []string `json:"user_events" mapstructure:"user_events"`
RetryMaxTime int `json:"retry_max_time" mapstructure:"retry_max_time"`
RetryQueueMaxSize int `json:"retry_queue_max_size" mapstructure:"retry_queue_max_size"`
}
func (c *NotifierConfig) hasActions() bool {
@@ -29,15 +35,88 @@ func (c *NotifierConfig) hasActions() bool {
return false
}
type eventsQueue struct {
sync.RWMutex
fsEvents []*proto.FsEvent
userEvents []*proto.UserEvent
}
func (q *eventsQueue) addFsEvent(timestamp time.Time, action, username, fsPath, fsTargetPath, sshCmd, protocol string, fileSize int64, status int) {
q.Lock()
defer q.Unlock()
q.fsEvents = append(q.fsEvents, &proto.FsEvent{
Timestamp: timestamppb.New(timestamp),
Action: action,
Username: username,
FsPath: fsPath,
FsTargetPath: fsTargetPath,
SshCmd: sshCmd,
FileSize: fileSize,
Protocol: protocol,
Status: int32(status),
})
}
func (q *eventsQueue) addUserEvent(timestamp time.Time, action string, userAsJSON []byte) {
q.Lock()
defer q.Unlock()
q.userEvents = append(q.userEvents, &proto.UserEvent{
Timestamp: timestamppb.New(timestamp),
Action: action,
User: userAsJSON,
})
}
func (q *eventsQueue) popFsEvent() *proto.FsEvent {
q.Lock()
defer q.Unlock()
if len(q.fsEvents) == 0 {
return nil
}
truncLen := len(q.fsEvents) - 1
ev := q.fsEvents[truncLen]
q.fsEvents[truncLen] = nil
q.fsEvents = q.fsEvents[:truncLen]
return ev
}
func (q *eventsQueue) popUserEvent() *proto.UserEvent {
q.Lock()
defer q.Unlock()
if len(q.userEvents) == 0 {
return nil
}
truncLen := len(q.userEvents) - 1
ev := q.userEvents[truncLen]
q.userEvents[truncLen] = nil
q.userEvents = q.userEvents[:truncLen]
return ev
}
func (q *eventsQueue) getSize() int {
q.RLock()
defer q.RUnlock()
return len(q.userEvents) + len(q.fsEvents)
}
type notifierPlugin struct {
config Config
notifier notifier.Notifier
client *plugin.Client
queue *eventsQueue
}
func newNotifierPlugin(config Config) (*notifierPlugin, error) {
p := &notifierPlugin{
config: config,
queue: &eventsQueue{},
}
if err := p.initialize(); err != nil {
logger.Warn(logSender, "", "unable to create notifier plugin: %v, config %+v", err, config)
@@ -101,7 +180,21 @@ func (p *notifierPlugin) initialize() error {
return nil
}
func (p *notifierPlugin) notifyFsAction(action, username, fsPath, fsTargetPath, sshCmd, protocol string, fileSize int64, errAction error) {
func (p *notifierPlugin) canQueueEvent(timestamp time.Time) bool {
if p.config.NotifierOptions.RetryMaxTime == 0 {
return false
}
if time.Now().After(timestamp.Add(time.Duration(p.config.NotifierOptions.RetryMaxTime) * time.Second)) {
return false
}
if p.config.NotifierOptions.RetryQueueMaxSize > 0 {
return p.queue.getSize() < p.config.NotifierOptions.RetryQueueMaxSize
}
return true
}
func (p *notifierPlugin) notifyFsAction(timestamp time.Time, action, username, fsPath, fsTargetPath, sshCmd,
protocol string, fileSize int64, errAction error) {
if !util.IsStringInSlice(action, p.config.NotifierOptions.FsEvents) {
return
}
@@ -111,13 +204,11 @@ func (p *notifierPlugin) notifyFsAction(action, username, fsPath, fsTargetPath,
if errAction != nil {
status = 0
}
if err := p.notifier.NotifyFsEvent(action, username, fsPath, fsTargetPath, sshCmd, protocol, fileSize, status); err != nil {
logger.Warn(logSender, "", "unable to send fs action notification to plugin %v: %v", p.config.Cmd, err)
}
p.sendFsEvent(timestamp, action, username, fsPath, fsTargetPath, sshCmd, protocol, fileSize, status)
}()
}
func (p *notifierPlugin) notifyUserAction(action string, user Renderer) {
func (p *notifierPlugin) notifyUserAction(timestamp time.Time, action string, user Renderer) {
if !util.IsStringInSlice(action, p.config.NotifierOptions.UserEvents) {
return
}
@@ -128,8 +219,46 @@ func (p *notifierPlugin) notifyUserAction(action string, user Renderer) {
logger.Warn(logSender, "", "unable to render user as json for action %v: %v", action, err)
return
}
if err := p.notifier.NotifyUserEvent(action, userAsJSON); err != nil {
logger.Warn(logSender, "", "unable to send user action notification to plugin %v: %v", p.config.Cmd, err)
}
p.sendUserEvent(timestamp, action, userAsJSON)
}()
}
func (p *notifierPlugin) sendFsEvent(timestamp time.Time, action, username, fsPath, fsTargetPath, sshCmd,
protocol string, fileSize int64, status int) {
if err := p.notifier.NotifyFsEvent(timestamp, action, username, fsPath, fsTargetPath, sshCmd, protocol, fileSize, status); err != nil {
logger.Warn(logSender, "", "unable to send fs action notification to plugin %v: %v", p.config.Cmd, err)
if p.canQueueEvent(timestamp) {
p.queue.addFsEvent(timestamp, action, username, fsPath, fsTargetPath, sshCmd, protocol, fileSize, status)
}
}
}
func (p *notifierPlugin) sendUserEvent(timestamp time.Time, action string, userAsJSON []byte) {
if err := p.notifier.NotifyUserEvent(timestamp, action, userAsJSON); err != nil {
logger.Warn(logSender, "", "unable to send user action notification to plugin %v: %v", p.config.Cmd, err)
if p.canQueueEvent(timestamp) {
p.queue.addUserEvent(timestamp, action, userAsJSON)
}
}
}
func (p *notifierPlugin) sendQueuedEvents() {
queueSize := p.queue.getSize()
if queueSize == 0 {
return
}
logger.Debug(logSender, "", "check queued events for notifier %#v, events size: %v", p.config.Cmd, queueSize)
fsEv := p.queue.popFsEvent()
for fsEv != nil {
go p.sendFsEvent(fsEv.Timestamp.AsTime(), fsEv.Action, fsEv.Username, fsEv.FsPath, fsEv.FsTargetPath,
fsEv.SshCmd, fsEv.Protocol, fsEv.FileSize, int(fsEv.Status))
fsEv = p.queue.popFsEvent()
}
userEv := p.queue.popUserEvent()
for userEv != nil {
go p.sendUserEvent(userEv.Timestamp.AsTime(), userEv.Action, userEv.User)
userEv = p.queue.popUserEvent()
}
logger.Debug(logSender, "", "queued events sent for notifier %#v, new events size: %v", p.config.Cmd, p.queue.getSize())
}

12
sdk/plugin/notifier/grpc.go
View File

@@ -20,12 +20,12 @@ type GRPCClient struct {
}
// NotifyFsEvent implements the Notifier interface
func (c *GRPCClient) NotifyFsEvent(action, username, fsPath, fsTargetPath, sshCmd, protocol string, fileSize int64, status int) error {
func (c *GRPCClient) NotifyFsEvent(timestamp time.Time, action, username, fsPath, fsTargetPath, sshCmd, protocol string, fileSize int64, status int) error {
ctx, cancel := context.WithTimeout(context.Background(), rpcTimeout)
defer cancel()
_, err := c.client.SendFsEvent(ctx, &proto.FsEvent{
Timestamp: timestamppb.New(time.Now()),
Timestamp: timestamppb.New(timestamp),
Action: action,
Username: username,
FsPath: fsPath,
@@ -40,12 +40,12 @@ func (c *GRPCClient) NotifyFsEvent(action, username, fsPath, fsTargetPath, sshCm
}
// NotifyUserEvent implements the Notifier interface
func (c *GRPCClient) NotifyUserEvent(action string, user []byte) error {
func (c *GRPCClient) NotifyUserEvent(timestamp time.Time, action string, user []byte) error {
ctx, cancel := context.WithTimeout(context.Background(), rpcTimeout)
defer cancel()
_, err := c.client.SendUserEvent(ctx, &proto.UserEvent{
Timestamp: timestamppb.New(time.Now()),
Timestamp: timestamppb.New(timestamp),
Action: action,
User: user,
})
@@ -60,13 +60,13 @@ type GRPCServer struct {
// SendFsEvent implements the serve side fs notify method
func (s *GRPCServer) SendFsEvent(ctx context.Context, req *proto.FsEvent) (*emptypb.Empty, error) {
err := s.Impl.NotifyFsEvent(req.Action, req.Username, req.FsPath, req.FsTargetPath, req.SshCmd,
err := s.Impl.NotifyFsEvent(req.Timestamp.AsTime(), req.Action, req.Username, req.FsPath, req.FsTargetPath, req.SshCmd,
req.Protocol, req.FileSize, int(req.Status))
return &emptypb.Empty{}, err
}
// SendUserEvent implements the serve side user notify method
func (s *GRPCServer) SendUserEvent(ctx context.Context, req *proto.UserEvent) (*emptypb.Empty, error) {
err := s.Impl.NotifyUserEvent(req.Action, req.User)
err := s.Impl.NotifyUserEvent(req.Timestamp.AsTime(), req.Action, req.User)
return &emptypb.Empty{}, err
}

6
sdk/plugin/notifier/notifier.go
View File

@@ -5,6 +5,7 @@ package notifier
import (
"context"
"time"
"github.com/hashicorp/go-plugin"
"google.golang.org/grpc"
@@ -31,8 +32,9 @@ var PluginMap = map[string]plugin.Plugin{
// Notifier defines the interface for notifiers plugins
type Notifier interface {
NotifyFsEvent(action, username, fsPath, fsTargetPath, sshCmd, protocol string, fileSize int64, status int) error
NotifyUserEvent(action string, user []byte) error
NotifyFsEvent(timestamp time.Time, action, username, fsPath, fsTargetPath, sshCmd, protocol string,
fileSize int64, status int) error
NotifyUserEvent(timestamp time.Time, action string, user []byte) error
}
// Plugin defines the implementation to serve/connect to a notifier plugin

31
sdk/plugin/plugin.go
View File

@@ -145,30 +145,23 @@ func (m *Manager) validateConfigs() error {
}
// NotifyFsEvent sends the fs event notifications using any defined notifier plugins
func (m *Manager) NotifyFsEvent(action, username, fsPath, fsTargetPath, sshCmd, protocol string, fileSize int64, err error) {
func (m *Manager) NotifyFsEvent(timestamp time.Time, action, username, fsPath, fsTargetPath, sshCmd, protocol string,
fileSize int64, err error) {
m.notifLock.RLock()
defer m.notifLock.RUnlock()
for _, n := range m.notifiers {
if n.exited() {
logger.Warn(logSender, "", "notifer plugin %v is not active, unable to send fs event", n.config.Cmd)
continue
}
n.notifyFsAction(action, username, fsPath, fsTargetPath, sshCmd, protocol, fileSize, err)
n.notifyFsAction(timestamp, action, username, fsPath, fsTargetPath, sshCmd, protocol, fileSize, err)
}
}
// NotifyUserEvent sends the user event notifications using any defined notifier plugins
func (m *Manager) NotifyUserEvent(action string, user Renderer) {
func (m *Manager) NotifyUserEvent(timestamp time.Time, action string, user Renderer) {
m.notifLock.RLock()
defer m.notifLock.RUnlock()
for _, n := range m.notifiers {
if n.exited() {
logger.Warn(logSender, "", "notifer plugin %v is not active, unable to send user event", n.config.Cmd)
continue
}
n.notifyUserAction(action, user)
n.notifyUserAction(timestamp, action, user)
}
}
@@ -192,7 +185,11 @@ func (m *Manager) checkCrashedPlugins() {
m.notifLock.RLock()
for idx, n := range m.notifiers {
if n.exited() {
defer Handler.restartNotifierPlugin(n.config, idx)
defer func(cfg Config, index int) {
Handler.restartNotifierPlugin(cfg, index)
}(n.config, idx)
} else {
n.sendQueuedEvents()
}
}
m.notifLock.RUnlock()
@@ -200,7 +197,9 @@ func (m *Manager) checkCrashedPlugins() {
m.kmsLock.RLock()
for idx, k := range m.kms {
if k.exited() {
defer Handler.restartKMSPlugin(k.config, idx)
defer func(cfg Config, index int) {
Handler.restartKMSPlugin(cfg, index)
}(k.config, idx)
}
}
m.kmsLock.RUnlock()
@@ -210,7 +209,7 @@ func (m *Manager) restartNotifierPlugin(config Config, idx int) {
if atomic.LoadInt32(&m.closed) == 1 {
return
}
logger.Info(logSender, "", "try to restart notifier crashed plugin %#v, idx: %v", config.Cmd, idx)
logger.Info(logSender, "", "try to restart crashed notifier plugin %#v, idx: %v", config.Cmd, idx)
plugin, err := newNotifierPlugin(config)
if err != nil {
logger.Warn(logSender, "", "unable to restart notifier plugin %#v, err: %v", config.Cmd, err)
@@ -218,8 +217,10 @@ func (m *Manager) restartNotifierPlugin(config Config, idx int) {
}
m.notifLock.Lock()
plugin.queue = m.notifiers[idx].queue
m.notifiers[idx] = plugin
m.notifLock.Unlock()
plugin.sendQueuedEvents()
}
func (m *Manager) restartKMSPlugin(config Config, idx int) {