cc-metric-collector/collectors/ipmiMetric.go

package collectors

import (
	"bufio"
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"os/exec"
	"strconv"
	"strings"
	"time"

	lp "github.com/ClusterCockpit/cc-lib/ccMessage"
	cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
)

const IPMISENSORS_PATH = `ipmi-sensors`

type IpmiCollectorConfig struct {
	ExcludeDevices  []string `json:"exclude_devices"`
	IpmitoolPath    string   `json:"ipmitool_path"`
	IpmisensorsPath string   `json:"ipmisensors_path"`
	ExcludeMetrics  []string `json:"exclude_metrics,omitempty"`
	OnlyMetrics     []string `json:"only_metrics,omitempty"`
}

type IpmiCollector struct {
	metricCollector
	config      IpmiCollectorConfig
	ipmitool    string
	ipmisensors string
}

// shouldOutput checks whether a metric should be forwarded based on only_metrics/exclude_metrics.
func (m *IpmiCollector) shouldOutput(metricName string) bool {
	if len(m.config.OnlyMetrics) > 0 {
		for _, name := range m.config.OnlyMetrics {
			if name == metricName {
				return true
			}
		}
		return false
	}
	for _, name := range m.config.ExcludeMetrics {
		if name == metricName {
			return false
		}
	}
	return true
}

func (m *IpmiCollector) Init(config json.RawMessage) error {
	if m.init {
		return nil
	}

	m.name = "IpmiCollector"
	m.setup()
	m.parallel = true
	m.meta = map[string]string{
		"source": m.name,
		"group":  "IPMI",
	}
	// Default paths:
	m.config.IpmitoolPath = "ipmitool"
	m.config.IpmisensorsPath = "ipmi-sensors"
	if len(config) > 0 {
		err := json.Unmarshal(config, &m.config)
		if err != nil {
			return err
		}
	}
	// Check ipmitool: test with "-V" to verify its existence.
	p, err := exec.LookPath(m.config.IpmitoolPath)
	if err == nil {
		command := exec.Command(p, "-V")
		err := command.Run()
		if err != nil {
			cclog.ComponentError(m.name, fmt.Sprintf("Failed to execute %s -V: %v", p, err.Error()))
			m.ipmitool = ""
		} else {
			m.ipmitool = p
		}
	}
	// Check ipmi-sensors executable.
	p, err = exec.LookPath(m.config.IpmisensorsPath)
	if err == nil {
		command := exec.Command(p)
		err := command.Run()
		if err != nil {
			cclog.ComponentError(m.name, fmt.Sprintf("Failed to execute %s: %v", p, err.Error()))
			m.ipmisensors = ""
		} else {
			m.ipmisensors = p
		}
	}
	if len(m.ipmitool) == 0 && len(m.ipmisensors) == 0 {
		return errors.New("no usable IPMI reader found")
	}

	m.init = true
	return nil
}

func (m *IpmiCollector) readIpmiTool(cmd string, output chan lp.CCMessage) {
	command := exec.Command(cmd, "sensor")
	stdout, _ := command.StdoutPipe()
	errBuf := new(bytes.Buffer)
	command.Stderr = errBuf

	if err := command.Start(); err != nil {
		cclog.ComponentError(m.name, fmt.Sprintf("readIpmiTool(): Failed to start command \"%s\": %v", command.String(), err))
		return
	}

	scanner := bufio.NewScanner(stdout)
	for scanner.Scan() {
		lv := strings.Split(scanner.Text(), "|")
		if len(lv) < 3 {
			continue
		}
		v, err := strconv.ParseFloat(strings.TrimSpace(lv[1]), 64)
		if err == nil {
			name := strings.ToLower(strings.Replace(strings.TrimSpace(lv[0]), " ", "_", -1))
			if !m.shouldOutput(name) {
				continue
			}
			unit := strings.TrimSpace(lv[2])
			// Standardize unit names.
			if unit == "Volts" {
				unit = "Volts"
			} else if unit == "degrees C" {
				unit = "degC"
			} else if unit == "degrees F" {
				unit = "degF"
			} else if unit == "Watts" {
				unit = "Watts"
			}
			y, err := lp.NewMessage(name, map[string]string{"type": "node"}, m.meta, map[string]interface{}{"value": v}, time.Now())
			if err == nil {
				y.AddMeta("unit", unit)
				output <- y
			}
		}
	}

	if err := command.Wait(); err != nil {
		errMsg, _ := io.ReadAll(errBuf)
		cclog.ComponentError(m.name, fmt.Sprintf("readIpmiTool(): Failed to wait for command \"%s\": %v\n", command.String(), err))
		cclog.ComponentError(m.name, fmt.Sprintf("readIpmiTool(): command stderr: \"%s\"\n", strings.TrimSpace(string(errMsg))))
		return
	}
}

func (m *IpmiCollector) readIpmiSensors(cmd string, output chan lp.CCMessage) {
	command := exec.Command(cmd, "--comma-separated-output", "--sdr-cache-recreate")
	command.Wait()
	stdout, err := command.Output()
	if err != nil {
		log.Print(err)
		return
	}

	ll := strings.Split(string(stdout), "\n")
	for _, line := range ll {
		lv := strings.Split(line, ",")
		if len(lv) > 3 {
			v, err := strconv.ParseFloat(lv[3], 64)
			if err == nil {
				name := strings.ToLower(strings.Replace(lv[1], " ", "_", -1))
				if !m.shouldOutput(name) {
					continue
				}
				y, err := lp.NewMessage(name, map[string]string{"type": "node"}, m.meta, map[string]interface{}{"value": v}, time.Now())
				if err == nil {
					if len(lv) > 4 {
						y.AddMeta("unit", lv[4])
					}
					output <- y
				}
			}
		}
	}
}

func (m *IpmiCollector) Read(interval time.Duration, output chan lp.CCMessage) {
	if !m.init {
		return
	}

	if len(m.config.IpmitoolPath) > 0 {
		m.readIpmiTool(m.config.IpmitoolPath, output)
	} else if len(m.config.IpmisensorsPath) > 0 {
		m.readIpmiSensors(m.config.IpmisensorsPath, output)
	}
}

func (m *IpmiCollector) Close() {
	m.init = false
}