mon: add vm based metrics

Author: pnsafonov

pkg/monitoring.go

@@ -1,15 +1,17 @@
 package pkg
 
-import "github.com/pnsafonov/pind/pkg/monitoring"
+import (
+	"github.com/pnsafonov/pind/pkg/monitoring/mon_state"
+)
 
 func setMonitoringState(ctx *Context) error {
 	state := fillMonitoringState(ctx)
 	ctx.Monitoring.SetState(state)
 	return nil
 }
 
-func fillMonitoringState(ctx *Context) *monitoring.State {
-	state := monitoring.NewState()
+func fillMonitoringState(ctx *Context) *mon_state.State {
+	state := mon_state.NewState()
 
 	state.Pool.IdleLoad0 = ctx.pool.IdleLoad0
 	state.Pool.IdleLoad1 = ctx.pool.IdleLoad1
@@ -22,7 +24,7 @@ func fillMonitoringState(ctx *Context) *monitoring.State {
 	for i := 0; i < l0; i++ {
 		node := ctx.pool.Nodes[i]
 
-		nodeMon := &monitoring.PoolNode{
+		nodeMon := &mon_state.PoolNode{
 			Index:     i,
 			LoadFree0: node.LoadFree0,
 			LoadFree1: node.LoadFree1,
@@ -32,5 +34,22 @@ func fillMonitoringState(ctx *Context) *monitoring.State {
 		state.Pool.Nodes = append(state.Pool.Nodes, nodeMon)
 	}
 
+	for _, pinProc := range ctx.state.Procs {
+		if pinProc.ProcInfo.VmName == "" {
+			continue
+		}
+
+		numa0 := pinProc.GetNuma0()
+		proc := &mon_state.Proc{
+			VmName: pinProc.ProcInfo.VmName,
+			Time:   state.Time,
+			CPU:    pinProc.ProcInfo.cpu0,
+			Load:   pinProc.ProcInfo.load,
+			Numa0:  numa0,
+		}
+
+		state.Procs[pinProc.ProcInfo.VmName] = proc
+	}
+
 	return state
 }

pkg/monitoring/collector/proc.go

@@ -0,0 +1,72 @@
+package collector
+
+import (
+	"fmt"
+	"github.com/pnsafonov/pind/pkg/monitoring/mon_state"
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+type Proc struct {
+	VmName string
+	Proc   *mon_state.Proc
+
+	CPU   *prometheus.Desc
+	Load  *prometheus.Desc
+	Numa0 *prometheus.Desc
+}
+
+func NewProc(proc *mon_state.Proc) (*Proc, error) {
+	vmName := proc.VmName
+	if vmName == "" {
+		return nil, fmt.Errorf("vm name is required")
+	}
+
+	ent := &Proc{
+		VmName: vmName,
+		Proc:   proc,
+	}
+
+	labels0 := prometheus.Labels{
+		"vm_name": vmName,
+	}
+
+	ent.CPU = prometheus.NewDesc("pind_vm_cpu",
+		"vm cpu",
+		nil, labels0,
+	)
+	ent.Load = prometheus.NewDesc("pind_vm_load",
+		"is vm on load",
+		nil, labels0,
+	)
+	ent.Numa0 = prometheus.NewDesc("pind_vm_numa",
+		"assigned numa for vm, -1 for not assigned",
+		nil, labels0,
+	)
+
+	return ent, nil
+}
+
+func (x *Proc) SetProc(proc *mon_state.Proc) {
+	x.Proc = proc
+}
+
+func (x *Proc) Describe(ch chan<- *prometheus.Desc) {
+	ch <- x.CPU
+	ch <- x.Load
+	ch <- x.Numa0
+}
+
+func (x *Proc) Collect(ch chan<- prometheus.Metric) {
+	m0 := prometheus.MustNewConstMetric(x.CPU, prometheus.GaugeValue, x.Proc.CPU)
+	m1 := prometheus.MustNewConstMetric(x.Load, prometheus.GaugeValue, x.Proc.GetLoad0())
+	m2 := prometheus.MustNewConstMetric(x.Numa0, prometheus.GaugeValue, float64(x.Proc.Numa0))
+
+	t0 := x.Proc.Time
+	mt0 := prometheus.NewMetricWithTimestamp(t0, m0)
+	mt1 := prometheus.NewMetricWithTimestamp(t0, m1)
+	mt2 := prometheus.NewMetricWithTimestamp(t0, m2)
+
+	ch <- mt0
+	ch <- mt1
+	ch <- mt2
+}

pkg/monitoring/collector.go renamed to pkg/monitoring/collector/static.go

@@ -1,37 +1,40 @@
-package monitoring
+package collector
 
 import (
+	"github.com/pnsafonov/pind/pkg/monitoring/mon_state"
 	"github.com/prometheus/client_golang/prometheus"
 	"strconv"
 )
 
-type StaticCollector struct {
-	PoolCollector *PoolCollector
+// Static - статический Collector, регистрируется при старте
+// далее для него не делается unregister
+type Static struct {
+	PoolCollector *Pool
 
-	State *State
+	State *mon_state.State
 }
 
-type PoolCollector struct {
+type Pool struct {
 	IdleLoad0 *prometheus.Desc
 	IdleLoad1 *prometheus.Desc
 	LoadFree0 *prometheus.Desc
 	LoadFree1 *prometheus.Desc
 	LoadUsed0 *prometheus.Desc
 	LoadUsed1 *prometheus.Desc
 
-	Nodes []*PoolNodeCollector
+	Nodes []*PoolNode
 }
 
-type PoolNodeCollector struct {
+type PoolNode struct {
 	Index     int
 	LoadFree0 *prometheus.Desc
 	LoadFree1 *prometheus.Desc
 	LoadUsed0 *prometheus.Desc
 	LoadUsed1 *prometheus.Desc
 }
 
-func NewPoolCollector(nodesCount int) *PoolCollector {
-	ent := &PoolCollector{}
+func NewPool(nodesCount int) *Pool {
+	ent := &Pool{}
 
 	ent.IdleLoad0 = prometheus.NewDesc("pind_pool_idle_load0",
 		"idle cores load, like 400, 600, 800 %",
@@ -59,7 +62,7 @@ func NewPoolCollector(nodesCount int) *PoolCollector {
 	)
 
 	for i := 0; i < nodesCount; i++ {
-		node := &PoolNodeCollector{
+		node := &PoolNode{
 			Index: i,
 		}
 
@@ -90,17 +93,17 @@ func NewPoolCollector(nodesCount int) *PoolCollector {
 	return ent
 }
 
-func NewStaticCollector(nodesCount int) *StaticCollector {
-	pool := NewPoolCollector(nodesCount)
+func NewStatic(nodesCount int) *Static {
+	pool := NewPool(nodesCount)
 
-	ent := &StaticCollector{
+	ent := &Static{
 		PoolCollector: pool,
 	}
 
 	return ent
 }
 
-func (x *StaticCollector) Describe(ch chan<- *prometheus.Desc) {
+func (x *Static) Describe(ch chan<- *prometheus.Desc) {
 	ch <- x.PoolCollector.IdleLoad0
 	ch <- x.PoolCollector.IdleLoad1
 	ch <- x.PoolCollector.LoadFree0
@@ -117,7 +120,7 @@ func (x *StaticCollector) Describe(ch chan<- *prometheus.Desc) {
 
 }
 
-func (x *StaticCollector) Collect(ch chan<- prometheus.Metric) {
+func (x *Static) Collect(ch chan<- prometheus.Metric) {
 	m0 := prometheus.MustNewConstMetric(x.PoolCollector.IdleLoad0, prometheus.GaugeValue, x.State.Pool.IdleLoad0)
 	m1 := prometheus.MustNewConstMetric(x.PoolCollector.IdleLoad1, prometheus.GaugeValue, x.State.Pool.IdleLoad1)
 	m2 := prometheus.MustNewConstMetric(x.PoolCollector.LoadFree0, prometheus.GaugeValue, x.State.Pool.LoadFree0)

pkg/monitoring/state.go renamed to pkg/monitoring/mon_state/state.go

@@ -1,10 +1,12 @@
-package monitoring
+package mon_state
 
 import "time"
 
 type State struct {
 	Time time.Time
 	Pool *Pool
+	//Procs []*Proc
+	Procs map[string]*Proc
 }
 
 type Pool struct {
@@ -26,12 +28,28 @@ type PoolNode struct {
 	LoadUsed1 float64
 }
 
+type Proc struct {
+	VmName string
+	Time   time.Time
+	CPU    float64
+	Load   bool
+	Numa0  int
+}
+
 func NewState() *State {
 	now := time.Now()
 	pool := &Pool{}
 
 	return &State{
-		Time: now,
-		Pool: pool,
+		Time:  now,
+		Pool:  pool,
+		Procs: make(map[string]*Proc),
+	}
+}
+
+func (x *Proc) GetLoad0() float64 {
+	if x.Load {
+		return 1
 	}
+	return 0
 }

pkg/monitoring/monitoring.go

@@ -1,7 +1,10 @@
 package monitoring
 
 import (
+	"fmt"
 	"github.com/pnsafonov/pind/pkg/config"
+	"github.com/pnsafonov/pind/pkg/monitoring/collector"
+	"github.com/pnsafonov/pind/pkg/monitoring/mon_state"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	log "github.com/sirupsen/logrus"
@@ -14,15 +17,17 @@ type Monitoring struct {
 	server          *http.Server
 	gatherer        prometheus.Gatherer
 	registerer      prometheus.Registerer
-	staticCollector *StaticCollector
+	staticCollector *collector.Static
+	procCollectors  map[string]*collector.Proc
 }
 
 func NewMonitoring(config *config.Monitoring, numaNodesCount int) *Monitoring {
-	collector := NewStaticCollector(numaNodesCount)
+	static0 := collector.NewStatic(numaNodesCount)
 
 	ent := &Monitoring{
 		config:          config,
-		staticCollector: collector,
+		staticCollector: static0,
+		procCollectors:  make(map[string]*collector.Proc),
 	}
 	return ent
 }
@@ -71,6 +76,46 @@ func (x *Monitoring) GoServe() error {
 	return nil
 }
 
-func (x *Monitoring) SetState(state *State) {
+func (x *Monitoring) SetState(state *mon_state.State) {
 	x.staticCollector.State = state
+
+	// для процесса регистрируем метрику
+	for _, proc := range state.Procs {
+		_ = x.setProc(proc)
+	}
+
+	// процесс завершился, убираем метрику
+	for vmName, procCollector := range x.procCollectors {
+		_, ok := state.Procs[vmName]
+		if !ok {
+			_ = x.registerer.Unregister(procCollector)
+			delete(x.procCollectors, vmName)
+		}
+	}
+}
+
+func (x *Monitoring) setProc(proc *mon_state.Proc) error {
+	if proc.VmName == "" {
+		return fmt.Errorf("proc.VmName is empty")
+	}
+
+	var err error
+	procCollector, ok := x.procCollectors[proc.VmName]
+	if !ok {
+		procCollector, err = collector.NewProc(proc)
+		if err != nil {
+			return err
+		}
+
+		err = x.registerer.Register(procCollector)
+		if err != nil {
+			log.Errorf("setProc, prometheus.Register err: %v", err)
+			return err
+		}
+
+		x.procCollectors[proc.VmName] = procCollector
+	}
+
+	procCollector.SetProc(proc)
+	return nil
 }

pkg/pin.go

@@ -261,6 +261,15 @@ func getThreadsCount0(threads map[int]*PinThread, selection ThreadSelection) int
 	return count
 }
 
+// GetNuma0 - returns assigned numa index.
+// If no numa assigned returns -1.
+func (x *PinProc) GetNuma0() int {
+	if x.Node == nil {
+		return -1
+	}
+	return x.Node.Index
+}
+
 func (x *PinProc) ContainsThread(selection ThreadSelection) bool {
 	procInfo := x
 	for _, threadInfo := range procInfo.Threads {