feat(agent) multi-driver db monitoring, hot-plug config, release pipeline
This commit is contained in:
94
internal/collectors/mongodb/mongodb.go
Normal file
94
internal/collectors/mongodb/mongodb.go
Normal file
@@ -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
|
||||
}
|
||||
92
internal/collectors/mysql/mysql.go
Normal file
92
internal/collectors/mysql/mysql.go
Normal file
@@ -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)
|
||||
}
|
||||
}
|
||||
63
internal/collectors/postgres/postgres.go
Normal file
63
internal/collectors/postgres/postgres.go
Normal file
@@ -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))
|
||||
}
|
||||
}
|
||||
78
internal/collectors/redis/redis.go
Normal file
78
internal/collectors/redis/redis.go
Normal file
@@ -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
|
||||
}
|
||||
30
internal/collectors/sqlite/sqlite.go
Normal file
30
internal/collectors/sqlite/sqlite.go
Normal file
@@ -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()))
|
||||
}
|
||||
Reference in New Issue
Block a user