diff --git a/config/config.go b/config/config.go index 8ee3846..fba26c7 100644 --- a/config/config.go +++ b/config/config.go @@ -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"` diff --git a/config/config.yaml b/config/config.yaml index 5058811..5afc6ca 100644 --- a/config/config.yaml +++ b/config/config.yaml @@ -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 diff --git a/exporters/influxdb.go b/exporters/influxdb.go new file mode 100644 index 0000000..7b3649d --- /dev/null +++ b/exporters/influxdb.go @@ -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= +// +// 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) +} diff --git a/go.mod b/go.mod index 49fb603..964cc7a 100644 --- a/go.mod +++ b/go.mod @@ -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 ) diff --git a/go.sum b/go.sum index 1d0cde8..d9979cb 100644 --- a/go.sum +++ b/go.sum @@ -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= diff --git a/main.go b/main.go index c49b982..5ddd337 100644 --- a/main.go +++ b/main.go @@ -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)) } } diff --git a/readers/system.go b/readers/system.go new file mode 100644 index 0000000..28ee1bc --- /dev/null +++ b/readers/system.go @@ -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 +}