LovroG05/nvllama-watchdog
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

influxdb+system

Browse Source
This commit is contained in:
Lovro Govekar
2026-10-11 14:39:54 +02:00
parent 8fea355298
commit d7989da6aa
7 changed files with 569 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
config/config.go
View File

@@ -15,11 +15,12 @@ type Config struct {
}
type ReaderCfg struct {
Type string `yaml:"type"`
Endpoint string `yaml:"endpoint,omitempty"`
Model string `yaml:"model,omitempty"`
Type string `yaml:"type"`
Endpoint string `yaml:"endpoint,omitempty"`
Model string `yaml:"model,omitempty"`
Enabled map[string]bool `yaml:"enabled,omitempty"`
TopProcesses int `yaml:"top_processes,omitempty"`
}
type ExporterCfg struct {
Type string `yaml:"type"`
Addr string `yaml:"addr,omitempty"`

8
config/config.yaml
View File

@@ -6,6 +6,14 @@ readers:
endpoint: "http://10.169.8.100:8080/metrics"
model: "qwen3.8-27b-UD-Q4_K_XL-262144"
- type: nvidia_smi
- type: system
enabled:
cpu: true
memory: true
disk: false
network: true
load: true
uptime: true
exporters:
- type: dev

174
exporters/influxdb.go Normal file
View File

@@ -0,0 +1,174 @@
package exporters
import (
"bytes"
"context"
"fmt"
"net/http"
"sort"
"strings"
"time"
"git.govekar.net/LovroG05/nvllama-watchdog/core"
)
type InfluxDBExporter struct {
url string // full write URL, e.g. "http://localhost:8086/write?db=telegraf"
token string // for v2/v3
user string // for v1 basic auth
pass string // for v1 basic auth
client *http.Client
}
// NewInfluxDBExporter creates an InfluxDB exporter.
//
// Version-specific URL examples:
//
// v1: http://localhost:8086/write?db=nvllama&rp=autogen
// v2: http://localhost:8086/api/v2/write?org=myorg&bucket=nvllama
// v3: http://localhost:8086/api/v2/write?bucket=nvllama
func NewInfluxDBExporter(url, token, user, pass string) *InfluxDBExporter {
return &InfluxDBExporter{
url: url,
token: token,
user: user,
pass: pass,
client: &http.Client{Timeout: 10 * time.Second},
}
}
func (e *InfluxDBExporter) Name() string { return "influxdb" }
func (e *InfluxDBExporter) Write(ctx context.Context, metrics []core.Metric) error {
body := EncodeLineProtocol(metrics)
if body == "" {
return nil
}
req, err := http.NewRequestWithContext(ctx, http.MethodPost, e.url, bytes.NewReader([]byte(body)))
if err != nil {
return fmt.Errorf("influxdb: %w", err)
}
req.Header.Set("Content-Type", "text/plain; charset=utf-8")
// Auth: token (v2/v3) or basic (v1)
if e.token != "" {
req.Header.Set("Authorization", "Token "+e.token)
} else if e.user != "" {
req.SetBasicAuth(e.user, e.pass)
}
resp, err := e.client.Do(req)
if err != nil {
return fmt.Errorf("influxdb: %w", err)
}
defer resp.Body.Close()
if resp.StatusCode >= 300 {
buf := make([]byte, 256)
n, _ := resp.Body.Read(buf)
return fmt.Errorf("influxdb: HTTP %d: %s", resp.StatusCode, string(buf[:n]))
}
return nil
}
// ─── Line Protocol Encoder ──────────────────────────────────────────────────
// EncodeLineProtocol converts []core.Metric to InfluxDB line protocol.
//
// Each metric becomes one line:
//
// measurement,tag1=val1,tag2=val2 value=<float> <timestamp_ns>
//
// The metric Name is the measurement, Labels are tags, and Value is
// written as a single "value" field.
func EncodeLineProtocol(metrics []core.Metric) string {
if len(metrics) == 0 {
return ""
}
// Group by (measurement + tags) so we can batch fields if needed.
// For our use case each metric is one value, so one line per metric.
var b strings.Builder
b.Grow(len(metrics) * 128)
for i, m := range metrics {
if i > 0 {
b.WriteByte('\n')
}
// Measurement (escape commas and spaces)
b.WriteString(escapeMeasurement(m.Name))
// Tags (sorted for determinism)
if len(m.Labels) > 0 {
keys := make([]string, 0, len(m.Labels))
for k := range m.Labels {
keys = append(keys, k)
}
sort.Strings(keys)
for _, k := range keys {
b.WriteByte(',')
b.WriteString(escapeTagKey(k))
b.WriteByte('=')
b.WriteString(escapeTagValue(m.Labels[k]))
}
}
// Fields
b.WriteString(" value=")
b.WriteString(formatFloat(m.Value))
// Timestamp (nanoseconds since epoch)
if !m.Timestamp.IsZero() {
b.WriteByte(' ')
fmt.Fprintf(&b, "%d", m.Timestamp.UnixNano())
}
}
return b.String()
}
func escapeMeasurement(s string) string {
// Escape commas and spaces in measurement names
if strings.ContainsAny(s, ", ") {
var b strings.Builder
for _, c := range s {
if c == ',' || c == ' ' {
b.WriteByte('\\')
}
b.WriteRune(c)
}
return b.String()
}
return s
}
func escapeTagKey(s string) string {
return escapeTagValue(s)
}
func escapeTagValue(s string) string {
if strings.ContainsAny(s, ", =") {
var b strings.Builder
for _, c := range s {
if c == ',' || c == '=' || c == ' ' {
b.WriteByte('\\')
}
b.WriteRune(c)
}
return b.String()
}
return s
}
func formatFloat(f float64) string {
// Write as integer if it's a whole number (InfluxDB treats unquoted
// integers as i64, which is more storage-efficient)
if f == float64(int64(f)) && f < 9.223372036854775807e18 && f > -9.223372036854775808e18 {
return fmt.Sprintf("%di", int64(f))
}
return fmt.Sprintf("%g", f)
}

9
go.mod
View File

@@ -4,11 +4,20 @@ go 1.27.1
require (
github.com/prometheus/common v0.72.1
github.com/shirou/gopsutil/v4 v4.26.9
gopkg.in/yaml.v3 v3.0.1
)
require (
github.com/ebitengine/purego v0.11.1 // indirect
github.com/go-ole/go-ole v1.2.6 // indirect
github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
github.com/power-devops/perfstat v0.0.0-20260805114148-88456608a4f6 // indirect
github.com/prometheus/client_model v0.6.3 // indirect
github.com/tklauser/go-sysconf v0.4.0 // indirect
github.com/tklauser/numcpus v0.12.0 // indirect
github.com/yusufpapurcu/wmi v1.2.4 // indirect
golang.org/x/sys v0.49.0 // indirect
google.golang.org/protobuf v1.36.12 // indirect
)

22
go.sum
View File

@@ -1,9 +1,31 @@
github.com/ebitengine/purego v0.11.1 h1:2zpWRSQNVKN4eKsKO9eM1ILDgWfYMY9GwqRmK6XeQ/0=
github.com/ebitengine/purego v0.11.1/go.mod h1:DCHPP08djqhNSoTfImcnHYQRZmd0qhakvrozqaEYhGQ=
github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY=
github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4=
github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
github.com/power-devops/perfstat v0.0.0-20260805114148-88456608a4f6 h1:jL3a8soXdzuTCcRnKhOmtcsVOObdDTFf4O2B403HPRU=
github.com/power-devops/perfstat v0.0.0-20260805114148-88456608a4f6/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
github.com/prometheus/client_model v0.6.3 h1:O0jaTVAYNxTHYInEPFJt5I3+sN8zqBtVMPTB1qyxiEo=
github.com/prometheus/client_model v0.6.3/go.mod h1:gpN5P9S7Rr6Yr92PiQ+Ixvhf6JZEkF1dnxsYL2aPBEM=
github.com/prometheus/common v0.72.1 h1:ZAiQed/nk8IIcsdzb2TlJu23NEUDyDLQ9TXTN2WRK6Y=
github.com/prometheus/common v0.72.1/go.mod h1:1IoBY4IRaJnhs2ESjNzjilP1cJ/P8pimGoomq7zRXrA=
github.com/shirou/gopsutil/v4 v4.26.9 h1:CaBo/hBFqvlJoLvAQEKdHdf5GD97/MN2ACfyv63+2Ig=
github.com/shirou/gopsutil/v4 v4.26.9/go.mod h1:nKH+8wX2zxr/mDbsR+AAy6Qd2z4ZOQ7wfQaEp+yrYmg=
github.com/tklauser/go-sysconf v0.4.0 h1:7H0uAN+7RkwWRaxhYXDLqa5V3LPrJeV8wmD9dRUgPQU=
github.com/tklauser/go-sysconf v0.4.0/go.mod h1:8mTNWyog7H+MpKijp4VmKJAd2bbYQ2zuUwkYRbUArPI=
github.com/tklauser/numcpus v0.12.0 h1:NR85qdvHA9pFse3x3weVZ0r0ST8R6l5RHbZrlRaqob4=
github.com/tklauser/numcpus v0.12.0/go.mod h1:ABHeXzJnr/qqwguhClkZKT1/8VABcYrsyUiUGobwWJg=
github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
golang.org/x/sys v0.49.0 h1:XbzkgYJHdqh/8m2Uu0W/dQv8nktxx4BFHp1M0gROTXA=
golang.org/x/sys v0.49.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og=
golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc=
google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

2
main.go
View File

@@ -27,6 +27,8 @@ func main() {
rs = append(rs, readers.NewLLamaCPPReader(rc.Endpoint, rc.Model))
case "nvidia_smi":
rs = append(rs, readers.NewNvidiaSmiReader())
case "system":
rs = append(rs, readers.NewSystemReader(rc.Enabled))
}
}

349
readers/system.go Normal file
View File

@@ -0,0 +1,349 @@
package readers
import (
"context"
"fmt"
"sort"
"time"
"github.com/shirou/gopsutil/v4/cpu"
"github.com/shirou/gopsutil/v4/disk"
"github.com/shirou/gopsutil/v4/host"
"github.com/shirou/gopsutil/v4/load"
"github.com/shirou/gopsutil/v4/mem"
"github.com/shirou/gopsutil/v4/net"
"github.com/shirou/gopsutil/v4/process"
"git.govekar.net/LovroG05/nvllama-watchdog/core"
)
type SystemReader struct {
enabled map[string]bool
}
func NewSystemReader(enabled map[string]bool) *SystemReader {
if enabled == nil {
enabled = map[string]bool{
"cpu": true,
"memory": true,
"disk": true,
"network": true,
"load": true,
"uptime": true,
}
}
return &SystemReader{enabled: enabled}
}
func (r *SystemReader) Name() string { return "system" }
func (r *SystemReader) Read(ctx context.Context) ([]core.Metric, error) {
now := time.Now()
var metrics []core.Metric
var errs []error
if r.enabled["cpu"] {
ms, err := r.readCPU(ctx)
if err != nil {
errs = append(errs, fmt.Errorf("cpu: %w", err))
}
metrics = append(metrics, ms...)
}
if r.enabled["memory"] {
ms, err := r.readMemory(ctx)
if err != nil {
errs = append(errs, fmt.Errorf("memory: %w", err))
}
metrics = append(metrics, ms...)
}
if r.enabled["disk"] {
ms, err := r.readDisk(ctx)
if err != nil {
errs = append(errs, fmt.Errorf("disk: %w", err))
}
metrics = append(metrics, ms...)
}
if r.enabled["network"] {
ms, err := r.readNetwork(ctx)
if err != nil {
errs = append(errs, fmt.Errorf("network: %w", err))
}
metrics = append(metrics, ms...)
}
if r.enabled["load"] {
ms, err := r.readLoad(ctx)
if err != nil {
errs = append(errs, fmt.Errorf("load: %w", err))
}
metrics = append(metrics, ms...)
}
if r.enabled["uptime"] {
ms, err := r.readUptime(ctx)
if err != nil {
errs = append(errs, fmt.Errorf("uptime: %w", err))
}
metrics = append(metrics, ms...)
}
for i := range metrics {
metrics[i].Timestamp = now
}
if len(errs) > 0 {
return metrics, fmt.Errorf("system reader: %v", errs)
}
return metrics, nil
}
// ─── CPU ────────────────────────────────────────────────────────────────────
func (r *SystemReader) readCPU(_ context.Context) ([]core.Metric, error) {
// Per-core usage (sample over 250ms to get a meaningful percentage)
percents, err := cpu.Percent(250*time.Millisecond, true)
if err != nil {
return nil, err
}
var metrics []core.Metric
for i, p := range percents {
metrics = append(metrics, core.Metric{
Name: "cpu_usage_percent",
Help: "CPU usage per core in percent",
Type: "gauge",
Value: p,
Labels: map[string]string{"core": fmt.Sprintf("%d", i)},
})
}
// Total CPU times (user, system, idle, iowait, irq, softirq, steal)
times, err := cpu.Times(false)
if err != nil {
return metrics, err
}
if len(times) > 0 {
t := times[0]
total := t.User + t.System + t.Idle + t.Iowait + t.Irq + t.Softirq + t.Steal + t.Nice
if total > 0 {
metrics = append(metrics,
core.Metric{Name: "cpu_user_percent", Value: t.User / total * 100, Labels: map[string]string{}},
core.Metric{Name: "cpu_system_percent", Value: t.System / total * 100, Labels: map[string]string{}},
core.Metric{Name: "cpu_idle_percent", Value: t.Idle / total * 100, Labels: map[string]string{}},
core.Metric{Name: "cpu_iowait_percent", Value: t.Iowait / total * 100, Labels: map[string]string{}},
core.Metric{Name: "cpu_steal_percent", Value: t.Steal / total * 100, Labels: map[string]string{}},
)
}
}
// Core count
count, _ := cpu.Counts(true)
metrics = append(metrics, core.Metric{
Name: "cpu_cores",
Help: "Number of logical CPU cores",
Type: "gauge",
Value: float64(count),
Labels: map[string]string{},
})
return metrics, nil
}
// ─── Memory ─────────────────────────────────────────────────────────────────
func (r *SystemReader) readMemory(_ context.Context) ([]core.Metric, error) {
vm, err := mem.VirtualMemory()
if err != nil {
return nil, err
}
used := vm.Total - vm.Available
labels := map[string]string{}
metrics := []core.Metric{
{Name: "memory_total_bytes", Value: float64(vm.Total), Labels: labels},
{Name: "memory_used_bytes", Value: float64(used), Labels: labels},
{Name: "memory_available_bytes", Value: float64(vm.Available), Labels: labels},
{Name: "memory_buffers_bytes", Value: float64(vm.Buffers), Labels: labels},
{Name: "memory_cached_bytes", Value: float64(vm.Cached), Labels: labels},
{Name: "memory_used_percent", Value: vm.UsedPercent, Labels: labels},
}
// Swap
sw, err := mem.SwapMemory()
if err == nil {
metrics = append(metrics,
core.Metric{Name: "memory_swap_total_bytes", Value: float64(sw.Total), Labels: labels},
core.Metric{Name: "memory_swap_used_bytes", Value: float64(sw.Used), Labels: labels},
core.Metric{Name: "memory_swap_used_percent", Value: sw.UsedPercent, Labels: labels},
)
}
return metrics, nil
}
// ─── Disk I/O ───────────────────────────────────────────────────────────────
func (r *SystemReader) readDisk(_ context.Context) ([]core.Metric, error) {
ioCounters, err := disk.IOCounters()
if err != nil {
return nil, err
}
var metrics []core.Metric
for name, io := range ioCounters {
labels := map[string]string{"device": name}
metrics = append(metrics,
core.Metric{Name: "disk_read_bytes_total", Value: float64(io.ReadBytes), Labels: labels},
core.Metric{Name: "disk_write_bytes_total", Value: float64(io.WriteBytes), Labels: labels},
core.Metric{Name: "disk_reads_total", Value: float64(io.ReadCount), Labels: labels},
core.Metric{Name: "disk_writes_total", Value: float64(io.WriteCount), Labels: labels},
core.Metric{Name: "disk_io_time_seconds_total", Value: float64(io.IoTime) / 1000.0, Labels: labels},
core.Metric{Name: "disk_weighted_io_time_seconds_total", Value: float64(io.WeightedIO) / 1000.0, Labels: labels},
core.Metric{Name: "disk_read_time_seconds_total", Value: float64(io.ReadTime) / 1000.0, Labels: labels},
core.Metric{Name: "disk_write_time_seconds_total", Value: float64(io.WriteTime) / 1000.0, Labels: labels},
)
}
// Disk usage (partition-level)
parts, err := disk.Partitions(false)
if err == nil {
for _, p := range parts {
usage, err := disk.Usage(p.Mountpoint)
if err != nil {
continue
}
labels := map[string]string{"mountpoint": p.Mountpoint, "fstype": p.Fstype}
metrics = append(metrics,
core.Metric{Name: "disk_total_bytes", Value: float64(usage.Total), Labels: labels},
core.Metric{Name: "disk_used_bytes", Value: float64(usage.Used), Labels: labels},
core.Metric{Name: "disk_free_bytes", Value: float64(usage.Free), Labels: labels},
core.Metric{Name: "disk_used_percent", Value: usage.UsedPercent, Labels: labels},
)
}
}
return metrics, nil
}
// ─── Network ────────────────────────────────────────────────────────────────
func (r *SystemReader) readNetwork(_ context.Context) ([]core.Metric, error) {
counters, err := net.IOCounters(true) // per-interface
if err != nil {
return nil, err
}
var metrics []core.Metric
for _, c := range counters {
labels := map[string]string{"interface": c.Name}
metrics = append(metrics,
core.Metric{Name: "network_receive_bytes_total", Value: float64(c.BytesRecv), Labels: labels},
core.Metric{Name: "network_transmit_bytes_total", Value: float64(c.BytesSent), Labels: labels},
core.Metric{Name: "network_receive_packets_total", Value: float64(c.PacketsRecv), Labels: labels},
core.Metric{Name: "network_transmit_packets_total", Value: float64(c.PacketsSent), Labels: labels},
core.Metric{Name: "network_receive_errors_total", Value: float64(c.Errin), Labels: labels},
core.Metric{Name: "network_transmit_errors_total", Value: float64(c.Errout), Labels: labels},
core.Metric{Name: "network_receive_dropped_total", Value: float64(c.Dropin), Labels: labels},
core.Metric{Name: "network_transmit_dropped_total", Value: float64(c.Dropout), Labels: labels},
)
}
return metrics, nil
}
// ─── Load ───────────────────────────────────────────────────────────────────
func (r *SystemReader) readLoad(_ context.Context) ([]core.Metric, error) {
loads, err := load.Avg()
if err != nil {
return nil, err
}
return []core.Metric{
{Name: "load_1m", Value: loads.Load1, Labels: map[string]string{}},
{Name: "load_5m", Value: loads.Load5, Labels: map[string]string{}},
{Name: "load_15m", Value: loads.Load15, Labels: map[string]string{}},
}, nil
}
// ─── Uptime ─────────────────────────────────────────────────────────────────
func (r *SystemReader) readUptime(_ context.Context) ([]core.Metric, error) {
secs, err := host.Uptime()
if err != nil {
return nil, err
}
info, _ := host.Info()
metrics := []core.Metric{
{Name: "system_uptime_seconds", Value: float64(secs), Labels: map[string]string{}},
}
if info != nil {
metrics = append(metrics, core.Metric{
Name: "system_boot_time_seconds",
Value: float64(info.BootTime),
Labels: map[string]string{"os": info.OS, "platform": info.Platform, "kernel_version": info.KernelVersion},
})
}
return metrics, nil
}
// ─── Optional: top-N processes (expensive, off by default) ─────────────────
func (r *SystemReader) ReadTopProcesses(n int) ([]core.Metric, error) {
procs, err := process.Processes()
if err != nil {
return nil, err
}
type procInfo struct {
pid int32
name string
cpu float64
memRss uint64
}
var infos []procInfo
for _, p := range procs {
pid := p.Pid
name, _ := p.Name()
cpuPct, _ := p.CPUPercent()
memInfo, _ := p.MemoryInfo()
var rss uint64
if memInfo != nil {
rss = memInfo.RSS
}
infos = append(infos, procInfo{pid, name, cpuPct, rss})
}
// Simple top-N by CPU (avoid importing container/heap for a small slice)
// For large process lists you'd want sort.Slice or a heap.
if len(infos) > n {
// partial selection: just sort and truncate
sort.Slice(infos, func(i, j int) bool {
return infos[i].cpu > infos[j].cpu
})
infos = infos[:n]
}
var metrics []core.Metric
for _, p := range infos {
labels := map[string]string{
"pid": fmt.Sprintf("%d", p.pid),
"name": p.name,
}
metrics = append(metrics,
core.Metric{Name: "process_cpu_percent", Value: p.cpu, Labels: labels},
core.Metric{Name: "process_memory_rss_bytes", Value: float64(p.memRss), Labels: labels},
)
}
return metrics, nil
}