sflow-cpu.go

package main

import (
	"github.com/PreetamJinka/sflow-go"
	"github.com/PreetamJinka/udpchan"

	"fmt"
	"log"
	"net"
	"time"
)

type counterState struct {
	lastUpdated time.Time
	previous    interface{}
	value       float32
}

var registry = map[string]map[string]counterState{}

func main() {
	c, err := udpchan.Listen(":6343", nil)
	if err != nil {
		log.Fatalln(err)
	}

	go func() {
		for _ = range time.Tick(time.Second * 5) {
			printRegistry()
		}
	}()

	for buf := range c {
		dgram := sflow.Decode(buf)
		ip := dgram.Header.IpAddress
		var records []sflow.Record

		for _, sample := range dgram.Samples {
			records = append(records, sample.GetRecords()...)
		}

		aggregateRecords(ip, records)
	}
}

func aggregateRecords(ip net.IP, records []sflow.Record) {
	registryKey := ip.String()

	for _, record := range records {
		switch record.RecordType() {
		case sflow.TypeHostCpuCounter:
			r := record.(sflow.HostCpuCounters)
			m, present := registry[registryKey]
			if !present {
				m = map[string]counterState{}
			}

			prev, present := m["cpu.user"]
			if present {
				now := time.Now()
				timeDelta := now.Sub(prev.lastUpdated)
				deriv := float64(r.CpuUser-prev.previous.(uint32)) / timeDelta.Seconds()
				prev.value = float32(deriv)
				prev.lastUpdated = now
				prev.previous = r.CpuUser
				m["cpu.user"] = prev
			} else {
				m["cpu.user"] = counterState{
					lastUpdated: time.Now(),
					previous:    r.CpuUser,
					value:       0,
				}
			}

			prev, present = m["cpu.sys"]
			if present {
				now := time.Now()
				timeDelta := now.Sub(prev.lastUpdated)
				deriv := float64(r.CpuSys-prev.previous.(uint32)) / timeDelta.Seconds()
				prev.value = float32(deriv)
				prev.lastUpdated = now
				prev.previous = r.CpuSys
				m["cpu.sys"] = prev
			} else {
				m["cpu.sys"] = counterState{
					lastUpdated: time.Now(),
					previous:    r.CpuSys,
					value:       0,
				}
			}

			prev, present = m["cpu.idle"]
			if present {
				now := time.Now()
				timeDelta := now.Sub(prev.lastUpdated)
				deriv := float64(r.CpuIdle-prev.previous.(uint32)) / timeDelta.Seconds()
				prev.value = float32(deriv)
				prev.lastUpdated = now
				prev.previous = r.CpuIdle
				m["cpu.idle"] = prev
			} else {
				m["cpu.idle"] = counterState{
					lastUpdated: time.Now(),
					previous:    r.CpuIdle,
					value:       0,
				}
			}

			prev, present = m["cpu.nice"]
			if present {
				now := time.Now()
				timeDelta := now.Sub(prev.lastUpdated)
				deriv := float64(r.CpuNice-prev.previous.(uint32)) / timeDelta.Seconds()
				prev.value = float32(deriv)
				prev.lastUpdated = now
				prev.previous = r.CpuNice
				m["cpu.nice"] = prev
			} else {
				m["cpu.nice"] = counterState{
					lastUpdated: time.Now(),
					previous:    r.CpuNice,
					value:       0,
				}
			}

			prev, present = m["cpu.wio"]
			if present {
				now := time.Now()
				timeDelta := now.Sub(prev.lastUpdated)
				deriv := float64(r.CpuWio-prev.previous.(uint32)) / timeDelta.Seconds()
				prev.value = float32(deriv)
				prev.lastUpdated = now
				prev.previous = r.CpuWio
				m["cpu.wio"] = prev
			} else {
				m["cpu.wio"] = counterState{
					lastUpdated: time.Now(),
					previous:    r.CpuWio,
					value:       0,
				}
			}

			registry[registryKey] = m
		}
	}
}

func printRegistry() {
	for host, metrics := range registry {
		//for metric, state := range metrics {
		//	fmt.Println(metric, state.value)
		//}

		user, ok := metrics["cpu.user"]
		if !ok {
			continue
		}
		sys, ok := metrics["cpu.sys"]
		if !ok {
			continue
		}
		idle, ok := metrics["cpu.idle"]
		if !ok {
			continue
		}
		nice, ok := metrics["cpu.nice"]
		if !ok {
			continue
		}
		wio, ok := metrics["cpu.wio"]
		if !ok {
			continue
		}

		totalTime := user.value + sys.value + idle.value +
			nice.value + wio.value

		if totalTime > 0 {
			fmt.Printf("%s: %.2f%%user\t%.2f%%sys\t%.2f%%wio\t%.2f%%nice\t%.2f%%idle\n", host, user.value/totalTime*100,
				sys.value/totalTime*100, wio.value/totalTime*100, nice.value/totalTime*100, idle.value/totalTime*100)
		}
	}
}