mirror of
https://github.com/drakkan/sftpgo.git
synced 2025-12-07 14:50:55 +03:00
fs events: add elapsed field to notifications
Signed-off-by: Nicola Murino <nicola.murino@gmail.com>
This commit is contained in:
@@ -92,7 +92,7 @@ func ExecutePreAction(conn *BaseConnection, operation, filePath, virtualPath str
|
||||
return 0, nil
|
||||
}
|
||||
event = newActionNotification(&conn.User, operation, filePath, virtualPath, "", "", "",
|
||||
conn.protocol, conn.GetRemoteIP(), conn.ID, fileSize, openFlags, conn.getNotificationStatus(nil))
|
||||
conn.protocol, conn.GetRemoteIP(), conn.ID, fileSize, openFlags, conn.getNotificationStatus(nil), 0)
|
||||
if hasNotifiersPlugin {
|
||||
plugin.Handler.NotifyFsEvent(event)
|
||||
}
|
||||
@@ -127,7 +127,7 @@ func ExecutePreAction(conn *BaseConnection, operation, filePath, virtualPath str
|
||||
|
||||
// ExecuteActionNotification executes the defined hook, if any, for the specified action
|
||||
func ExecuteActionNotification(conn *BaseConnection, operation, filePath, virtualPath, target, virtualTarget, sshCmd string,
|
||||
fileSize int64, err error,
|
||||
fileSize int64, err error, elapsed int64,
|
||||
) error {
|
||||
hasNotifiersPlugin := plugin.Handler.HasNotifiers()
|
||||
hasHook := util.Contains(Config.Actions.ExecuteOn, operation)
|
||||
@@ -136,7 +136,7 @@ func ExecuteActionNotification(conn *BaseConnection, operation, filePath, virtua
|
||||
return nil
|
||||
}
|
||||
notification := newActionNotification(&conn.User, operation, filePath, virtualPath, target, virtualTarget, sshCmd,
|
||||
conn.protocol, conn.GetRemoteIP(), conn.ID, fileSize, 0, conn.getNotificationStatus(err))
|
||||
conn.protocol, conn.GetRemoteIP(), conn.ID, fileSize, 0, conn.getNotificationStatus(err), elapsed)
|
||||
if hasNotifiersPlugin {
|
||||
plugin.Handler.NotifyFsEvent(notification)
|
||||
}
|
||||
@@ -152,6 +152,7 @@ func ExecuteActionNotification(conn *BaseConnection, operation, filePath, virtua
|
||||
FsTargetPath: notification.TargetPath,
|
||||
ObjectName: path.Base(notification.VirtualPath),
|
||||
FileSize: notification.FileSize,
|
||||
Elapsed: notification.Elapsed,
|
||||
Protocol: notification.Protocol,
|
||||
IP: notification.IP,
|
||||
Role: notification.Role,
|
||||
@@ -190,7 +191,7 @@ func newActionNotification(
|
||||
user *dataprovider.User,
|
||||
operation, filePath, virtualPath, target, virtualTarget, sshCmd, protocol, ip, sessionID string,
|
||||
fileSize int64,
|
||||
openFlags, status int,
|
||||
openFlags, status int, elapsed int64,
|
||||
) *notifier.FsEvent {
|
||||
var bucket, endpoint string
|
||||
|
||||
@@ -232,6 +233,7 @@ func newActionNotification(
|
||||
OpenFlags: openFlags,
|
||||
Role: user.Role,
|
||||
Timestamp: time.Now().UnixNano(),
|
||||
Elapsed: elapsed,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -321,6 +323,7 @@ func notificationAsEnvVars(event *notifier.FsEvent) []string {
|
||||
fmt.Sprintf("SFTPGO_ACTION_VIRTUAL_TARGET=%s", event.VirtualTargetPath),
|
||||
fmt.Sprintf("SFTPGO_ACTION_SSH_CMD=%s", event.SSHCmd),
|
||||
fmt.Sprintf("SFTPGO_ACTION_FILE_SIZE=%d", event.FileSize),
|
||||
fmt.Sprintf("SFTPGO_ACTION_ELAPSED=%d", event.Elapsed),
|
||||
fmt.Sprintf("SFTPGO_ACTION_FS_PROVIDER=%d", event.FsProvider),
|
||||
fmt.Sprintf("SFTPGO_ACTION_BUCKET=%s", event.Bucket),
|
||||
fmt.Sprintf("SFTPGO_ACTION_ENDPOINT=%s", event.Endpoint),
|
||||
|
||||
@@ -71,7 +71,7 @@ func TestNewActionNotification(t *testing.T) {
|
||||
c := NewBaseConnection("id", ProtocolSSH, "", "", user)
|
||||
sessionID := xid.New().String()
|
||||
a := newActionNotification(&user, operationDownload, "path", "vpath", "target", "", "", ProtocolSFTP, "", sessionID,
|
||||
123, 0, c.getNotificationStatus(errors.New("fake error")))
|
||||
123, 0, c.getNotificationStatus(errors.New("fake error")), 0)
|
||||
assert.Equal(t, user.Username, a.Username)
|
||||
assert.Equal(t, 0, len(a.Bucket))
|
||||
assert.Equal(t, 0, len(a.Endpoint))
|
||||
@@ -79,38 +79,38 @@ func TestNewActionNotification(t *testing.T) {
|
||||
|
||||
user.FsConfig.Provider = sdk.S3FilesystemProvider
|
||||
a = newActionNotification(&user, operationDownload, "path", "vpath", "target", "", "", ProtocolSSH, "", sessionID,
|
||||
123, 0, c.getNotificationStatus(nil))
|
||||
123, 0, c.getNotificationStatus(nil), 0)
|
||||
assert.Equal(t, "s3bucket", a.Bucket)
|
||||
assert.Equal(t, "endpoint", a.Endpoint)
|
||||
assert.Equal(t, 1, a.Status)
|
||||
|
||||
user.FsConfig.Provider = sdk.GCSFilesystemProvider
|
||||
a = newActionNotification(&user, operationDownload, "path", "vpath", "target", "", "", ProtocolSCP, "", sessionID,
|
||||
123, 0, c.getNotificationStatus(ErrQuotaExceeded))
|
||||
123, 0, c.getNotificationStatus(ErrQuotaExceeded), 0)
|
||||
assert.Equal(t, "gcsbucket", a.Bucket)
|
||||
assert.Equal(t, 0, len(a.Endpoint))
|
||||
assert.Equal(t, 3, a.Status)
|
||||
a = newActionNotification(&user, operationDownload, "path", "vpath", "target", "", "", ProtocolSCP, "", sessionID,
|
||||
123, 0, c.getNotificationStatus(fmt.Errorf("wrapper quota error: %w", ErrQuotaExceeded)))
|
||||
123, 0, c.getNotificationStatus(fmt.Errorf("wrapper quota error: %w", ErrQuotaExceeded)), 0)
|
||||
assert.Equal(t, "gcsbucket", a.Bucket)
|
||||
assert.Equal(t, 0, len(a.Endpoint))
|
||||
assert.Equal(t, 3, a.Status)
|
||||
|
||||
user.FsConfig.Provider = sdk.HTTPFilesystemProvider
|
||||
a = newActionNotification(&user, operationDownload, "path", "vpath", "target", "", "", ProtocolSSH, "", sessionID,
|
||||
123, 0, c.getNotificationStatus(nil))
|
||||
123, 0, c.getNotificationStatus(nil), 0)
|
||||
assert.Equal(t, "httpendpoint", a.Endpoint)
|
||||
assert.Equal(t, 1, a.Status)
|
||||
|
||||
user.FsConfig.Provider = sdk.AzureBlobFilesystemProvider
|
||||
a = newActionNotification(&user, operationDownload, "path", "vpath", "target", "", "", ProtocolSCP, "", sessionID,
|
||||
123, 0, c.getNotificationStatus(nil))
|
||||
123, 0, c.getNotificationStatus(nil), 0)
|
||||
assert.Equal(t, "azcontainer", a.Bucket)
|
||||
assert.Equal(t, "azendpoint", a.Endpoint)
|
||||
assert.Equal(t, 1, a.Status)
|
||||
|
||||
a = newActionNotification(&user, operationDownload, "path", "vpath", "target", "", "", ProtocolSCP, "", sessionID,
|
||||
123, os.O_APPEND, c.getNotificationStatus(nil))
|
||||
123, os.O_APPEND, c.getNotificationStatus(nil), 0)
|
||||
assert.Equal(t, "azcontainer", a.Bucket)
|
||||
assert.Equal(t, "azendpoint", a.Endpoint)
|
||||
assert.Equal(t, 1, a.Status)
|
||||
@@ -118,7 +118,7 @@ func TestNewActionNotification(t *testing.T) {
|
||||
|
||||
user.FsConfig.Provider = sdk.SFTPFilesystemProvider
|
||||
a = newActionNotification(&user, operationDownload, "path", "vpath", "target", "", "", ProtocolSFTP, "", sessionID,
|
||||
123, 0, c.getNotificationStatus(nil))
|
||||
123, 0, c.getNotificationStatus(nil), 0)
|
||||
assert.Equal(t, "sftpendpoint", a.Endpoint)
|
||||
}
|
||||
|
||||
@@ -135,7 +135,7 @@ func TestActionHTTP(t *testing.T) {
|
||||
},
|
||||
}
|
||||
a := newActionNotification(user, operationDownload, "path", "vpath", "target", "", "", ProtocolSFTP, "",
|
||||
xid.New().String(), 123, 0, 1)
|
||||
xid.New().String(), 123, 0, 1, 0)
|
||||
status, err := actionHandler.Handle(a)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, 1, status)
|
||||
@@ -175,16 +175,16 @@ func TestActionCMD(t *testing.T) {
|
||||
}
|
||||
sessionID := shortuuid.New()
|
||||
a := newActionNotification(user, operationDownload, "path", "vpath", "target", "", "", ProtocolSFTP, "", sessionID,
|
||||
123, 0, 1)
|
||||
123, 0, 1, 0)
|
||||
status, err := actionHandler.Handle(a)
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, 1, status)
|
||||
|
||||
c := NewBaseConnection("id", ProtocolSFTP, "", "", *user)
|
||||
err = ExecuteActionNotification(c, OperationSSHCmd, "path", "vpath", "target", "vtarget", "sha1sum", 0, nil)
|
||||
err = ExecuteActionNotification(c, OperationSSHCmd, "path", "vpath", "target", "vtarget", "sha1sum", 0, nil, 0)
|
||||
assert.NoError(t, err)
|
||||
|
||||
err = ExecuteActionNotification(c, operationDownload, "path", "vpath", "", "", "", 0, nil)
|
||||
err = ExecuteActionNotification(c, operationDownload, "path", "vpath", "", "", "", 0, nil, 0)
|
||||
assert.NoError(t, err)
|
||||
|
||||
Config.Actions = actionsCopy
|
||||
@@ -208,7 +208,7 @@ func TestWrongActions(t *testing.T) {
|
||||
}
|
||||
|
||||
a := newActionNotification(user, operationUpload, "", "", "", "", "", ProtocolSFTP, "", xid.New().String(),
|
||||
123, 0, 1)
|
||||
123, 0, 1, 0)
|
||||
status, err := actionHandler.Handle(a)
|
||||
assert.Error(t, err, "action with bad command must fail")
|
||||
assert.Equal(t, 1, status)
|
||||
@@ -307,7 +307,7 @@ func TestUnconfiguredHook(t *testing.T) {
|
||||
assert.NoError(t, err)
|
||||
assert.Equal(t, status, 0)
|
||||
|
||||
err = ExecuteActionNotification(c, operationDownload, "", "", "", "", "", 0, nil)
|
||||
err = ExecuteActionNotification(c, operationDownload, "", "", "", "", "", 0, nil, 0)
|
||||
assert.NoError(t, err)
|
||||
|
||||
err = plugin.Initialize(nil, "debug")
|
||||
|
||||
@@ -360,15 +360,17 @@ func (c *BaseConnection) CreateDir(virtualPath string, checkFilePatterns bool) e
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
startTime := time.Now()
|
||||
if err := fs.Mkdir(fsPath); err != nil {
|
||||
c.Log(logger.LevelError, "error creating dir: %#v error: %+v", fsPath, err)
|
||||
return c.GetFsError(fs, err)
|
||||
}
|
||||
vfs.SetPathPermissions(fs, fsPath, c.User.GetUID(), c.User.GetGID())
|
||||
elapsed := time.Since(startTime).Nanoseconds() / 1000000
|
||||
|
||||
logger.CommandLog(mkdirLogSender, fsPath, "", c.User.Username, "", c.ID, c.protocol, -1, -1, "", "", "", -1,
|
||||
c.localAddr, c.remoteAddr)
|
||||
ExecuteActionNotification(c, operationMkdir, fsPath, virtualPath, "", "", "", 0, nil) //nolint:errcheck
|
||||
c.localAddr, c.remoteAddr, elapsed)
|
||||
ExecuteActionNotification(c, operationMkdir, fsPath, virtualPath, "", "", "", 0, nil, elapsed) //nolint:errcheck
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -397,6 +399,7 @@ func (c *BaseConnection) RemoveFile(fs vfs.Fs, fsPath, virtualPath string, info
|
||||
return c.GetPermissionDeniedError()
|
||||
}
|
||||
updateQuota := true
|
||||
startTime := time.Now()
|
||||
if err := fs.Remove(fsPath, false); err != nil {
|
||||
if status > 0 && fs.IsNotExist(err) {
|
||||
// file removed in the pre-action, if the file was deleted from the EventManager the quota is already updated
|
||||
@@ -407,9 +410,10 @@ func (c *BaseConnection) RemoveFile(fs vfs.Fs, fsPath, virtualPath string, info
|
||||
return c.GetFsError(fs, err)
|
||||
}
|
||||
}
|
||||
elapsed := time.Since(startTime).Nanoseconds() / 1000000
|
||||
|
||||
logger.CommandLog(removeLogSender, fsPath, "", c.User.Username, "", c.ID, c.protocol, -1, -1, "", "", "", -1,
|
||||
c.localAddr, c.remoteAddr)
|
||||
c.localAddr, c.remoteAddr, elapsed)
|
||||
if updateQuota && info.Mode()&os.ModeSymlink == 0 {
|
||||
vfolder, err := c.User.GetVirtualFolderForPath(path.Dir(virtualPath))
|
||||
if err == nil {
|
||||
@@ -421,7 +425,7 @@ func (c *BaseConnection) RemoveFile(fs vfs.Fs, fsPath, virtualPath string, info
|
||||
dataprovider.UpdateUserQuota(&c.User, -1, -size, false) //nolint:errcheck
|
||||
}
|
||||
}
|
||||
ExecuteActionNotification(c, operationDelete, fsPath, virtualPath, "", "", "", size, nil) //nolint:errcheck
|
||||
ExecuteActionNotification(c, operationDelete, fsPath, virtualPath, "", "", "", size, nil, elapsed) //nolint:errcheck
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -478,14 +482,16 @@ func (c *BaseConnection) RemoveDir(virtualPath string) error {
|
||||
return c.GetGenericError(nil)
|
||||
}
|
||||
|
||||
startTime := time.Now()
|
||||
if err := fs.Remove(fsPath, true); err != nil {
|
||||
c.Log(logger.LevelError, "failed to remove directory %#v: %+v", fsPath, err)
|
||||
return c.GetFsError(fs, err)
|
||||
}
|
||||
elapsed := time.Since(startTime).Nanoseconds() / 1000000
|
||||
|
||||
logger.CommandLog(rmdirLogSender, fsPath, "", c.User.Username, "", c.ID, c.protocol, -1, -1, "", "", "", -1,
|
||||
c.localAddr, c.remoteAddr)
|
||||
ExecuteActionNotification(c, operationRmdir, fsPath, virtualPath, "", "", "", 0, nil) //nolint:errcheck
|
||||
c.localAddr, c.remoteAddr, elapsed)
|
||||
ExecuteActionNotification(c, operationRmdir, fsPath, virtualPath, "", "", "", 0, nil, elapsed) //nolint:errcheck
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -597,8 +603,10 @@ func (c *BaseConnection) copyFile(virtualSourcePath, virtualTargetPath string, s
|
||||
}
|
||||
defer wCancelFn()
|
||||
|
||||
startTime := time.Now()
|
||||
_, err = io.Copy(writer, reader)
|
||||
return closeWriterAndUpdateQuota(writer, c, virtualSourcePath, virtualTargetPath, numFiles, truncatedSize, err, operationCopy)
|
||||
return closeWriterAndUpdateQuota(writer, c, virtualSourcePath, virtualTargetPath, numFiles, truncatedSize,
|
||||
err, operationCopy, startTime)
|
||||
}
|
||||
|
||||
func (c *BaseConnection) doRecursiveCopy(virtualSourcePath, virtualTargetPath string, srcInfo os.FileInfo,
|
||||
@@ -709,6 +717,7 @@ func (c *BaseConnection) renameInternal(virtualSourcePath, virtualTargetPath str
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
startTime := time.Now()
|
||||
srcInfo, err := fsSrc.Lstat(fsSourcePath)
|
||||
if err != nil {
|
||||
return c.GetFsError(fsSrc, err)
|
||||
@@ -756,11 +765,12 @@ func (c *BaseConnection) renameInternal(virtualSourcePath, virtualTargetPath str
|
||||
return c.GetFsError(fsSrc, err)
|
||||
}
|
||||
vfs.SetPathPermissions(fsDst, fsTargetPath, c.User.GetUID(), c.User.GetGID())
|
||||
elapsed := time.Since(startTime).Nanoseconds() / 1000000
|
||||
c.updateQuotaAfterRename(fsDst, virtualSourcePath, virtualTargetPath, fsTargetPath, initialSize, files, size) //nolint:errcheck
|
||||
logger.CommandLog(renameLogSender, fsSourcePath, fsTargetPath, c.User.Username, "", c.ID, c.protocol, -1, -1,
|
||||
"", "", "", -1, c.localAddr, c.remoteAddr)
|
||||
"", "", "", -1, c.localAddr, c.remoteAddr, elapsed)
|
||||
ExecuteActionNotification(c, operationRename, fsSourcePath, virtualSourcePath, fsTargetPath, //nolint:errcheck
|
||||
virtualTargetPath, "", 0, nil)
|
||||
virtualTargetPath, "", 0, nil, elapsed)
|
||||
|
||||
return nil
|
||||
}
|
||||
@@ -810,12 +820,14 @@ func (c *BaseConnection) CreateSymlink(virtualSourcePath, virtualTargetPath stri
|
||||
if relativePath != "" {
|
||||
fsSourcePath = relativePath
|
||||
}
|
||||
startTime := time.Now()
|
||||
if err := fs.Symlink(fsSourcePath, fsTargetPath); err != nil {
|
||||
c.Log(logger.LevelError, "failed to create symlink %#v -> %#v: %+v", fsSourcePath, fsTargetPath, err)
|
||||
c.Log(logger.LevelError, "failed to create symlink %q -> %q: %+v", fsSourcePath, fsTargetPath, err)
|
||||
return c.GetFsError(fs, err)
|
||||
}
|
||||
elapsed := time.Since(startTime).Nanoseconds() / 1000000
|
||||
logger.CommandLog(symlinkLogSender, fsSourcePath, fsTargetPath, c.User.Username, "", c.ID, c.protocol, -1, -1, "",
|
||||
"", "", -1, c.localAddr, c.remoteAddr)
|
||||
"", "", -1, c.localAddr, c.remoteAddr, elapsed)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -899,12 +911,14 @@ func (c *BaseConnection) handleChmod(fs vfs.Fs, fsPath, pathForPerms string, att
|
||||
if c.ignoreSetStat(fs) {
|
||||
return nil
|
||||
}
|
||||
startTime := time.Now()
|
||||
if err := fs.Chmod(c.getRealFsPath(fsPath), attributes.Mode); err != nil {
|
||||
c.Log(logger.LevelError, "failed to chmod path %#v, mode: %v, err: %+v", fsPath, attributes.Mode.String(), err)
|
||||
return c.GetFsError(fs, err)
|
||||
}
|
||||
elapsed := time.Since(startTime).Nanoseconds() / 1000000
|
||||
logger.CommandLog(chmodLogSender, fsPath, "", c.User.Username, attributes.Mode.String(), c.ID, c.protocol,
|
||||
-1, -1, "", "", "", -1, c.localAddr, c.remoteAddr)
|
||||
-1, -1, "", "", "", -1, c.localAddr, c.remoteAddr, elapsed)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -915,13 +929,15 @@ func (c *BaseConnection) handleChown(fs vfs.Fs, fsPath, pathForPerms string, att
|
||||
if c.ignoreSetStat(fs) {
|
||||
return nil
|
||||
}
|
||||
startTime := time.Now()
|
||||
if err := fs.Chown(c.getRealFsPath(fsPath), attributes.UID, attributes.GID); err != nil {
|
||||
c.Log(logger.LevelError, "failed to chown path %#v, uid: %v, gid: %v, err: %+v", fsPath, attributes.UID,
|
||||
attributes.GID, err)
|
||||
return c.GetFsError(fs, err)
|
||||
}
|
||||
elapsed := time.Since(startTime).Nanoseconds() / 1000000
|
||||
logger.CommandLog(chownLogSender, fsPath, "", c.User.Username, "", c.ID, c.protocol, attributes.UID, attributes.GID,
|
||||
"", "", "", -1, c.localAddr, c.remoteAddr)
|
||||
"", "", "", -1, c.localAddr, c.remoteAddr, elapsed)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -932,6 +948,7 @@ func (c *BaseConnection) handleChtimes(fs vfs.Fs, fsPath, pathForPerms string, a
|
||||
if Config.SetstatMode == 1 {
|
||||
return nil
|
||||
}
|
||||
startTime := time.Now()
|
||||
isUploading := c.setTimes(fsPath, attributes.Atime, attributes.Mtime)
|
||||
if err := fs.Chtimes(c.getRealFsPath(fsPath), attributes.Atime, attributes.Mtime, isUploading); err != nil {
|
||||
c.setTimes(fsPath, time.Time{}, time.Time{})
|
||||
@@ -942,10 +959,11 @@ func (c *BaseConnection) handleChtimes(fs vfs.Fs, fsPath, pathForPerms string, a
|
||||
fsPath, attributes.Atime, attributes.Mtime, err)
|
||||
return c.GetFsError(fs, err)
|
||||
}
|
||||
elapsed := time.Since(startTime).Nanoseconds() / 1000000
|
||||
accessTimeString := attributes.Atime.Format(chtimesFormat)
|
||||
modificationTimeString := attributes.Mtime.Format(chtimesFormat)
|
||||
logger.CommandLog(chtimesLogSender, fsPath, "", c.User.Username, "", c.ID, c.protocol, -1, -1,
|
||||
accessTimeString, modificationTimeString, "", -1, c.localAddr, c.remoteAddr)
|
||||
accessTimeString, modificationTimeString, "", -1, c.localAddr, c.remoteAddr, elapsed)
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -982,13 +1000,14 @@ func (c *BaseConnection) SetStat(virtualPath string, attributes *StatAttributes)
|
||||
if !c.User.HasPerm(dataprovider.PermOverwrite, pathForPerms) {
|
||||
return c.GetPermissionDeniedError()
|
||||
}
|
||||
|
||||
startTime := time.Now()
|
||||
if err = c.truncateFile(fs, fsPath, virtualPath, attributes.Size); err != nil {
|
||||
c.Log(logger.LevelError, "failed to truncate path %#v, size: %v, err: %+v", fsPath, attributes.Size, err)
|
||||
return c.GetFsError(fs, err)
|
||||
}
|
||||
elapsed := time.Since(startTime).Nanoseconds() / 1000000
|
||||
logger.CommandLog(truncateLogSender, fsPath, "", c.User.Username, "", c.ID, c.protocol, -1, -1, "", "",
|
||||
"", attributes.Size, c.localAddr, c.remoteAddr)
|
||||
"", attributes.Size, c.localAddr, c.remoteAddr, elapsed)
|
||||
}
|
||||
|
||||
return nil
|
||||
|
||||
@@ -447,6 +447,7 @@ type EventParams struct {
|
||||
ObjectName string
|
||||
ObjectType string
|
||||
FileSize int64
|
||||
Elapsed int64
|
||||
Protocol string
|
||||
IP string
|
||||
Role string
|
||||
@@ -633,6 +634,7 @@ func (p *EventParams) getStringReplacements(addObjectData bool) []string {
|
||||
"{{ObjectName}}", p.ObjectName,
|
||||
"{{ObjectType}}", p.ObjectType,
|
||||
"{{FileSize}}", fmt.Sprintf("%d", p.FileSize),
|
||||
"{{Elapsed}}", fmt.Sprintf("%d", p.Elapsed),
|
||||
"{{Protocol}}", p.Protocol,
|
||||
"{{IP}}", p.IP,
|
||||
"{{Role}}", p.Role,
|
||||
@@ -685,7 +687,7 @@ func getCSVRetentionReport(results []folderRetentionCheckResult) ([]byte, error)
|
||||
}
|
||||
|
||||
func closeWriterAndUpdateQuota(w io.WriteCloser, conn *BaseConnection, virtualSourcePath, virtualTargetPath string,
|
||||
numFiles int, truncatedSize int64, errTransfer error, operation string,
|
||||
numFiles int, truncatedSize int64, errTransfer error, operation string, startTime time.Time,
|
||||
) error {
|
||||
errWrite := w.Close()
|
||||
targetPath := virtualSourcePath
|
||||
@@ -704,14 +706,15 @@ func closeWriterAndUpdateQuota(w io.WriteCloser, conn *BaseConnection, virtualSo
|
||||
_, fsDstPath, errDstFs = conn.GetFsAndResolvedPath(virtualTargetPath)
|
||||
}
|
||||
if errSrcFs == nil && errDstFs == nil {
|
||||
elapsed := time.Since(startTime).Nanoseconds() / 1000000
|
||||
if errTransfer == nil {
|
||||
errTransfer = errWrite
|
||||
}
|
||||
if operation == operationCopy {
|
||||
logger.CommandLog(copyLogSender, fsSrcPath, fsDstPath, conn.User.Username, "", conn.ID, conn.protocol, -1, -1,
|
||||
"", "", "", info.Size(), conn.localAddr, conn.remoteAddr)
|
||||
"", "", "", info.Size(), conn.localAddr, conn.remoteAddr, elapsed)
|
||||
}
|
||||
ExecuteActionNotification(conn, operation, fsSrcPath, virtualSourcePath, fsDstPath, virtualTargetPath, "", info.Size(), errTransfer) //nolint:errcheck
|
||||
ExecuteActionNotification(conn, operation, fsSrcPath, virtualSourcePath, fsDstPath, virtualTargetPath, "", info.Size(), errTransfer, elapsed) //nolint:errcheck
|
||||
}
|
||||
} else {
|
||||
eventManagerLog(logger.LevelWarn, "unable to update quota after writing %q: %v", targetPath, err)
|
||||
@@ -1735,18 +1738,19 @@ func executeCompressFsActionForUser(c dataprovider.EventActionFsCompress, replac
|
||||
Writer: zip.NewWriter(writer),
|
||||
Entries: make(map[string]bool),
|
||||
}
|
||||
startTime := time.Now()
|
||||
for _, item := range paths {
|
||||
if err := addZipEntry(zipWriter, conn, item, baseDir); err != nil {
|
||||
closeWriterAndUpdateQuota(writer, conn, name, "", numFiles, truncatedSize, err, operationUpload) //nolint:errcheck
|
||||
closeWriterAndUpdateQuota(writer, conn, name, "", numFiles, truncatedSize, err, operationUpload, startTime) //nolint:errcheck
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := zipWriter.Writer.Close(); err != nil {
|
||||
eventManagerLog(logger.LevelError, "unable to close zip file %q: %v", name, err)
|
||||
closeWriterAndUpdateQuota(writer, conn, name, "", numFiles, truncatedSize, err, operationUpload) //nolint:errcheck
|
||||
closeWriterAndUpdateQuota(writer, conn, name, "", numFiles, truncatedSize, err, operationUpload, startTime) //nolint:errcheck
|
||||
return fmt.Errorf("unable to close zip file %q: %w", name, err)
|
||||
}
|
||||
return closeWriterAndUpdateQuota(writer, conn, name, "", numFiles, truncatedSize, err, operationUpload)
|
||||
return closeWriterAndUpdateQuota(writer, conn, name, "", numFiles, truncatedSize, err, operationUpload, startTime)
|
||||
}
|
||||
|
||||
func executeExistFsRuleAction(exist []string, replacer *strings.Replacer, conditions dataprovider.ConditionOptions,
|
||||
|
||||
@@ -393,7 +393,7 @@ func (t *BaseTransfer) Close() error {
|
||||
logger.TransferLog(downloadLogSender, t.fsPath, elapsed, t.BytesSent.Load(), t.Connection.User.Username,
|
||||
t.Connection.ID, t.Connection.protocol, t.Connection.localAddr, t.Connection.remoteAddr, t.ftpMode)
|
||||
ExecuteActionNotification(t.Connection, operationDownload, t.fsPath, t.requestPath, "", "", "", //nolint:errcheck
|
||||
t.BytesSent.Load(), t.ErrTransfer)
|
||||
t.BytesSent.Load(), t.ErrTransfer, elapsed)
|
||||
} else {
|
||||
statSize, deletedFiles, errStat := t.getUploadFileSize()
|
||||
if errStat == nil {
|
||||
@@ -408,7 +408,7 @@ func (t *BaseTransfer) Close() error {
|
||||
numFiles -= deletedFiles
|
||||
t.Connection.Log(logger.LevelDebug, "upload file size %d, num files %d, deleted files %d, fs path %q",
|
||||
uploadFileSize, numFiles, deletedFiles, t.fsPath)
|
||||
numFiles, uploadFileSize = t.executeUploadHook(numFiles, uploadFileSize)
|
||||
numFiles, uploadFileSize = t.executeUploadHook(numFiles, uploadFileSize, elapsed)
|
||||
t.updateQuota(numFiles, uploadFileSize)
|
||||
t.updateTimes()
|
||||
logger.TransferLog(uploadLogSender, t.fsPath, elapsed, t.BytesReceived.Load(), t.Connection.User.Username,
|
||||
@@ -420,11 +420,11 @@ func (t *BaseTransfer) Close() error {
|
||||
err = t.ErrTransfer
|
||||
}
|
||||
}
|
||||
t.updateTransferTimestamps(uploadFileSize)
|
||||
t.updateTransferTimestamps(uploadFileSize, elapsed)
|
||||
return err
|
||||
}
|
||||
|
||||
func (t *BaseTransfer) updateTransferTimestamps(uploadFileSize int64) {
|
||||
func (t *BaseTransfer) updateTransferTimestamps(uploadFileSize, elapsed int64) {
|
||||
if t.ErrTransfer != nil {
|
||||
return
|
||||
}
|
||||
@@ -433,7 +433,7 @@ func (t *BaseTransfer) updateTransferTimestamps(uploadFileSize int64) {
|
||||
if err := dataprovider.UpdateUserTransferTimestamps(t.Connection.User.Username, true); err == nil {
|
||||
t.Connection.uploadDone.Store(true)
|
||||
ExecuteActionNotification(t.Connection, operationFirstUpload, t.fsPath, t.requestPath, "", //nolint:errcheck
|
||||
"", "", uploadFileSize, t.ErrTransfer)
|
||||
"", "", uploadFileSize, t.ErrTransfer, elapsed)
|
||||
}
|
||||
}
|
||||
return
|
||||
@@ -442,14 +442,14 @@ func (t *BaseTransfer) updateTransferTimestamps(uploadFileSize int64) {
|
||||
if err := dataprovider.UpdateUserTransferTimestamps(t.Connection.User.Username, false); err == nil {
|
||||
t.Connection.downloadDone.Store(true)
|
||||
ExecuteActionNotification(t.Connection, operationFirstDownload, t.fsPath, t.requestPath, "", //nolint:errcheck
|
||||
"", "", t.BytesSent.Load(), t.ErrTransfer)
|
||||
"", "", t.BytesSent.Load(), t.ErrTransfer, elapsed)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (t *BaseTransfer) executeUploadHook(numFiles int, fileSize int64) (int, int64) {
|
||||
func (t *BaseTransfer) executeUploadHook(numFiles int, fileSize, elapsed int64) (int, int64) {
|
||||
err := ExecuteActionNotification(t.Connection, operationUpload, t.fsPath, t.requestPath, "", "", "",
|
||||
fileSize, t.ErrTransfer)
|
||||
fileSize, t.ErrTransfer, elapsed)
|
||||
if err != nil {
|
||||
if t.ErrTransfer == nil {
|
||||
t.ErrTransfer = err
|
||||
|
||||
Reference in New Issue
Block a user