Add internal logic to process activity events and send them to the panel
This commit is contained in:
55
internal/cron/activity_cron.go
Normal file
55
internal/cron/activity_cron.go
Normal file
@@ -0,0 +1,55 @@
|
||||
package cron
|
||||
|
||||
import (
|
||||
"context"
|
||||
"emperror.dev/errors"
|
||||
"github.com/pterodactyl/wings/internal/database"
|
||||
"github.com/pterodactyl/wings/server"
|
||||
"github.com/pterodactyl/wings/system"
|
||||
"github.com/xujiajun/nutsdb"
|
||||
)
|
||||
|
||||
var processing system.AtomicBool
|
||||
|
||||
func processActivityLogs(m *server.Manager) 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 !processing.SwapIf(true) {
|
||||
return nil
|
||||
}
|
||||
defer processing.Store(false)
|
||||
|
||||
var b [][]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.
|
||||
list, err := tx.LRange(database.ServerActivityBucket, []byte("events"), 0, 1)
|
||||
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)
|
||||
}
|
||||
b = list
|
||||
return nil
|
||||
})
|
||||
|
||||
// If there is an error, return it. If there is no data to send to the Panel don't waste
|
||||
// an API call, just return here. WithStackIf will return "nil" when the value provided to
|
||||
// it is also nil.
|
||||
if err != nil || len(b) == 0 {
|
||||
return errors.WithStackIf(err)
|
||||
}
|
||||
|
||||
if err := m.Client().SendActivityLogs(context.Background(), b); err != nil {
|
||||
return errors.WrapIf(err, "cron: failed to send activity events to Panel")
|
||||
}
|
||||
|
||||
return database.DB().Update(func(tx *nutsdb.Tx) error {
|
||||
return tx.LTrim(database.ServerActivityBucket, []byte("events"), 2, -1)
|
||||
})
|
||||
}
|
||||
36
internal/cron/cron.go
Normal file
36
internal/cron/cron.go
Normal file
@@ -0,0 +1,36 @@
|
||||
package cron
|
||||
|
||||
import (
|
||||
"emperror.dev/errors"
|
||||
"github.com/apex/log"
|
||||
"github.com/go-co-op/gocron"
|
||||
"github.com/pterodactyl/wings/config"
|
||||
"github.com/pterodactyl/wings/server"
|
||||
"github.com/pterodactyl/wings/system"
|
||||
"time"
|
||||
)
|
||||
|
||||
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) {
|
||||
if o.Load() {
|
||||
return nil, errors.New("cron: cannot call scheduler more than once in application lifecycle")
|
||||
}
|
||||
o.Store(true)
|
||||
l, err := time.LoadLocation(config.Get().System.Timezone)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "cron: failed to parse configured system timezone")
|
||||
}
|
||||
|
||||
s := gocron.NewScheduler(l)
|
||||
_, _ = s.Every(5).Seconds().Do(func() {
|
||||
if err := processActivityLogs(m); err != nil {
|
||||
log.WithField("error", err).Error("cron: failed to process activity events")
|
||||
}
|
||||
})
|
||||
|
||||
return s, nil
|
||||
}
|
||||
38
internal/database/database.go
Normal file
38
internal/database/database.go
Normal file
@@ -0,0 +1,38 @@
|
||||
package database
|
||||
|
||||
import (
|
||||
"emperror.dev/errors"
|
||||
"github.com/apex/log"
|
||||
"github.com/pterodactyl/wings/config"
|
||||
"github.com/xujiajun/nutsdb"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
)
|
||||
|
||||
var db *nutsdb.DB
|
||||
var syncer sync.Once
|
||||
|
||||
const (
|
||||
ServerActivityBucket = "server_activity"
|
||||
)
|
||||
|
||||
func initialize() error {
|
||||
opt := nutsdb.DefaultOptions
|
||||
opt.Dir = filepath.Join(config.Get().System.RootDirectory, "db")
|
||||
|
||||
instance, err := nutsdb.Open(opt)
|
||||
if err != nil {
|
||||
return errors.WithStack(err)
|
||||
}
|
||||
db = instance
|
||||
return nil
|
||||
}
|
||||
|
||||
func DB() *nutsdb.DB {
|
||||
syncer.Do(func() {
|
||||
if err := initialize(); err != nil {
|
||||
log.WithField("error", err).Fatal("database: failed to initialize instance, this is an unrecoverable error")
|
||||
}
|
||||
})
|
||||
return db
|
||||
}
|
||||
Reference in New Issue
Block a user