diff --git a/.gitea/workflows/build.yml b/.gitea/workflows/build.yml new file mode 100644 index 0000000..e7cc17a --- /dev/null +++ b/.gitea/workflows/build.yml @@ -0,0 +1,45 @@ +name: build-and-release + +on: + push: + tags: + - "v*" + +jobs: + release: + runs-on: ubuntu-latest + + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-go@v5 + with: + go-version: '1.22' + + - name: Build binaries + run: make build-all + + - name: Create release + id: release + env: + NODE_AUTH_TOKEN: ${{ secrets.NODE_AUTH_TOKEN }} + run: | + TAG="${{ github.ref_name }}" + RESPONSE=$(curl -sf -X POST \ + -H "Authorization: token ${NODE_AUTH_TOKEN}" \ + -H "Content-Type: application/json" \ + -d "{\"tag_name\":\"${TAG}\",\"name\":\"${TAG}\",\"draft\":false,\"prerelease\":false}" \ + "${{ github.server_url }}/api/v1/repos/${{ github.repository }}/releases") + echo "id=$(echo "$RESPONSE" | jq -r .id)" >> $GITHUB_OUTPUT + + - name: Upload binaries + env: + NODE_AUTH_TOKEN: ${{ secrets.NODE_AUTH_TOKEN }} + run: | + for FILE in dist/bugswatch-*; do + curl -sf -X POST \ + -H "Authorization: token ${NODE_AUTH_TOKEN}" \ + -H "Content-Type: application/octet-stream" \ + --data-binary "@${FILE}" \ + "${{ github.server_url }}/api/v1/repos/${{ github.repository }}/releases/${{ steps.release.outputs.id }}/assets?name=$(basename ${FILE})" + done diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..8eadfd5 --- /dev/null +++ b/LICENSE @@ -0,0 +1,18 @@ +MIT No Attribution + +Copyright 2025 bugs.watch + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/Makefile b/Makefile new file mode 100644 index 0000000..99fcb55 --- /dev/null +++ b/Makefile @@ -0,0 +1,19 @@ +VERSION ?= $(shell git describe --tags --always --dirty 2>/dev/null || echo dev) +LDFLAGS = -ldflags="-s -w -X main.version=$(VERSION)" +OUT = dist + +.PHONY: build build-all clean + +build: + go mod tidy + CGO_ENABLED=0 go build $(LDFLAGS) -o $(OUT)/bugswatch ./cmd/agent + +build-all: + go mod tidy + CGO_ENABLED=0 GOOS=linux GOARCH=amd64 go build $(LDFLAGS) -o $(OUT)/bugswatch-linux-amd64 ./cmd/agent + CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build $(LDFLAGS) -o $(OUT)/bugswatch-linux-arm64 ./cmd/agent + CGO_ENABLED=0 GOOS=darwin GOARCH=amd64 go build $(LDFLAGS) -o $(OUT)/bugswatch-darwin-amd64 ./cmd/agent + CGO_ENABLED=0 GOOS=darwin GOARCH=arm64 go build $(LDFLAGS) -o $(OUT)/bugswatch-darwin-arm64 ./cmd/agent + +clean: + rm -rf $(OUT) diff --git a/cmd/agent/main.go b/cmd/agent/main.go index de37e70..4dfa6e7 100644 --- a/cmd/agent/main.go +++ b/cmd/agent/main.go @@ -2,24 +2,144 @@ package main import ( "context" + "encoding/json" + "flag" + "fmt" + "log" "os" "os/signal" "syscall" "bugs.watch/agent/internal/agent" + "bugs.watch/agent/internal/config" + "bugs.watch/agent/internal/dbmanager" ) func main() { + // Peek at the first arg before flag parsing to detect sub-commands. + if len(os.Args) > 1 && os.Args[1] == "db" { + runDBCmd(os.Args[2:]) + return + } + + cfg, err := config.Load() + if err != nil { + log.Fatalf("configuration error: %v", err) + } + + flag.StringVar(&cfg.DBConfigPath, "db-config", cfg.DBConfigPath, + "path to databases.json (default /etc/bugswatch/databases.json, or $BUGSWATCH_DB_CONFIG)") + flag.Parse() + ctx, cancel := context.WithCancel(context.Background()) defer cancel() sigCh := make(chan os.Signal, 1) signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM) - go func() { <-sigCh cancel() }() - agent.Run(ctx) -} \ No newline at end of file + agent.Run(ctx, cfg) +} + +// ── db sub-commands ────────────────────────────────────────────────────────── + +func runDBCmd(args []string) { + if len(args) == 0 { + fmt.Fprintln(os.Stderr, "usage: bugswatch db [args]") + os.Exit(1) + } + + // Allow --db-config as the first flag of a db sub-command. + cfgPath := config.DefaultDBConfigPath + if v := os.Getenv("BUGSWATCH_DB_CONFIG"); v != "" { + cfgPath = v + } + if len(args) >= 2 && args[0] == "--db-config" { + cfgPath = args[1] + args = args[2:] + } + + switch args[0] { + case "add": + if len(args) < 3 { + fmt.Fprintln(os.Stderr, "usage: bugswatch db add ") + fmt.Fprintln(os.Stderr, "drivers: mysql, mariadb, postgres, redis, mongodb, sqlite") + os.Exit(1) + } + dbAdd(cfgPath, args[1], args[2]) + + case "remove": + if len(args) < 2 { + fmt.Fprintln(os.Stderr, "usage: bugswatch db remove ") + os.Exit(1) + } + dbRemove(cfgPath, args[1]) + + case "list": + dbList(cfgPath) + + default: + fmt.Fprintf(os.Stderr, "unknown db command %q\n", args[0]) + os.Exit(1) + } +} + +func dbAdd(cfgPath, driver, url string) { + entries, err := dbmanager.ReadEntries(cfgPath) + if err != nil { + log.Fatalf("read config: %v", err) + } + + found := false + for i, e := range entries { + if e.Driver == driver { + entries[i].URL = url + found = true + break + } + } + if !found { + entries = append(entries, dbmanager.Entry{Driver: driver, URL: url}) + } + + if err := dbmanager.WriteEntries(cfgPath, entries); err != nil { + log.Fatalf("write config: %v", err) + } + fmt.Printf("added %s → %s\n", driver, cfgPath) +} + +func dbRemove(cfgPath, driver string) { + entries, err := dbmanager.ReadEntries(cfgPath) + if err != nil { + log.Fatalf("read config: %v", err) + } + + next := entries[:0] + for _, e := range entries { + if e.Driver != driver { + next = append(next, e) + } + } + + if err := dbmanager.WriteEntries(cfgPath, next); err != nil { + log.Fatalf("write config: %v", err) + } + fmt.Printf("removed %s from %s\n", driver, cfgPath) +} + +func dbList(cfgPath string) { + entries, err := dbmanager.ReadEntries(cfgPath) + if err != nil { + log.Fatalf("read config: %v", err) + } + if len(entries) == 0 { + fmt.Println("no database monitors configured") + return + } + enc := json.NewEncoder(os.Stdout) + enc.SetIndent("", " ") + enc.Encode(entries) +} diff --git a/go.mod b/go.mod index e3da3f3..afaaab0 100644 --- a/go.mod +++ b/go.mod @@ -1,3 +1,10 @@ module bugs.watch/agent go 1.22.12 + +require ( + github.com/go-sql-driver/mysql v1.8.1 + github.com/lib/pq v1.10.9 + github.com/redis/go-redis/v9 v9.7.3 + go.mongodb.org/mongo-driver v1.17.1 +) diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 761e49b..88b92b7 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -11,6 +11,8 @@ import ( "bugs.watch/agent/internal/collectors/disk" "bugs.watch/agent/internal/collectors/memory" "bugs.watch/agent/internal/collectors/uptime" + "bugs.watch/agent/internal/config" + "bugs.watch/agent/internal/dbmanager" "bugs.watch/agent/internal/sender" ) @@ -19,13 +21,12 @@ const ( windowDuration = 60 * time.Second ) -func Run(ctx context.Context) { +func Run(ctx context.Context, cfg config.Config) { log.Println("bugswatch agent starting") + log.Printf("db config: %s", cfg.DBConfigPath) - // Aggregation window window := aggregate.NewWindow() - // Register collectors cols := []collectors.Collector{ uptime.New(), cpu.New(), @@ -33,8 +34,10 @@ func Run(ctx context.Context) { disk.New(), } - // Sender (HTTP or logging) - snd := sender.NewHTTPSender("http://172.17.0.1:5173/api/health") + mgr := dbmanager.New(cfg.DBConfigPath) + go mgr.Watch(ctx) + + snd := sender.NewHTTPSender(cfg) ticker := time.NewTicker(sampleInterval) defer ticker.Stop() @@ -43,21 +46,19 @@ func Run(ctx context.Context) { select { case <-ctx.Done(): log.Println("bugswatch agent shutting down") - - // Best-effort final flush with bounded timeout shutdownCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) defer cancel() - - emitSnapshot(shutdownCtx, snd, window) + emitSnapshot(shutdownCtx, snd, window, mgr) return case <-ticker.C: for _, c := range cols { c.Collect(window) } + mgr.Collect(window) if window.Ready(windowDuration) { - emitSnapshot(ctx, snd, window) + emitSnapshot(ctx, snd, window, mgr) window = aggregate.NewWindow() } } @@ -68,6 +69,7 @@ func emitSnapshot( ctx context.Context, snd sender.Sender, window *aggregate.Window, + mgr *dbmanager.Manager, ) { snap := window.Snapshot() _ = snd.Send(ctx, snap) diff --git a/internal/aggregate/window.go b/internal/aggregate/window.go index b111634..b61921d 100644 --- a/internal/aggregate/window.go +++ b/internal/aggregate/window.go @@ -15,22 +15,36 @@ type Window struct { diskCount uint64 uptime uint64 + + // DB metrics — populated only when a DB collector is registered. + hasDB bool + dbDriver uint8 + dbConnPctSum float64 + dbConnPctCount uint64 + dbThreadsSum float64 + dbThreadsCount uint64 + dbSlowQueriesDelta uint64 + dbSizeBytes uint64 } type Snapshot struct { Timestamp time.Time - CPUAvg float64 - MemAvg float64 + CPUAvg float64 + MemAvg float64 DiskAvg float64 + Uptime uint64 - Uptime uint64 + // nil when DB monitoring is not enabled. + DBDriver *uint8 + DBConnPct *float64 + DBThreadsRunning *float64 + DBSlowQueries *uint64 + DBSizeBytes *uint64 } func NewWindow() *Window { - return &Window{ - start: time.Now(), - } + return &Window{start: time.Now()} } func (w *Window) AddCPU(value float64) { @@ -43,10 +57,41 @@ func (w *Window) AddMemory(value float64) { w.memCount++ } +func (w *Window) AddDisk(value float64) { + w.diskSum += value + w.diskCount++ +} + func (w *Window) SetUptime(value uint64) { w.uptime = value } +func (w *Window) AddDBConnPct(v float64) { + w.dbConnPctSum += v + w.dbConnPctCount++ + w.hasDB = true +} + +func (w *Window) AddDBThreadsRunning(v float64) { + w.dbThreadsSum += v + w.dbThreadsCount++ +} + +func (w *Window) AddDBSlowQueriesDelta(v uint64) { + w.dbSlowQueriesDelta += v +} + +func (w *Window) SetDBSizeBytes(v uint64) { + w.dbSizeBytes = v + w.hasDB = true +} + +// SetDBDriver records the numeric driver identifier (see dbmanager.Driver* constants). +// When multiple collectors are active this is called once per window by the manager. +func (w *Window) SetDBDriver(v uint8) { + w.dbDriver = v +} + func (w *Window) Ready(windowDuration time.Duration) bool { return time.Since(w.start) >= windowDuration } @@ -67,16 +112,36 @@ func (w *Window) Snapshot() Snapshot { diskAvg = w.diskSum / float64(w.diskCount) } - return Snapshot{ + snap := Snapshot{ Timestamp: time.Now(), CPUAvg: cpuAvg, MemAvg: memAvg, DiskAvg: diskAvg, Uptime: w.uptime, } -} -func (w *Window) AddDisk(value float64) { - w.diskSum += value - w.diskCount++ -} \ No newline at end of file + if w.hasDB { + driver := w.dbDriver + + var connPct float64 + if w.dbConnPctCount > 0 { + connPct = w.dbConnPctSum / float64(w.dbConnPctCount) + } + + var threads float64 + if w.dbThreadsCount > 0 { + threads = w.dbThreadsSum / float64(w.dbThreadsCount) + } + + slow := w.dbSlowQueriesDelta + sizeBytes := w.dbSizeBytes + + snap.DBDriver = &driver + snap.DBConnPct = &connPct + snap.DBThreadsRunning = &threads + snap.DBSlowQueries = &slow + snap.DBSizeBytes = &sizeBytes + } + + return snap +} diff --git a/internal/collectors/mongodb/mongodb.go b/internal/collectors/mongodb/mongodb.go new file mode 100644 index 0000000..d16e275 --- /dev/null +++ b/internal/collectors/mongodb/mongodb.go @@ -0,0 +1,94 @@ +package mongodb + +import ( + "context" + "time" + + "go.mongodb.org/mongo-driver/bson" + "go.mongodb.org/mongo-driver/mongo" + "go.mongodb.org/mongo-driver/mongo/options" + + "bugs.watch/agent/internal/aggregate" +) + +// Collector queries MongoDB serverStatus and listDatabases each sample interval. +// All queries run under a 3-second timeout; any failure is silently skipped. +// Metrics mapping: +// db_conn_pct → current / (current + available) * 100 +// db_threads_running → connections.current +// db_size_bytes → totalSize from listDatabases +type Collector struct { + client *mongo.Client +} + +func New(uri string) (*Collector, error) { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + client, err := mongo.Connect(ctx, options.Client().ApplyURI(uri). + SetConnectTimeout(3*time.Second). + SetServerSelectionTimeout(3*time.Second)) + if err != nil { + return nil, err + } + return &Collector{client: client}, nil +} + +func (c *Collector) Name() string { return "mongodb" } + +func (c *Collector) Close() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + c.client.Disconnect(ctx) +} + +func (c *Collector) Collect(window *aggregate.Window) { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + admin := c.client.Database("admin") + + var status bson.M + if err := admin.RunCommand(ctx, bson.D{ + {Key: "serverStatus", Value: 1}, + {Key: "repl", Value: 0}, + {Key: "metrics", Value: 0}, + {Key: "locks", Value: 0}, + }).Decode(&status); err != nil { + return + } + + if conns, ok := status["connections"].(bson.M); ok { + current := toFloat64(conns["current"]) + available := toFloat64(conns["available"]) + total := current + available + if total > 0 { + window.AddDBConnPct(current / total * 100) + } + if current > 0 { + window.AddDBThreadsRunning(current) + } + } + + var listResult bson.M + if err := admin.RunCommand(ctx, bson.D{ + {Key: "listDatabases", Value: 1}, + {Key: "nameOnly", Value: false}, + }).Decode(&listResult); err == nil { + if totalSize := toFloat64(listResult["totalSize"]); totalSize > 0 { + window.SetDBSizeBytes(uint64(totalSize)) + } + } +} + +func toFloat64(v interface{}) float64 { + switch x := v.(type) { + case int32: + return float64(x) + case int64: + return float64(x) + case float64: + return x + } + return 0 +} diff --git a/internal/collectors/mysql/mysql.go b/internal/collectors/mysql/mysql.go new file mode 100644 index 0000000..02f59b0 --- /dev/null +++ b/internal/collectors/mysql/mysql.go @@ -0,0 +1,92 @@ +package mysql + +import ( + "context" + "database/sql" + "strconv" + "time" + + _ "github.com/go-sql-driver/mysql" + + "bugs.watch/agent/internal/aggregate" +) + +// Collector queries MySQL global status each sample interval. +// All queries run under a 3-second timeout; any failure is silently skipped. +type Collector struct { + db *sql.DB + prevSlowQueries uint64 + hasPrev bool +} + +func New(dsn string) (*Collector, error) { + db, err := sql.Open("mysql", dsn) + if err != nil { + return nil, err + } + db.SetMaxOpenConns(2) + db.SetMaxIdleConns(1) + db.SetConnMaxLifetime(5 * time.Minute) + return &Collector{db: db}, nil +} + +func (c *Collector) Name() string { return "mysql" } + +func (c *Collector) Close() { c.db.Close() } + +func (c *Collector) Collect(window *aggregate.Window) { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + rows, err := c.db.QueryContext(ctx, + "SHOW GLOBAL STATUS WHERE Variable_name IN ('Threads_connected','Threads_running','Slow_queries')", + ) + if err != nil { + return + } + defer rows.Close() + + vars := make(map[string]uint64, 3) + for rows.Next() { + var name, raw string + if err := rows.Scan(&name, &raw); err != nil { + continue + } + if v, err := strconv.ParseUint(raw, 10, 64); err == nil { + vars[name] = v + } + } + if err := rows.Err(); err != nil { + return + } + + var maxConn uint64 + if err := c.db.QueryRowContext(ctx, "SELECT @@max_connections").Scan(&maxConn); err != nil || maxConn == 0 { + maxConn = 1 + } + + connPct := float64(vars["Threads_connected"]) / float64(maxConn) * 100 + window.AddDBConnPct(connPct) + window.AddDBThreadsRunning(float64(vars["Threads_running"])) + + slow := vars["Slow_queries"] + if c.hasPrev { + var delta uint64 + if slow >= c.prevSlowQueries { + delta = slow - c.prevSlowQueries + } + window.AddDBSlowQueriesDelta(delta) + } + c.prevSlowQueries = slow + c.hasPrev = true + + var sizeBytes uint64 + err = c.db.QueryRowContext(ctx, + "SELECT COALESCE(SUM(data_length + index_length), 0)"+ + " FROM information_schema.tables"+ + " WHERE table_schema NOT IN ('information_schema','performance_schema','mysql','sys')", + ).Scan(&sizeBytes) + if err == nil { + window.SetDBSizeBytes(sizeBytes) + } +} diff --git a/internal/collectors/postgres/postgres.go b/internal/collectors/postgres/postgres.go new file mode 100644 index 0000000..8c8c02a --- /dev/null +++ b/internal/collectors/postgres/postgres.go @@ -0,0 +1,63 @@ +package postgres + +import ( + "context" + "database/sql" + "time" + + _ "github.com/lib/pq" + + "bugs.watch/agent/internal/aggregate" +) + +// Collector queries PostgreSQL status each sample interval. +// All queries run under a 3-second timeout; any failure is silently skipped. +type Collector struct { + db *sql.DB +} + +func New(dsn string) (*Collector, error) { + db, err := sql.Open("postgres", dsn) + if err != nil { + return nil, err + } + db.SetMaxOpenConns(2) + db.SetMaxIdleConns(1) + db.SetConnMaxLifetime(5 * time.Minute) + return &Collector{db: db}, nil +} + +func (c *Collector) Name() string { return "postgres" } + +func (c *Collector) Close() { c.db.Close() } + +func (c *Collector) Collect(window *aggregate.Window) { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + // pg_stat_database is readable by any user for their own database. + var numBackends, maxConn int + err := c.db.QueryRowContext(ctx, + `SELECT d.numbackends, s.setting::int + FROM pg_stat_database d, pg_settings s + WHERE d.datname = current_database() AND s.name = 'max_connections'`, + ).Scan(&numBackends, &maxConn) + if err == nil && maxConn > 0 { + window.AddDBConnPct(float64(numBackends) / float64(maxConn) * 100) + } + + // pg_stat_activity requires pg_monitor role (PG 10+); silently skip on error. + var activeQueries int + if err := c.db.QueryRowContext(ctx, + `SELECT count(*) FROM pg_stat_activity WHERE state = 'active'`, + ).Scan(&activeQueries); err == nil { + window.AddDBThreadsRunning(float64(activeQueries)) + } + + var sizeBytes int64 + if err := c.db.QueryRowContext(ctx, + `SELECT pg_database_size(current_database())`, + ).Scan(&sizeBytes); err == nil { + window.SetDBSizeBytes(uint64(sizeBytes)) + } +} diff --git a/internal/collectors/redis/redis.go b/internal/collectors/redis/redis.go new file mode 100644 index 0000000..31cfa7f --- /dev/null +++ b/internal/collectors/redis/redis.go @@ -0,0 +1,78 @@ +package redis + +import ( + "context" + "strconv" + "strings" + "time" + + "github.com/redis/go-redis/v9" + + "bugs.watch/agent/internal/aggregate" +) + +// Collector queries Redis INFO each sample interval. +// All queries run under a 3-second timeout; any failure is silently skipped. +// Metrics mapping: +// db_conn_pct → used_memory / maxmemory * 100 (skipped when maxmemory=0) +// db_threads_running → connected_clients +// db_size_bytes → used_memory +type Collector struct { + client *redis.Client +} + +func New(addr string) (*Collector, error) { + opts, err := redis.ParseURL(addr) + if err != nil { + return nil, err + } + opts.DialTimeout = 3 * time.Second + opts.ReadTimeout = 3 * time.Second + opts.WriteTimeout = 3 * time.Second + return &Collector{client: redis.NewClient(opts)}, nil +} + +func (c *Collector) Name() string { return "redis" } + +func (c *Collector) Close() { c.client.Close() } + +func (c *Collector) Collect(window *aggregate.Window) { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + + info, err := c.client.Info(ctx, "memory", "clients").Result() + if err != nil { + return + } + + fields := parseInfo(info) + + usedMem, _ := strconv.ParseUint(fields["used_memory"], 10, 64) + maxMem, _ := strconv.ParseUint(fields["maxmemory"], 10, 64) + clients, _ := strconv.ParseUint(fields["connected_clients"], 10, 64) + + if usedMem > 0 { + window.SetDBSizeBytes(usedMem) + } + if clients > 0 { + window.AddDBThreadsRunning(float64(clients)) + } + if maxMem > 0 && usedMem > 0 { + window.AddDBConnPct(float64(usedMem) / float64(maxMem) * 100) + } +} + +func parseInfo(s string) map[string]string { + out := make(map[string]string) + for _, line := range strings.Split(s, "\n") { + line = strings.TrimSpace(line) + if line == "" || strings.HasPrefix(line, "#") { + continue + } + k, v, ok := strings.Cut(line, ":") + if ok { + out[strings.TrimSpace(k)] = strings.TrimSpace(v) + } + } + return out +} diff --git a/internal/collectors/sqlite/sqlite.go b/internal/collectors/sqlite/sqlite.go new file mode 100644 index 0000000..47b3bec --- /dev/null +++ b/internal/collectors/sqlite/sqlite.go @@ -0,0 +1,30 @@ +package sqlite + +import ( + "os" + + "bugs.watch/agent/internal/aggregate" +) + +// Collector reports the on-disk size of a SQLite database file. +// No driver or connection is needed; we simply stat the file. +// Failures are silently skipped. +type Collector struct { + path string +} + +func New(path string) *Collector { + return &Collector{path: path} +} + +func (c *Collector) Name() string { return "sqlite" } + +func (c *Collector) Close() {} + +func (c *Collector) Collect(window *aggregate.Window) { + info, err := os.Stat(c.path) + if err != nil { + return + } + window.SetDBSizeBytes(uint64(info.Size())) +} diff --git a/internal/config/config.go b/internal/config/config.go new file mode 100644 index 0000000..974246e --- /dev/null +++ b/internal/config/config.go @@ -0,0 +1,45 @@ +package config + +import ( + "errors" + "os" +) + +const DefaultDBConfigPath = "/etc/bugswatch/databases.json" + +type Config struct { + Endpoint string + AgentID string + APIToken string + HostID string + Env string + DBConfigPath string +} + +func Load() (Config, error) { + dbConfig := os.Getenv("BUGSWATCH_DB_CONFIG") + if dbConfig == "" { + dbConfig = DefaultDBConfigPath + } + + cfg := Config{ + Endpoint: os.Getenv("BUGSWATCH_ENDPOINT"), + AgentID: os.Getenv("BUGSWATCH_AGENT_ID"), + APIToken: os.Getenv("BUGSWATCH_API_TOKEN"), + HostID: os.Getenv("BUGSWATCH_HOST_ID"), + Env: os.Getenv("BUGSWATCH_ENVIRONMENT"), + DBConfigPath: dbConfig, + } + + if cfg.Endpoint == "" { + return Config{}, errors.New("BUGSWATCH_ENDPOINT not set") + } + if cfg.AgentID == "" { + return Config{}, errors.New("BUGSWATCH_AGENT_ID not set") + } + if cfg.APIToken == "" { + return Config{}, errors.New("BUGSWATCH_API_TOKEN not set") + } + + return cfg, nil +} diff --git a/internal/dbmanager/manager.go b/internal/dbmanager/manager.go new file mode 100644 index 0000000..80ded0a --- /dev/null +++ b/internal/dbmanager/manager.go @@ -0,0 +1,203 @@ +package dbmanager + +import ( + "context" + "crypto/md5" + "encoding/json" + "fmt" + "log" + "os" + "path/filepath" + "sync" + "time" + + "bugs.watch/agent/internal/aggregate" + "bugs.watch/agent/internal/collectors" + mongocol "bugs.watch/agent/internal/collectors/mongodb" + mysqlcol "bugs.watch/agent/internal/collectors/mysql" + pgcol "bugs.watch/agent/internal/collectors/postgres" + rediscol "bugs.watch/agent/internal/collectors/redis" + sqlitecol "bugs.watch/agent/internal/collectors/sqlite" +) + +// Driver type identifiers — used as the db_driver field in the health payload +// so the dashboard can render the correct icon without parsing strings. +const ( + DriverNone uint8 = 0 + DriverMySQL uint8 = 1 // also MariaDB + DriverPostgres uint8 = 2 + DriverRedis uint8 = 3 + DriverMongoDB uint8 = 4 + DriverSQLite uint8 = 5 +) + +// Entry is one database config record in databases.json. +type Entry struct { + Driver string `json:"driver"` + URL string `json:"url"` +} + +type dbCollector interface { + collectors.Collector + Close() +} + +// Manager watches a config file and hot-swaps DB collectors without agent restart. +// Collect() is safe to call concurrently with Watch(). +type Manager struct { + mu sync.RWMutex + active []dbCollector + configPath string + lastHash [16]byte +} + +func New(configPath string) *Manager { + return &Manager{configPath: configPath} +} + +// Watch polls the config file every 30 seconds and reloads on change. +// Blocks until ctx is cancelled. +func (m *Manager) Watch(ctx context.Context) { + m.reload() + + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + m.swap(nil) + return + case <-ticker.C: + m.reload() + } + } +} + +// Collect runs all active DB collectors against the window. +// Holds RLock for the full duration so swaps never race with in-progress queries. +func (m *Manager) Collect(window *aggregate.Window) { + m.mu.RLock() + defer m.mu.RUnlock() + if len(m.active) == 0 { + return + } + window.SetDBDriver(m.driverID(m.active[0].Name())) + for _, c := range m.active { + c.Collect(window) + } +} + +func (m *Manager) driverID(name string) uint8 { + switch name { + case "mysql", "mariadb": + return DriverMySQL + case "postgres": + return DriverPostgres + case "redis": + return DriverRedis + case "mongodb": + return DriverMongoDB + case "sqlite": + return DriverSQLite + default: + return DriverNone + } +} + +func (m *Manager) reload() { + data, err := os.ReadFile(m.configPath) + if err != nil { + if os.IsNotExist(err) { + m.swap(nil) + } + return + } + + hash := md5.Sum(data) + m.mu.RLock() + unchanged := hash == m.lastHash + m.mu.RUnlock() + if unchanged { + return + } + + var entries []Entry + if err := json.Unmarshal(data, &entries); err != nil { + log.Printf("dbmanager: invalid config: %v", err) + return + } + + var next []dbCollector + for _, e := range entries { + c, err := newCollector(e) + if err != nil { + log.Printf("dbmanager: %s: %v", e.Driver, err) + continue + } + next = append(next, c) + log.Printf("dbmanager: loaded %s", e.Driver) + } + + m.swap(next) + + m.mu.Lock() + m.lastHash = hash + m.mu.Unlock() +} + +func (m *Manager) swap(next []dbCollector) { + m.mu.Lock() + old := m.active + m.active = next + m.mu.Unlock() + + for _, c := range old { + c.Close() + } +} + +func newCollector(e Entry) (dbCollector, error) { + switch e.Driver { + case "mysql", "mariadb": + return mysqlcol.New(e.URL) + case "postgres", "postgresql": + return pgcol.New(e.URL) + case "redis": + return rediscol.New(e.URL) + case "mongodb": + return mongocol.New(e.URL) + case "sqlite": + return sqlitecol.New(e.URL), nil + default: + return nil, fmt.Errorf("unknown driver %q (supported: mysql, mariadb, postgres, redis, mongodb, sqlite)", e.Driver) + } +} + +// ReadEntries reads the config file, returning an empty slice if it doesn't exist. +func ReadEntries(configPath string) ([]Entry, error) { + data, err := os.ReadFile(configPath) + if os.IsNotExist(err) { + return nil, nil + } + if err != nil { + return nil, err + } + var entries []Entry + if err := json.Unmarshal(data, &entries); err != nil { + return nil, err + } + return entries, nil +} + +// WriteEntries writes entries to the config file, creating the directory if needed. +func WriteEntries(configPath string, entries []Entry) error { + if err := os.MkdirAll(filepath.Dir(configPath), 0755); err != nil { + return err + } + data, err := json.MarshalIndent(entries, "", " ") + if err != nil { + return err + } + return os.WriteFile(configPath, data, 0644) +} diff --git a/internal/sender/http.go b/internal/sender/http.go index 75a247f..ea984e8 100644 --- a/internal/sender/http.go +++ b/internal/sender/http.go @@ -7,30 +7,44 @@ import ( "net/http" "time" + "bugs.watch/agent/internal/config" "bugs.watch/agent/internal/aggregate" ) type HTTPSender struct { client *http.Client endpoint string + token string + hostID string + env string } -func NewHTTPSender(endpoint string) *HTTPSender { +func NewHTTPSender(cfg config.Config) *HTTPSender { return &HTTPSender{ client: &http.Client{ Timeout: 5 * time.Second, }, - endpoint: endpoint, + endpoint: cfg.Endpoint + "/api/v1/agents/" + cfg.AgentID, + token: cfg.APIToken, + hostID: cfg.HostID, + env: cfg.Env, } } func (s *HTTPSender) Send(ctx context.Context, snap aggregate.Snapshot) error { payload := SnapshotPayload{ - Timestamp: snap.Timestamp, - CPUAvg: snap.CPUAvg, - MemAvg: snap.MemAvg, - DiskAvg: snap.DiskAvg, - Uptime: snap.Uptime, + Timestamp: snap.Timestamp, + CPUAvg: snap.CPUAvg, + MemAvg: snap.MemAvg, + DiskAvg: snap.DiskAvg, + Uptime: snap.Uptime, + HostID: s.hostID, + Environment: s.env, + DBDriver: snap.DBDriver, + DBConnPct: snap.DBConnPct, + DBThreadsRunning: snap.DBThreadsRunning, + DBSlowQueries: snap.DBSlowQueries, + DBSizeBytes: snap.DBSizeBytes, } body, err := json.Marshal(payload) @@ -40,7 +54,7 @@ func (s *HTTPSender) Send(ctx context.Context, snap aggregate.Snapshot) error { req, err := http.NewRequestWithContext( ctx, - http.MethodPost, + http.MethodPut, s.endpoint, bytes.NewReader(body), ) @@ -49,6 +63,7 @@ func (s *HTTPSender) Send(ctx context.Context, snap aggregate.Snapshot) error { } req.Header.Set("Content-Type", "application/json") + req.Header.Set("Authorization", "Bearer " + s.token) resp, err := s.client.Do(req) if err != nil { diff --git a/internal/sender/payload.go b/internal/sender/payload.go index 373bc29..000ef88 100644 --- a/internal/sender/payload.go +++ b/internal/sender/payload.go @@ -9,6 +9,16 @@ type SnapshotPayload struct { CPUAvg float64 `json:"cpu_avg"` MemAvg float64 `json:"mem_avg"` DiskAvg float64 `json:"disk_avg"` + Uptime uint64 `json:"uptime"` - Uptime uint64 `json:"uptime"` + HostID string `json:"host_id,omitempty"` + Environment string `json:"environment,omitempty"` + + // Optional — only present when DB monitoring is enabled. + // DBDriver is a numeric identifier: 1=mysql/mariadb 2=postgres 3=redis 4=mongodb 5=sqlite + DBDriver *uint8 `json:"db_driver,omitempty"` + DBConnPct *float64 `json:"db_conn_pct,omitempty"` + DBThreadsRunning *float64 `json:"db_threads_running,omitempty"` + DBSlowQueries *uint64 `json:"db_slow_queries,omitempty"` + DBSizeBytes *uint64 `json:"db_size_bytes,omitempty"` } diff --git a/package.json b/package.json new file mode 100644 index 0000000..0e685a5 --- /dev/null +++ b/package.json @@ -0,0 +1,9 @@ +{ + "name": "bugswatch-agent", + "version": "0.1.0", + "scripts": { + "release": "bun version patch && git push && git push --tags", + "release:minor": "bun version minor && git push && git push --tags", + "release:major": "bun version major && git push && git push --tags" + } +}