xmtop/devices/remote.go

package devices

import (
	"bufio"
	"log"
	"net/http"
	"net/url"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/droundy/goopt"
)

var nameP *string
var remoteUrlP *string
var sleepP *string
var sleep time.Duration
var remoteLock sync.Mutex

// FIXME Widgets don't align values
// TODO remote network & disk aren't reported
// TODO network resiliency; I believe it currently crashes gotop when the network goes down
// TODO Replace custom decoder with https://github.com/prometheus/common/blob/master/expfmt/decode.go
// FIXME high CPU use when remote goes offline
// FIXME higher CPU use when using remote in general
func init() {
	nameP = goopt.String([]string{"--remote-name"}, "", "Remote: name of remote gotop")
	remoteUrlP = goopt.String([]string{"--remote-url"}, "", "Remote: URL of remote gotop")
	sleepP = goopt.String([]string{"--remote-refresh"}, "", "Remote: Frequency to refresh data, in seconds")

	RegisterStartup(startup)
}

type Remote struct {
	url     string
	refresh time.Duration
}

func startup(vars map[string]string) error {
	_cpuData = make(map[string]int)
	_tempData = make(map[string]int)
	_netData = make(map[string]float64)
	_diskData = make(map[string]float64)
	_memData = make(map[string]MemoryInfo)

	remoteLock = sync.Mutex{}
	remotes := parseConfig(vars)

	if remoteUrlP != nil && *remoteUrlP != "" {
		var name string
		r := Remote{
			url:     *remoteUrlP,
			refresh: 5 * time.Second,
		}
		if nameP == nil && *nameP != "" {
			name = *nameP
		} else {
			name = "Remote"
		}
		if sleepP == nil && *sleepP != "" {
			sleep, err := time.ParseDuration(*sleepP)
			if err == nil {
				r.refresh = sleep
			} else {
				log.Printf("invalid refresh duration %s for %s; using default", *sleepP, *remoteUrlP)
			}
		}
		remotes[name] = r
	}

	if len(remotes) == 0 {
		log.Println("Remote: no remote URL provided; disabling extension")
		return nil
	}

	RegisterTemp(updateTemp)
	RegisterMem(updateMem)
	RegisterCPU(updateUsage)

	// We need to know what we're dealing with, so the following code does two
	// things, one of them sneakily. It forks off background processes
	// to periodically pull data from remote sources and cache the results for
	// when the UI wants it. When it's run the first time, it sets up a WaitGroup
	// so that it can hold off returning until it's received data from the remote
	// so that the rest of the program knows how many cores, disks, etc. it needs
	// to set up UI elements for. After the first run, each process discards the
	// the wait group.
	w := &sync.WaitGroup{}
	for n, r := range remotes {
		n = n + "-"
		var u *url.URL
		w.Add(1)
		go func(name string, remote Remote, wg *sync.WaitGroup) {
			for {
				res, err := http.Get(remote.url)
				if err == nil {
					u, err = url.Parse(remote.url)
					if err == nil {
						if res.StatusCode == http.StatusOK {
							bi := bufio.NewScanner(res.Body)
							process(name, bi)
						} else {
							u.User = nil
							log.Printf("unsuccessful connection to %s: http status %s", u.String(), res.Status)
						}
					} else {
						log.Print("error processing remote URL")
					}
				} else {
				}
				res.Body.Close()
				if wg != nil {
					wg.Done()
					wg = nil
				}
				time.Sleep(remote.refresh)
			}
		}(n, r, w)
	}
	w.Wait()
	return nil
}

var (
	_cpuData  map[string]int
	_tempData map[string]int
	_netData  map[string]float64
	_diskData map[string]float64
	_memData  map[string]MemoryInfo
)

func process(host string, data *bufio.Scanner) {
	remoteLock.Lock()
	for data.Scan() {
		line := data.Text()
		if line[0] == '#' {
			continue
		}
		if line[0:6] != _gotop {
			continue
		}
		sub := line[6:]
		switch {
		case strings.HasPrefix(sub, _cpu): // int gotop_cpu_CPU0
			procInt(host, line, sub[4:], _cpuData)
		case strings.HasPrefix(sub, _temp): // int gotop_temp_acpitz
			procInt(host, line, sub[5:], _tempData)
		case strings.HasPrefix(sub, _net): // int gotop_net_recv
			parts := strings.Split(sub[5:], " ")
			if len(parts) < 2 {
				log.Printf(`bad data; not enough columns in "%s"`, line)
				continue
			}
			val, err := strconv.ParseFloat(parts[1], 64)
			if err != nil {
				log.Print(err)
				continue
			}
			_netData[host+parts[0]] = val
		case strings.HasPrefix(sub, _disk): // float % gotop_disk_:dev:mmcblk0p1
			parts := strings.Split(sub[5:], " ")
			if len(parts) < 2 {
				log.Printf(`bad data; not enough columns in "%s"`, line)
				continue
			}
			val, err := strconv.ParseFloat(parts[1], 64)
			if err != nil {
				log.Print(err)
				continue
			}
			_diskData[host+parts[0]] = val
		case strings.HasPrefix(sub, _mem): // float % gotop_memory_Main
			parts := strings.Split(sub[7:], " ")
			if len(parts) < 2 {
				log.Printf(`bad data; not enough columns in "%s"`, line)
				continue
			}
			val, err := strconv.ParseFloat(parts[1], 64)
			if err != nil {
				log.Print(err)
				continue
			}
			_memData[host+parts[0]] = MemoryInfo{
				Total:       100,
				Used:        uint64(100.0 / val),
				UsedPercent: val,
			}
		default:
			// NOP!  This is a metric we don't care about.
		}
	}
	remoteLock.Unlock()
}

func procInt(host, line, sub string, data map[string]int) {
	parts := strings.Split(sub, " ")
	if len(parts) < 2 {
		log.Printf(`bad data; not enough columns in "%s"`, line)
		return
	}
	val, err := strconv.Atoi(parts[1])
	if err != nil {
		log.Print(err)
		return
	}
	data[host+parts[0]] = val
}

func updateTemp(temps map[string]int) map[string]error {
	remoteLock.Lock()
	for name, val := range _tempData {
		temps[name] = val
	}
	remoteLock.Unlock()
	return nil
}

// FIXME The units are wrong: getting bytes, assuming they're %
func updateMem(mems map[string]MemoryInfo) map[string]error {
	remoteLock.Lock()
	for name, val := range _memData {
		mems[name] = val
	}
	remoteLock.Unlock()
	return nil
}

func updateUsage(cpus map[string]int, _ bool) map[string]error {
	remoteLock.Lock()
	for name, val := range _cpuData {
		cpus[name] = val
	}
	remoteLock.Unlock()
	return nil
}

func parseConfig(vars map[string]string) map[string]Remote {
	rv := make(map[string]Remote)
	for key, value := range vars {
		if strings.HasPrefix(key, "remote-") {
			parts := strings.Split(key, "-")
			if len(parts) == 2 {
				log.Printf("malformed Remote extension configuration '%s'; must be 'remote-NAME-url' or 'remote-NAME-refresh'", key)
				continue
			}
			name := parts[1]
			remote, ok := rv[name]
			if !ok {
				remote = Remote{}
			}
			if parts[2] == "url" {
				remote.url = value
			} else if parts[2] == "refresh" {
				sleep, err := strconv.Atoi(value)
				if err != nil {
					log.Printf("illegal Remote extension value for %s: '%s'.  Must be a duration in seconds, e.g. '2'", key, value)
					continue
				}
				remote.refresh = time.Duration(sleep) * time.Second
			} else {
				log.Printf("bad configuration option for Remote extension: '%s'; must be 'remote-NAME-url' or 'remote-NAME-refresh'", key)
				continue
			}
			rv[name] = remote
		}
	}
	return rv
}

const (
	_gotop = "gotop_"
	_cpu   = "cpu_"
	_temp  = "temp_"
	_net   = "net_"
	_disk  = "disk_"
	_mem   = "memory_"
)