// Package manager runs Megalodon's side of things: it keeps a connection to every server, tracks // who is online, records events, enforces bans, runs scheduled jobs and sends notifications. package manager import ( "context" "encoding/json" "errors" "fmt" "sort" "strings" "sync" "time" "git.veillax.com/Veillax/megalodon/internal/ark" "git.veillax.com/Veillax/megalodon/internal/config" ) // Action is something a person or a schedule asks to be done. type Action struct { Name string `json:"action"` Servers []string `json:"servers,omitempty"` // empty or ["*"]: every server SteamID string `json:"steam_id,omitempty"` Player string `json:"player,omitempty"` // name, for the record Message string `json:"message,omitempty"` Command string `json:"command,omitempty"` Time string `json:"time,omitempty"` Reason string `json:"reason,omitempty"` Minutes float64 `json:"minutes,omitempty"` // restart countdown, or ban length (0 = permanent) } // Result is how an action went on one server. type Result struct { Server string `json:"server"` OK bool `json:"ok"` Output string `json:"output,omitempty"` Error string `json:"error,omitempty"` } // ActionRole is the lowest role allowed to run an action. func ActionRole(name string) config.Role { switch name { case "broadcast", "say", "whisper", "kick", "ban", "unban", "allow", "disallow", "save": return config.Moderator } return config.Admin } type Manager struct { cfg *config.Holder dir string hub *Hub players *PlayerDB bans *BanDB mu sync.RWMutex inst map[string]*Instance ctx context.Context wg sync.WaitGroup jobMu sync.Mutex lastRun map[string]time.Time lastFire map[string]string client httpDoer } // New opens the data directory. Call Start to begin watching servers. func New(h *config.Holder) (*Manager, error) { dir := h.Get().DataDir hub, err := NewHub(dir) if err != nil { return nil, err } players, err := openPlayers(dir) if err != nil { return nil, err } bans, err := openBans(dir) if err != nil { return nil, err } return &Manager{ cfg: h, dir: dir, hub: hub, players: players, bans: bans, inst: map[string]*Instance{}, lastRun: map[string]time.Time{}, lastFire: map[string]string{}, client: defaultClient(), }, nil } func (m *Manager) Hub() *Hub { return m.hub } // Start begins watching every configured server and running background work until ctx ends. func (m *Manager) Start(ctx context.Context) { m.ctx = ctx m.Reload() m.go_(m.scheduler) m.go_(m.notifier) m.go_(m.housekeeping) } func (m *Manager) go_(f func(context.Context)) { m.wg.Add(1) go func() { defer m.wg.Done(); f(m.ctx) }() } // Wait blocks until background work has finished (after the context given to Start is cancelled) // and saves what is still in memory. func (m *Manager) Wait() { m.wg.Wait() m.mu.Lock() for id, i := range m.inst { i.stop() delete(m.inst, id) } m.mu.Unlock() m.players.Flush() m.hub.Close() } // Reload makes the running servers match the configuration: new ones start, removed ones stop, // changed ones are restarted. func (m *Manager) Reload() { servers := m.cfg.Get().Servers want := map[string]config.Server{} for _, s := range servers { want[s.ID] = s } m.mu.Lock() defer m.mu.Unlock() for id, i := range m.inst { if s, ok := want[id]; !ok || !sameServer(s, i.cfg) { i.stop() delete(m.inst, id) } } for _, s := range servers { if _, ok := m.inst[s.ID]; !ok { i := newInstance(m, s) m.inst[s.ID] = i i.start(m.ctx) } } } func sameServer(a, b config.Server) bool { x, _ := json.Marshal(a) y, _ := json.Marshal(b) return string(x) == string(y) } func (m *Manager) Instance(id string) (*Instance, bool) { m.mu.RLock() defer m.mu.RUnlock() i, ok := m.inst[id] return i, ok } // Views returns every server in configuration order. func (m *Manager) Views() []View { var out []View for _, s := range m.cfg.Get().Servers { if i, ok := m.Instance(s.ID); ok { out = append(out, i.View()) } } return out } func (m *Manager) Players() []PlayerRecord { return m.players.List() } func (m *Manager) Annotate(id, notes string, watch bool) error { if !ark.ValidSteamID(id) { return errors.New("not a Steam ID") } m.players.Annotate(id, ark.Clean(notes), watch) return nil } func (m *Manager) Bans() []Ban { return m.bans.List() } // Audit records something done outside a server (a settings change, say). func (m *Manager) Audit(actor, server, msg string) { m.hub.Publish(Event{Server: server, Type: "audit", Actor: actor, Message: msg}) } // targets resolves server ids; empty or "*" means all. func (m *Manager) targets(ids []string) ([]*Instance, error) { if len(ids) == 0 || (len(ids) == 1 && ids[0] == "*") { var out []*Instance for _, v := range m.Views() { if i, ok := m.Instance(v.ID); ok { out = append(out, i) } } if len(out) == 0 { return nil, errors.New("no servers are configured") } return out, nil } var out []*Instance for _, id := range ids { i, ok := m.Instance(id) if !ok { return nil, fmt.Errorf("unknown server %q", id) } out = append(out, i) } return out, nil } // Do carries out an action on its target servers, one after another, and reports each result. // The caller has already checked that actor may do this. func (m *Manager) Do(actor string, a Action) ([]Result, error) { if a.Name == "ban" || a.Name == "unban" { return m.banAction(actor, a) } targets, err := m.targets(a.Servers) if err != nil { return nil, err } results := make([]Result, len(targets)) var wg sync.WaitGroup for k, i := range targets { wg.Add(1) go func() { defer wg.Done() out, err := i.Perform(actor, a) r := Result{Server: i.cfg.ID, OK: err == nil, Output: out} if err != nil { r.Error = err.Error() } results[k] = r }() } wg.Wait() return results, nil } func (m *Manager) banAction(actor string, a Action) ([]Result, error) { if a.Name == "unban" { return m.Unban(actor, a.SteamID) } var servers []string if !(len(a.Servers) == 0 || (len(a.Servers) == 1 && a.Servers[0] == "*")) { servers = a.Servers } return m.banOn(actor, a.SteamID, a.Player, a.Reason, time.Duration(a.Minutes*float64(time.Minute)), servers) } // Ban records a ban and applies it to the servers it covers (all if servers is empty). dur 0 is // permanent. The ban is also enforced later: anyone banned who turns up on a covered server is // removed, even if that server was offline when the ban was made. func (m *Manager) Ban(actor, steamID, name, reason string, dur time.Duration, servers []string) error { _, err := m.banOn(actor, steamID, name, reason, dur, servers) return err } func (m *Manager) banOn(actor, steamID, name, reason string, dur time.Duration, servers []string) ([]Result, error) { if !ark.ValidSteamID(steamID) { return nil, errors.New("not a Steam ID") } targets, err := m.targets(servers) if err != nil { return nil, err } if name == "" { if r, ok := m.players.Get(steamID); ok { name = r.Name } } b := Ban{SteamID: steamID, Name: ark.Clean(name), Reason: ark.Clean(reason), By: actor, At: time.Now(), Servers: servers} desc := "permanently" if dur > 0 { exp := b.At.Add(dur) b.Expires = &exp desc = "for " + ark.Duration(dur) } m.bans.Add(b) who := steamID if b.Name != "" { who = b.Name + " (" + steamID + ")" } m.hub.Publish(Event{Type: "ban", Actor: actor, SteamID: steamID, Message: fmt.Sprintf("Banned %s %s%s", who, desc, withReason(b.Reason))}) cmd, _ := ark.ForPlayer("BanPlayer", steamID) var results []Result for _, i := range targets { out, err := i.rc.Exec(cmd) r := Result{Server: i.cfg.ID, OK: err == nil, Output: strings.TrimSpace(out)} if err != nil { r.Error = err.Error() } results = append(results, r) } return results, nil } // Unban lifts a ban everywhere. func (m *Manager) Unban(actor, steamID string) ([]Result, error) { if !ark.ValidSteamID(steamID) { return nil, errors.New("not a Steam ID") } b, _ := m.bans.Remove(steamID) who := steamID if b.Name != "" { who = b.Name + " (" + steamID + ")" } m.hub.Publish(Event{Type: "unban", Actor: actor, SteamID: steamID, Message: "Unbanned " + who}) cmd, _ := ark.ForPlayer("UnbanPlayer", steamID) targets, _ := m.targets(nil) var results []Result for _, i := range targets { out, err := i.rc.Exec(cmd) r := Result{Server: i.cfg.ID, OK: err == nil, Output: strings.TrimSpace(out)} if err != nil { r.Error = err.Error() } results = append(results, r) } return results, nil } // RunMacro sends each command of a macro to the server and gathers the replies. func (m *Manager) RunMacro(actor, server string, mc config.Macro) ([]Result, error) { i, ok := m.Instance(server) if !ok { return nil, fmt.Errorf("unknown server %q", server) } m.Audit(actor, server, "ran macro "+mc.Name) var out []Result for _, c := range mc.Commands { reply, err := i.Exec(c) r := Result{Server: server, OK: err == nil, Output: strings.TrimSpace(reply)} if err != nil { r.Error = err.Error() } out = append(out, r) if err != nil { break } } return out, nil } // ---- background work ---- // housekeeping flushes the player list and lifts expired bans. func (m *Manager) housekeeping(ctx context.Context) { t := time.NewTicker(30 * time.Second) defer t.Stop() for { select { case <-ctx.Done(): return case <-t.C: m.players.Flush() for _, b := range m.bans.Expired(time.Now()) { m.Unban("megalodon", b.SteamID) } } } } // scheduler runs the configured jobs. func (m *Manager) scheduler(ctx context.Context) { t := time.NewTicker(15 * time.Second) defer t.Stop() for { select { case <-ctx.Done(): return case now := <-t.C: for _, j := range m.cfg.Get().Jobs { if j.Enabled && m.due(j, now) { go m.RunJob(j, "job: "+j.Name) } } } } } var dayNames = [...]string{"sun", "mon", "tue", "wed", "thu", "fri", "sat"} func (m *Manager) due(j config.Job, now time.Time) bool { m.jobMu.Lock() defer m.jobMu.Unlock() if j.Every != "" { d, err := time.ParseDuration(j.Every) if err != nil { return false } last, ok := m.lastRun[j.ID] if !ok { // first sight: count from now rather than firing at once m.lastRun[j.ID] = now return false } if now.Sub(last) >= d { m.lastRun[j.ID] = now return true } return false } if len(j.Days) > 0 { ok := false for _, d := range j.Days { ok = ok || d == dayNames[now.Weekday()] } if !ok { return false } } minute := now.Format("2006-01-02 15:04") for _, at := range j.At { if at == now.Format("15:04") && m.lastFire[j.ID] != minute { m.lastFire[j.ID] = minute return true } } return false } // RunJob runs a job now. func (m *Manager) RunJob(j config.Job, actor string) []Result { a := Action{Name: j.Action, Servers: j.Servers, Message: j.Message, Reason: j.Reason} switch j.Action { case "command": a.Name, a.Command = "exec", j.Command case "restart": d := 5 * time.Minute if j.Countdown != "" { d, _ = time.ParseDuration(j.Countdown) } a.Minutes = d.Minutes() } results, err := m.Do(actor, a) if err != nil { m.hub.Publish(Event{Type: "error", Actor: actor, Message: "Job " + j.Name + " failed: " + err.Error()}) return nil } failed := 0 for _, r := range results { if !r.OK { failed++ } } msg := fmt.Sprintf("Ran job %q on %d server(s)", j.Name, len(results)) typ := "schedule" if failed > 0 { msg += fmt.Sprintf(", %d failed", failed) typ = "error" } m.hub.Publish(Event{Type: typ, Actor: actor, Message: msg}) return results } // Metrics renders the current state in Prometheus text format. func (m *Manager) Metrics() string { var sb strings.Builder views := m.Views() sort.Slice(views, func(a, b int) bool { return views[a].ID < views[b].ID }) sb.WriteString("# TYPE megalodon_server_up gauge\n# TYPE megalodon_players gauge\n# TYPE megalodon_players_max gauge\n# TYPE megalodon_rcon_rtt_ms gauge\n# TYPE megalodon_uptime_seconds gauge\n# TYPE megalodon_restart_pending gauge\n") for _, v := range views { l := fmt.Sprintf(`{server=%q}`, v.ID) up := 0 if v.Online { up = 1 } fmt.Fprintf(&sb, "megalodon_server_up%s %d\nmegalodon_players%s %d\nmegalodon_rcon_rtt_ms%s %d\n", l, up, l, len(v.Players), l, v.RTTMS) if v.Info != nil { fmt.Fprintf(&sb, "megalodon_players_max%s %d\n", l, v.Info.MaxPlayers) } if v.Since != nil { fmt.Fprintf(&sb, "megalodon_uptime_seconds%s %d\n", l, int(time.Since(*v.Since).Seconds())) } pending := 0 if v.Restart != nil { pending = 1 } fmt.Fprintf(&sb, "megalodon_restart_pending%s %d\n", l, pending) } return sb.String() }