Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
167 changes: 96 additions & 71 deletions cmd/maintenant/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -46,8 +46,55 @@ func main() {
args = args[1:]
}

// --help and --version answer before the logger exists, so slog JSON does
// not land in the middle of human-readable output.
answerHelpOrVersion(args)
cfg, visited := loadConfig(args)

logOutput := os.Stdout
if mcpStdio {
logOutput = os.Stderr
}
logger := slog.New(slog.NewJSONHandler(logOutput, &slog.HandlerOptions{
Level: slogLevelFrom(cfg.LogLevel),
}))
if mcpStdio {
runMCPStdio(cfg, logger)
return
}

if cfg.MultiHost.InsecureSkipVerify {
logger.Warn("insecure TLS verification disabled — do not use in production")
}
logger.Info("maintenant starting", "version", version, "commit", commit, "build_date", buildDate, "mode", cfg.Mode)
if cfg.DemoMode {
logger.Info("demo mode enabled: instance is read-only")
}

validateStartup(cfg, logger)

// --copy-store-to runs the copy and exits: the binary has
// no subcommands and this feature does not introduce any.
if target := visited["copy-store-to"]; target != "" {
if cfg.Mode == "agent" {
logger.Error("--copy-store-to is not accepted in agent mode",
"fix", "an agent has no server data set to carry")
os.Exit(copyExitAgentBad)
}
os.Exit(runCopy(cfg.DBPath, target, visited["yes"] == "true", os.Stdout, os.Stdin, logger))
}

ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()

if cfg.Mode == "agent" {
runAgent(ctx, cfg, logger)
return
}
runServer(ctx, cfg, logger)
}

// answerHelpOrVersion prints --help or --version and exits, before the logger
// exists, so slog JSON does not land in the middle of human-readable output.
func answerHelpOrVersion(args []string) {
for _, arg := range args {
switch arg {
case "--help", "-h":
Expand All @@ -59,7 +106,10 @@ func main() {
os.Exit(0)
}
}
}

// loadConfig builds the configuration from the environment then the flags, exiting 2 on an unknown flag or a bad value.
func loadConfig(args []string) (app.Config, map[string]string) {
// One parser for every option: app.Registry holds the flag, its environment
// variable and how it lands in Config, so --help cannot advertise a flag the
// binary would then refuse. Unknown flag or bad value exits 2.
Expand All @@ -80,37 +130,25 @@ func main() {
fmt.Fprintf(os.Stderr, "error: %v\n", err)
os.Exit(2)
}
return cfg, visited
}

logOutput := os.Stdout
if mcpStdio {
logOutput = os.Stderr
}
logger := slog.New(slog.NewJSONHandler(logOutput, &slog.HandlerOptions{
Level: slogLevelFrom(cfg.LogLevel),
}))
if mcpStdio {
if cfg.Mode == "agent" {
logger.Error("--mcp-stdio is not accepted in agent mode", "fix", "run it against the server")
os.Exit(2)
}
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
err := mcp.Relay(ctx, mcp.SocketPath(cfg.DBPath), os.Stdin, os.Stdout)
stop()
if err != nil {
logger.Error("MCP stdio relay failed", "error", err)
os.Exit(1)
}
return
}

if cfg.MultiHost.InsecureSkipVerify {
logger.Warn("insecure TLS verification disabled — do not use in production")
func runMCPStdio(cfg app.Config, logger *slog.Logger) {
if cfg.Mode == "agent" {
logger.Error("--mcp-stdio is not accepted in agent mode", "fix", "run it against the server")
os.Exit(2)
}
logger.Info("maintenant starting", "version", version, "commit", commit, "build_date", buildDate, "mode", cfg.Mode)
if cfg.DemoMode {
logger.Info("demo mode enabled: instance is read-only")
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
err := mcp.Relay(ctx, mcp.SocketPath(cfg.DBPath), os.Stdin, os.Stdout)
stop()
if err != nil {
logger.Error("MCP stdio relay failed", "error", err)
os.Exit(1)
}
}

// validateStartup loads the extra CA bundle and checks the configuration, exiting 1 on the first failure.
func validateStartup(cfg app.Config, logger *slog.Logger) {
// Loaded before anything can probe: both server and agent modes run from
// here, and a bad path must stop the process rather than turn every TLS
// check into a misleading "unknown authority".
Expand Down Expand Up @@ -161,52 +199,39 @@ func main() {
logger.Error("invalid agent spool configuration", "error", err)
os.Exit(1)
}
}

// --copy-store-to runs the copy and exits: the binary has
// no subcommands and this feature does not introduce any.
if target := visited["copy-store-to"]; target != "" {
if cfg.Mode == "agent" {
logger.Error("--copy-store-to is not accepted in agent mode",
"fix", "an agent has no server data set to carry")
os.Exit(copyExitAgentBad)
}
os.Exit(runCopy(cfg.DBPath, target, visited["yes"] == "true", os.Stdout, os.Stdin, logger))
func runAgent(ctx context.Context, cfg app.Config, logger *slog.Logger) {
dataDir := os.Getenv("MAINTENANT_DATA_DIR")
if dataDir == "" {
dataDir = defaultAgentDataDir
}

ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()

if cfg.Mode == "agent" {
dataDir := os.Getenv("MAINTENANT_DATA_DIR")
if dataDir == "" {
dataDir = defaultAgentDataDir
}
agentCfg := agent.AgentConfig{
DataDir: dataDir,
ServerURL: cfg.MultiHost.ServerURL,
EnrollmentToken: cfg.MultiHost.EnrollmentToken,
RuntimeOverride: cfg.MultiHost.RuntimeOverride,
Label: cfg.MultiHost.Label,
NodeName: cfg.MultiHost.NodeName,
AgentVersion: version,
InsecureSkipVerify: cfg.MultiHost.InsecureSkipVerify,
ServerFingerprint: cfg.MultiHost.ServerFingerprint,
ProxyLabels: cfg.ProxyLabels,
SpoolMaxMemoryBytes: cfg.MultiHost.AgentSpoolMaxMemoryBytes,
SpoolMaxDiskBytes: cfg.MultiHost.AgentSpoolMaxDiskBytes,
SpoolMaxAgeSeconds: cfg.MultiHost.AgentSpoolMaxAgeSeconds,
}
if window := int64(resource.DefaultSnapshotRetention.Seconds()); agentCfg.SpoolMaxAgeSeconds > window {
logger.Warn("agent spool max age is past the server raw-sample window, older events are replayed after their purge",
"maxAgeSeconds", agentCfg.SpoolMaxAgeSeconds, "windowSeconds", window)
}
if err := agent.Run(ctx, agentCfg, logger); err != nil {
logger.Error("agent run failed", "error", err)
os.Exit(1)
}
return
agentCfg := agent.AgentConfig{
DataDir: dataDir,
ServerURL: cfg.MultiHost.ServerURL,
EnrollmentToken: cfg.MultiHost.EnrollmentToken,
RuntimeOverride: cfg.MultiHost.RuntimeOverride,
Label: cfg.MultiHost.Label,
NodeName: cfg.MultiHost.NodeName,
AgentVersion: version,
InsecureSkipVerify: cfg.MultiHost.InsecureSkipVerify,
ServerFingerprint: cfg.MultiHost.ServerFingerprint,
ProxyLabels: cfg.ProxyLabels,
SpoolMaxMemoryBytes: cfg.MultiHost.AgentSpoolMaxMemoryBytes,
SpoolMaxDiskBytes: cfg.MultiHost.AgentSpoolMaxDiskBytes,
SpoolMaxAgeSeconds: cfg.MultiHost.AgentSpoolMaxAgeSeconds,
}
if window := int64(resource.DefaultSnapshotRetention.Seconds()); agentCfg.SpoolMaxAgeSeconds > window {
logger.Warn("agent spool max age is past the server raw-sample window, older events are replayed after their purge",
"maxAgeSeconds", agentCfg.SpoolMaxAgeSeconds, "windowSeconds", window)
}
if err := agent.Run(ctx, agentCfg, logger); err != nil {
logger.Error("agent run failed", "error", err)
os.Exit(1)
}
}

func runServer(ctx context.Context, cfg app.Config, logger *slog.Logger) {
commercial.Register()
application, err := app.New(cfg, logger, app.WithExtensions(commercial.Extensions()))
if err != nil {
Expand Down
7 changes: 0 additions & 7 deletions internal/agent/client.go
Original file line number Diff line number Diff line change
Expand Up @@ -103,13 +103,6 @@ func NewClientWithFingerprint(ctx context.Context, serverURL string, fingerprint
}, nil
}

// NewClient dials the server at serverURL and returns a ready-to-use Client.
// serverURL should start with "grpcs://" (TLS) or "grpc://" (plaintext, not recommended).
// If insecureSkipVerify is true, TLS certificate validation is skipped (debug only).
func NewClient(ctx context.Context, serverURL string, insecureSkipVerify bool, logger *slog.Logger) (*Client, error) {
return NewClientWithFingerprint(ctx, serverURL, "", insecureSkipVerify, logger)
}

func normalizeFingerprint(fp string) (string, error) {
fp = strings.ToLower(strings.ReplaceAll(fp, ":", ""))
if fp == "" {
Expand Down
7 changes: 0 additions & 7 deletions internal/agent/commands.go
Original file line number Diff line number Diff line change
Expand Up @@ -92,13 +92,6 @@ func (r *CommandRunner) cancel(requestID string) {
}
}

// inFlight reports how many commands are currently running.
func (r *CommandRunner) inFlight() int {
r.mu.Lock()
defer r.mu.Unlock()
return len(r.running)
}

// CancelAll aborts every in-flight command, used when the stream goes away.
func (r *CommandRunner) CancelAll() {
r.mu.Lock()
Expand Down
6 changes: 6 additions & 0 deletions internal/agent/commands_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -267,3 +267,9 @@ func TestClampLogLines(t *testing.T) {
func TestCapabilities_AdvertisesLogs(t *testing.T) {
require.Contains(t, Capabilities(), CapabilityLogs)
}

func (r *CommandRunner) inFlight() int {
r.mu.Lock()
defer r.mu.Unlock()
return len(r.running)
}
2 changes: 1 addition & 1 deletion internal/agent/refusal_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ func startIngest(t *testing.T, srv agentpb.IngestServer, cert *tls.Certificate)

func dialIngest(t *testing.T, url string) *Client {
t.Helper()
c, err := NewClient(context.Background(), url, false, quietLogger())
c, err := NewClientWithFingerprint(context.Background(), url, "", false, quietLogger())
require.NoError(t, err)
t.Cleanup(func() { _ = c.Close() })
return c
Expand Down
12 changes: 0 additions & 12 deletions internal/agent/spool_store.go
Original file line number Diff line number Diff line change
Expand Up @@ -250,18 +250,6 @@ func (s *spoolStore) Depth() (int64, error) {
return n, nil
}

// MinSeq returns the lowest queued sequence, or 0 when the spool is empty.
func (s *spoolStore) MinSeq() (int64, error) {
var seq sql.NullInt64
if err := s.db.QueryRow(`SELECT MIN(seq) FROM spool_events`).Scan(&seq); err != nil {
return 0, fmt.Errorf("read lowest spool sequence: %w", err)
}
if !seq.Valid {
return 0, nil
}
return seq.Int64, nil
}

func (s *spoolStore) meta(key string) (int64, error) {
var v int64
if err := s.db.QueryRow(`SELECT value FROM spool_meta WHERE key = ?`, key).Scan(&v); err != nil {
Expand Down
27 changes: 0 additions & 27 deletions internal/alert/alerts.go
Original file line number Diff line number Diff line change
Expand Up @@ -75,30 +75,3 @@ func (d *RestartDetector) Check(ctx context.Context, c *container.Container) (in
AgentID: c.AgentID,
}, nil
}

// HealthAlert represents a health status change alert.
type HealthAlert struct {
ContainerID string
ContainerName string
PreviousHealth *container.HealthStatus
NewHealth container.HealthStatus
Severity container.AlertSeverity
Timestamp time.Time
}

// CheckHealthTransition returns an alert if a container transitions from healthy to unhealthy.
func CheckHealthTransition(c *container.Container, previousHealth *container.HealthStatus, newHealth container.HealthStatus) *HealthAlert {
// Only alert on healthy → unhealthy transition
if previousHealth == nil || *previousHealth != container.HealthHealthy || newHealth != container.HealthUnhealthy {
return nil
}

return &HealthAlert{
ContainerID: c.ID,
ContainerName: c.Name,
PreviousHealth: previousHealth,
NewHealth: newHealth,
Severity: c.AlertSeverity,
Timestamp: time.Now(),
}
}
79 changes: 79 additions & 0 deletions internal/alert/channel_patch.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,79 @@
// Copyright 2026 Benjamin Touchard (Kolapsis)
// SPDX-License-Identifier: Apache-2.0

package alert

import "strings"

// ChannelPatch is a partial update of a notification channel: a nil field keeps the stored value.
type ChannelPatch struct {
Name *string
Type *string
URL *string
Headers *string
Secret *string
Config *string
Enabled *bool
}

// ChannelFieldError names the patch field that failed validation.
type ChannelFieldError struct {
Field string
Msg string
}

func (e *ChannelFieldError) Error() string { return "field=" + e.Field + ": " + e.Msg }

// NormalizeConfig trims a channel config and turns a JSON null into the empty string.
func NormalizeConfig(config string) string {
config = strings.TrimSpace(config)
if config == "null" {
return ""
}
return config
}

// Extends reports whether the patch does more than switch the channel off, which never needs the edition (FR-001d).
func (p *ChannelPatch) Extends() bool {
return p.Name != nil || p.Type != nil || p.URL != nil ||
p.Headers != nil || p.Secret != nil || p.Config != nil ||
(p.Enabled != nil && *p.Enabled)
}

// TouchesCredentials reports whether the patch changes the secret or the config.
func (p *ChannelPatch) TouchesCredentials() bool { return p.Secret != nil || p.Config != nil }

// TouchesDestination reports whether the patch changes the URL or the type.
func (p *ChannelPatch) TouchesDestination() bool { return p.URL != nil || p.Type != nil }

// Apply writes the patch onto ch, refusing invalid headers and an empty secret; an absent secret keeps the stored one (FR-006).
func (p *ChannelPatch) Apply(ch *NotificationChannel) error {
if p.Name != nil {
ch.Name = *p.Name
}
if p.Type != nil {
ch.Type = *p.Type
}
if p.URL != nil {
ch.URL = *p.URL
}
if p.Headers != nil {
if _, err := ParseHeaders(*p.Headers); err != nil {
return &ChannelFieldError{Field: "headers", Msg: err.Error()}
}
ch.Headers = *p.Headers
}
if p.Secret != nil {
if *p.Secret == "" {
return &ChannelFieldError{Field: "secret", Msg: "cannot be cleared; delete the channel instead"}
}
ch.Secret = *p.Secret
}
if p.Config != nil {
ch.Config = NormalizeConfig(*p.Config)
}
if p.Enabled != nil {
ch.Enabled = *p.Enabled
}
return nil
}
Loading