Switch to SQLite for activity tracking
This commit is contained in:
DaneEveritt
@@ -1,102 +1,102 @@
|
||||
package cron
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"database/sql"
|
||||
"emperror.dev/errors"
|
||||
"github.com/apex/log"
|
||||
"github.com/goccy/go-json"
|
||||
"github.com/pterodactyl/wings/internal/database"
|
||||
"encoding/gob"
|
||||
"github.com/pterodactyl/wings/internal/sqlite"
|
||||
"github.com/pterodactyl/wings/server"
|
||||
"github.com/pterodactyl/wings/system"
|
||||
"github.com/xujiajun/nutsdb"
|
||||
"strings"
|
||||
)
|
||||
|
||||
var key = []byte("events")
|
||||
var activityCron system.AtomicBool
|
||||
type activityCron struct {
|
||||
mu *system.AtomicBool
|
||||
manager *server.Manager
|
||||
max int64
|
||||
}
|
||||
|
||||
func processActivityLogs(m *server.Manager, c int64) error {
|
||||
const queryRegularActivity = `
|
||||
SELECT id, event, user_uuid, server_uuid, metadata, ip, timestamp FROM activity_logs
|
||||
WHERE event NOT LIKE 'server:sftp.%'
|
||||
ORDER BY timestamp
|
||||
LIMIT ?
|
||||
`
|
||||
|
||||
type QueriedActivity struct {
|
||||
id int
|
||||
b []byte
|
||||
server.Activity
|
||||
}
|
||||
|
||||
// Parse parses the internal query results into the QueriedActivity type and then properly
|
||||
// sets the Metadata onto it. This also sets the ID that was returned to ensure we're able
|
||||
// to then delete all of the matching rows in the database after we're done.
|
||||
func (qa *QueriedActivity) Parse(r *sql.Rows) error {
|
||||
if err := r.Scan(&qa.id, &qa.Event, &qa.User, &qa.Server, &qa.b, &qa.IP, &qa.Timestamp); err != nil {
|
||||
return errors.Wrap(err, "cron: failed to parse activity log")
|
||||
}
|
||||
if err := gob.NewDecoder(bytes.NewBuffer(qa.b)).Decode(&qa.Metadata); err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Run executes the cronjob and ensures we fetch and send all of the stored activity to the
|
||||
// Panel instance. Once activity is sent it is deleted from the local database instance. Any
|
||||
// SFTP specific events are not handled in this cron, they're handled seperately to account
|
||||
// for de-duplication and event merging.
|
||||
func (ac *activityCron) Run(ctx context.Context) error {
|
||||
// Don't execute this cron if there is currently one running. Once this task is completed
|
||||
// go ahead and mark it as no longer running.
|
||||
if !activityCron.SwapIf(true) {
|
||||
log.WithField("subsystem", "cron").WithField("cron", "activity_logs").Warn("cron: process overlap detected, skipping this run")
|
||||
if !ac.mu.SwapIf(true) {
|
||||
return errors.WithStack(ErrCronRunning)
|
||||
}
|
||||
defer ac.mu.Store(false)
|
||||
|
||||
rows, err := sqlite.Instance().QueryContext(ctx, queryRegularActivity, ac.max)
|
||||
if err != nil {
|
||||
return errors.Wrap(err, "cron: failed to query activity logs")
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
var logs []server.Activity
|
||||
var ids []int
|
||||
for rows.Next() {
|
||||
var qa QueriedActivity
|
||||
if err := qa.Parse(rows); err != nil {
|
||||
return err
|
||||
}
|
||||
ids = append(ids, qa.id)
|
||||
logs = append(logs, qa.Activity)
|
||||
}
|
||||
|
||||
if err := rows.Err(); err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
if len(logs) == 0 {
|
||||
return nil
|
||||
}
|
||||
defer activityCron.Store(false)
|
||||
|
||||
var list [][]byte
|
||||
err := database.DB().View(func(tx *nutsdb.Tx) error {
|
||||
// Grab the oldest 100 activity events that have been logged and send them back to the
|
||||
// Panel for processing. Once completed, delete those events from the database and then
|
||||
// release the lock on this process.
|
||||
end := int(c)
|
||||
if s, err := tx.LSize(database.ServerActivityBucket, key); err != nil {
|
||||
if errors.Is(err, nutsdb.ErrBucket) {
|
||||
return nil
|
||||
}
|
||||
return errors.WithStackIf(err)
|
||||
} else if s < end || s == 0 {
|
||||
if s == 0 {
|
||||
return nil
|
||||
}
|
||||
end = s
|
||||
}
|
||||
l, err := tx.LRange(database.ServerActivityBucket, key, 0, end)
|
||||
if err != nil {
|
||||
// This error is returned when the bucket doesn't exist, which is likely on the
|
||||
// first invocations of Wings since we haven't yet logged any data. There is nothing
|
||||
// that needs to be done if this error occurs.
|
||||
if errors.Is(err, nutsdb.ErrBucket) {
|
||||
return nil
|
||||
}
|
||||
return errors.WithStackIf(err)
|
||||
}
|
||||
list = l
|
||||
return nil
|
||||
})
|
||||
|
||||
if err != nil || len(list) == 0 {
|
||||
return errors.WithStackIf(err)
|
||||
}
|
||||
|
||||
var processed []json.RawMessage
|
||||
for _, l := range list {
|
||||
var v json.RawMessage
|
||||
if err := json.Unmarshal(l, &v); err != nil {
|
||||
log.WithField("error", errors.WithStack(err)).Warn("failed to parse activity event json, skipping entry")
|
||||
continue
|
||||
}
|
||||
processed = append(processed, v)
|
||||
}
|
||||
|
||||
if err := m.Client().SendActivityLogs(context.Background(), processed); err != nil {
|
||||
if err := ac.manager.Client().SendActivityLogs(context.Background(), logs); err != nil {
|
||||
return errors.WrapIf(err, "cron: failed to send activity events to Panel")
|
||||
}
|
||||
|
||||
return database.DB().Update(func(tx *nutsdb.Tx) error {
|
||||
if m, err := tx.LSize(database.ServerActivityBucket, key); err != nil {
|
||||
return errors.WithStack(err)
|
||||
} else if m > len(list) {
|
||||
// As long as there are more elements than we have in the length of our list
|
||||
// we can just use the existing `LTrim` functionality of nutsdb. This will remove
|
||||
// all of the values we've already pulled and sent to the API.
|
||||
return errors.WithStack(tx.LTrim(database.ServerActivityBucket, key, len(list), -1))
|
||||
} else {
|
||||
i := 0
|
||||
// This is the only way I can figure out to actually empty the items out of the list
|
||||
// because you cannot use `LTrim` (or I cannot for the life of me figure out how) to
|
||||
// trim the slice down to 0 items without it triggering an internal logic error. Perhaps
|
||||
// in a future release they'll have a function to do this (based on my skimming of issues
|
||||
// on GitHub that I cannot read due to translation barriers).
|
||||
for {
|
||||
if i >= m {
|
||||
break
|
||||
}
|
||||
if _, err := tx.LPop(database.ServerActivityBucket, key); err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
i++
|
||||
}
|
||||
if tx, err := sqlite.Instance().Begin(); err != nil {
|
||||
return err
|
||||
} else {
|
||||
t := make([]string, len(ids))
|
||||
params := make([]interface{}, len(ids))
|
||||
for i := 0; i < len(ids); i++ {
|
||||
t[i] = "?"
|
||||
params[i] = ids[i]
|
||||
}
|
||||
return nil
|
||||
})
|
||||
q := strings.Join(t, ",")
|
||||
_, err := tx.Exec(`DELETE FROM activity_logs WHERE id IN(`+q+`)`, params...)
|
||||
if err != nil {
|
||||
return errors.Combine(errors.WithStack(err), tx.Rollback())
|
||||
}
|
||||
return errors.WithStack(tx.Commit())
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package cron
|
||||
|
||||
import (
|
||||
"context"
|
||||
"emperror.dev/errors"
|
||||
"github.com/apex/log"
|
||||
"github.com/go-co-op/gocron"
|
||||
@@ -17,7 +18,7 @@ var o system.AtomicBool
|
||||
// Scheduler configures the internal cronjob system for Wings and returns the scheduler
|
||||
// instance to the caller. This should only be called once per application lifecycle, additional
|
||||
// calls will result in an error being returned.
|
||||
func Scheduler(m *server.Manager) (*gocron.Scheduler, error) {
|
||||
func Scheduler(ctx context.Context, m *server.Manager) (*gocron.Scheduler, error) {
|
||||
if !o.SwapIf(true) {
|
||||
return nil, errors.New("cron: cannot call scheduler more than once in application lifecycle")
|
||||
}
|
||||
@@ -26,20 +27,20 @@ func Scheduler(m *server.Manager) (*gocron.Scheduler, error) {
|
||||
return nil, errors.Wrap(err, "cron: failed to parse configured system timezone")
|
||||
}
|
||||
|
||||
s := gocron.NewScheduler(l)
|
||||
_, _ = s.Tag("activity").Every(int(config.Get().System.ActivitySendInterval)).Seconds().Do(func() {
|
||||
if err := processActivityLogs(m, config.Get().System.ActivitySendCount); err != nil {
|
||||
log.WithField("error", err).Error("cron: failed to process activity events")
|
||||
}
|
||||
})
|
||||
activity := activityCron{
|
||||
mu: system.NewAtomicBool(false),
|
||||
manager: m,
|
||||
max: config.Get().System.ActivitySendCount,
|
||||
}
|
||||
|
||||
_, _ = s.Tag("sftp").Every(20).Seconds().Do(func() {
|
||||
runner := sftpEventProcessor{mu: system.NewAtomicBool(false), manager: m}
|
||||
if err := runner.Run(); err != nil {
|
||||
s := gocron.NewScheduler(l)
|
||||
// int(config.Get().System.ActivitySendInterval)
|
||||
_, _ = s.Tag("activity").Every(5).Seconds().Do(func() {
|
||||
if err := activity.Run(ctx); err != nil {
|
||||
if errors.Is(err, ErrCronRunning) {
|
||||
log.WithField("cron", "sftp_events").Warn("cron: job already running, skipping...")
|
||||
log.WithField("cron", "activity").Warn("cron: process is already running, skipping...")
|
||||
} else {
|
||||
log.WithField("error", err).Error("cron: failed to process sftp events")
|
||||
log.WithField("error", err).Error("cron: failed to process activity events")
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
@@ -1,188 +0,0 @@
|
||||
package cron
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"emperror.dev/errors"
|
||||
"encoding/gob"
|
||||
"github.com/pterodactyl/wings/internal/database"
|
||||
"github.com/pterodactyl/wings/server"
|
||||
"github.com/pterodactyl/wings/sftp"
|
||||
"github.com/pterodactyl/wings/system"
|
||||
"github.com/xujiajun/nutsdb"
|
||||
"path/filepath"
|
||||
)
|
||||
|
||||
type UserDetail struct {
|
||||
UUID string
|
||||
IP string
|
||||
}
|
||||
|
||||
type Users map[UserDetail][]sftp.EventRecord
|
||||
type Events map[sftp.Event]Users
|
||||
|
||||
type sftpEventProcessor struct {
|
||||
mu *system.AtomicBool
|
||||
manager *server.Manager
|
||||
}
|
||||
|
||||
// Run executes the cronjob and processes sftp activities into normal activity log entries
|
||||
// by merging together similar records. This helps to reduce the sheer amount of data that
|
||||
// gets passed back to the Panel and provides simpler activity logging.
|
||||
func (sep *sftpEventProcessor) Run() error {
|
||||
if !sep.mu.SwapIf(true) {
|
||||
return errors.WithStack(ErrCronRunning)
|
||||
}
|
||||
defer sep.mu.Store(false)
|
||||
|
||||
set, err := sep.Events()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for s, el := range set {
|
||||
events := make(Events)
|
||||
// Take all of the events that we've pulled out of the system for every server and then
|
||||
// parse them into a more usable format in order to create activity log entries for each
|
||||
// user, ip, and server instance.
|
||||
for _, e := range el {
|
||||
u := UserDetail{UUID: e.User, IP: e.IP}
|
||||
if _, ok := events[e.Event]; !ok {
|
||||
events[e.Event] = make(Users)
|
||||
}
|
||||
if _, ok := events[e.Event][u]; !ok {
|
||||
events[e.Event][u] = []sftp.EventRecord{}
|
||||
}
|
||||
events[e.Event][u] = append(events[e.Event][u], e)
|
||||
}
|
||||
|
||||
// Now that we have all of the events, go ahead and create a normal activity log entry
|
||||
// for each instance grouped by user & IP for easier Panel reporting.
|
||||
for k, v := range events {
|
||||
for u, records := range v {
|
||||
files := make([]interface{}, len(records))
|
||||
for i, r := range records {
|
||||
if r.Action.Target != "" {
|
||||
files[i] = map[string]string{
|
||||
"from": filepath.Clean(r.Action.Entity),
|
||||
"to": filepath.Clean(r.Action.Target),
|
||||
}
|
||||
} else {
|
||||
files[i] = filepath.Clean(r.Action.Entity)
|
||||
}
|
||||
}
|
||||
|
||||
entry := server.Activity{
|
||||
Server: s,
|
||||
User: u.UUID,
|
||||
Event: server.Event("server:sftp." + string(k)),
|
||||
Metadata: server.ActivityMeta{"files": files},
|
||||
IP: u.IP,
|
||||
// Just assume that the first record in the set is the oldest and the most relevant
|
||||
// of the timestamps to use.
|
||||
Timestamp: records[0].Timestamp,
|
||||
}
|
||||
|
||||
if err := entry.Save(); err != nil {
|
||||
return errors.Wrap(err, "cron: failed to save new event for server")
|
||||
}
|
||||
|
||||
if err := sep.Cleanup([]byte(s)); err != nil {
|
||||
return errors.Wrap(err, "cron: failed to cleanup events")
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Cleanup runs through all of the events we have currently tracked in the bucket and removes
|
||||
// them once we've managed to process them and created the associated server activity events.
|
||||
func (sep *sftpEventProcessor) Cleanup(key []byte) error {
|
||||
err := database.DB().Update(func(tx *nutsdb.Tx) error {
|
||||
s, err := sep.sizeOf(tx, key)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if s == 0 {
|
||||
return nil
|
||||
} else if s < sep.limit() {
|
||||
for i := 0; i < s; i++ {
|
||||
if _, err := tx.LPop(database.SftpActivityBucket, key); err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if err := tx.LTrim(database.ServerActivityBucket, key, sep.limit()-1, -1); err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
// Sometimes the key will end up not being found depending on the order of operations for
|
||||
// different events that are happening on the system. Make sure to account for that here,
|
||||
// if the key isn't found we can just safely assume it is a non issue and move on with our
|
||||
// day since there is nothing to clean up.
|
||||
if err != nil && errors.Is(err, nutsdb.ErrKeyNotFound) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
}
|
||||
|
||||
// Events pulls all of the events in the SFTP event bucket and parses them into an iterable
|
||||
// set allowing Wings to process the events and send them back to the Panel instance.
|
||||
func (sep *sftpEventProcessor) Events() (map[string][]sftp.EventRecord, error) {
|
||||
set := make(map[string][]sftp.EventRecord, len(sep.manager.Keys()))
|
||||
err := database.DB().View(func(tx *nutsdb.Tx) error {
|
||||
for _, k := range sep.manager.Keys() {
|
||||
lim := sep.limit()
|
||||
if s, err := sep.sizeOf(tx, []byte(k)); err != nil {
|
||||
// Not every server instance will have events tracked, so don't treat this
|
||||
// as a true error.
|
||||
if errors.Is(err, nutsdb.ErrKeyNotFound) {
|
||||
continue
|
||||
}
|
||||
return err
|
||||
} else if s == 0 {
|
||||
continue
|
||||
} else if s < lim {
|
||||
lim = -1
|
||||
}
|
||||
list, err := tx.LRange(database.SftpActivityBucket, []byte(k), 0, lim)
|
||||
if err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
set[k] = make([]sftp.EventRecord, len(list))
|
||||
for i, l := range list {
|
||||
if err := gob.NewDecoder(bytes.NewBuffer(l)).Decode(&set[k][i]); err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
return set, err
|
||||
}
|
||||
|
||||
// sizeOf is a wrapper around a nutsdb transaction to get the size of a key in the
|
||||
// bucket while also accounting for some expected error conditions and handling those
|
||||
// automatically.
|
||||
func (sep *sftpEventProcessor) sizeOf(tx *nutsdb.Tx, key []byte) (int, error) {
|
||||
s, err := tx.LSize(database.SftpActivityBucket, key)
|
||||
if err != nil {
|
||||
if errors.Is(err, nutsdb.ErrBucket) {
|
||||
return 0, nil
|
||||
}
|
||||
return 0, errors.WithStack(err)
|
||||
}
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// limit returns the number of records that are processed for each server at
|
||||
// once. This will then be translated into a variable number of activity log
|
||||
// events, with the worst case being a single event with "n" associated files.
|
||||
func (sep *sftpEventProcessor) limit() int {
|
||||
return 500
|
||||
}
|
||||
Reference in New Issue
Block a user