95 lines
2.3 KiB
Go
95 lines
2.3 KiB
Go
|
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
|
||
|
|
}
|