mirror of
https://github.com/ClusterCockpit/cc-metric-collector.git
synced 2024-12-27 07:39:05 +01:00
7840de7b82
* Add cpu_used (all-cpu_idle) to CpustatCollector * Update cc-metric-collector.init * Allow selection of timestamp precision in HttpSink * Add comment about precision requirement for cc-metric-store * Fix for API changes in gofish@v0.15.0 * Update requirements to latest version * Read sensors through redfish * Update golang toolchain to 1.21 * Remove stray error check * Update main config in configuration.md * Update Release action to use golang 1.22 stable release, no golang RPMs anymore * Update runonce action to use golang 1.22 stable release, no golang RPMs anymore * Update README.md Use right JSON type in configuration * Update sink's README * Test whether ipmitool or ipmi-sensors can be executed without errors * Little fixes to the prometheus sink (#115) * Add uint64 to float64 cast option * Add prometheus sink to the list of available sinks * Add aggregated counters by gpu for nvlink errors --------- Co-authored-by: Michael Schwarz <schwarz@uni-paderborn.de> * Ccmessage migration (#119) * Add cpu_used (all-cpu_idle) to CpustatCollector * Update cc-metric-collector.init * Allow selection of timestamp precision in HttpSink * Add comment about precision requirement for cc-metric-store * Fix for API changes in gofish@v0.15.0 * Update requirements to latest version * Read sensors through redfish * Update golang toolchain to 1.21 * Remove stray error check * Update main config in configuration.md * Update Release action to use golang 1.22 stable release, no golang RPMs anymore * Update runonce action to use golang 1.22 stable release, no golang RPMs anymore * Switch to CCMessage for all files. --------- Co-authored-by: Holger Obermaier <Holger.Obermaier@kit.edu> Co-authored-by: Holger Obermaier <40787752+ho-ob@users.noreply.github.com> * Switch to ccmessage also for latest additions in nvidiaMetric * New Message processor (#118) * Add cpu_used (all-cpu_idle) to CpustatCollector * Update cc-metric-collector.init * Allow selection of timestamp precision in HttpSink * Add comment about precision requirement for cc-metric-store * Fix for API changes in gofish@v0.15.0 * Update requirements to latest version * Read sensors through redfish * Update golang toolchain to 1.21 * Remove stray error check * Update main config in configuration.md * Update Release action to use golang 1.22 stable release, no golang RPMs anymore * Update runonce action to use golang 1.22 stable release, no golang RPMs anymore * New message processor to check whether a message should be dropped or manipulate it in flight * Create a copy of message before manipulation --------- Co-authored-by: Holger Obermaier <Holger.Obermaier@kit.edu> Co-authored-by: Holger Obermaier <40787752+ho-ob@users.noreply.github.com> * Update collector's Makefile and go.mod/sum files * Use message processor in router, all sinks and all receivers * Add support for credential file (NKEY) to NATS sink and receiver * Fix JSON keys in message processor configuration * Update docs for message processor, router and the default router config file * Add link to expr syntax and fix regex matching docs * Update sample collectors * Minor style change in collector manager * Some helpers for ccTopology * LIKWID collector: write log owner change only once * Fix for metrics without units and reduce debugging messages for messageProcessor * Use shorted hostname for hostname added by router * Define default port for NATS * CPUstat collector: only add unit for applicable metrics * Add precision option to all sinks using Influx's encoder * Add message processor to all sink documentation * Add units to documentation of cpustat collector --------- Co-authored-by: Holger Obermaier <Holger.Obermaier@kit.edu> Co-authored-by: Holger Obermaier <40787752+ho-ob@users.noreply.github.com> Co-authored-by: oscarminus <me@oscarminus.de> Co-authored-by: Michael Schwarz <schwarz@uni-paderborn.de>
223 lines
6.0 KiB
Go
223 lines
6.0 KiB
Go
package sinks
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"strings"
|
|
"sync"
|
|
|
|
lp "github.com/ClusterCockpit/cc-energy-manager/pkg/cc-message"
|
|
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
|
mp "github.com/ClusterCockpit/cc-metric-collector/pkg/messageProcessor"
|
|
"github.com/gorilla/mux"
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
"github.com/prometheus/client_golang/prometheus/promhttp"
|
|
)
|
|
|
|
type PrometheusSinkConfig struct {
|
|
defaultSinkConfig
|
|
Host string `json:"host,omitempty"`
|
|
Port string `json:"port"`
|
|
Path string `json:"path,omitempty"`
|
|
GroupAsNameSpace bool `json:"group_as_namespace,omitempty"`
|
|
// User string `json:"user,omitempty"`
|
|
// Password string `json:"password,omitempty"`
|
|
// FlushDelay string `json:"flush_delay,omitempty"`
|
|
}
|
|
|
|
type PrometheusSink struct {
|
|
sink
|
|
config PrometheusSinkConfig
|
|
labelMetrics map[string]*prometheus.GaugeVec
|
|
nodeMetrics map[string]prometheus.Gauge
|
|
promWg sync.WaitGroup
|
|
promServer *http.Server
|
|
}
|
|
|
|
func intToFloat64(input interface{}) (float64, error) {
|
|
switch value := input.(type) {
|
|
case float64:
|
|
return value, nil
|
|
case float32:
|
|
return float64(value), nil
|
|
case int:
|
|
return float64(value), nil
|
|
case int32:
|
|
return float64(value), nil
|
|
case int64:
|
|
return float64(value), nil
|
|
case uint64:
|
|
return float64(value), nil
|
|
}
|
|
return 0, errors.New("cannot cast value to float64")
|
|
}
|
|
|
|
func getLabelValue(metric lp.CCMessage) []string {
|
|
labelValues := []string{}
|
|
if tid, tidok := metric.GetTag("type-id"); tidok && metric.HasTag("type") {
|
|
labelValues = append(labelValues, tid)
|
|
}
|
|
if d, ok := metric.GetTag("device"); ok {
|
|
labelValues = append(labelValues, d)
|
|
} else if d, ok := metric.GetMeta("device"); ok {
|
|
labelValues = append(labelValues, d)
|
|
}
|
|
return labelValues
|
|
}
|
|
|
|
func getLabelNames(metric lp.CCMessage) []string {
|
|
labelNames := []string{}
|
|
if t, tok := metric.GetTag("type"); tok && metric.HasTag("type-id") {
|
|
labelNames = append(labelNames, t)
|
|
}
|
|
if _, ok := metric.GetTag("device"); ok {
|
|
labelNames = append(labelNames, "device")
|
|
} else if _, ok := metric.GetMeta("device"); ok {
|
|
labelNames = append(labelNames, "device")
|
|
}
|
|
return labelNames
|
|
}
|
|
|
|
func (s *PrometheusSink) newMetric(metric lp.CCMessage) error {
|
|
var value float64 = 0
|
|
name := metric.Name()
|
|
opts := prometheus.GaugeOpts{
|
|
Name: name,
|
|
}
|
|
labels := getLabelNames(metric)
|
|
labelValues := getLabelValue(metric)
|
|
if len(labels) > 0 && len(labels) != len(labelValues) {
|
|
return fmt.Errorf("cannot detect metric labels for metric %s", name)
|
|
}
|
|
|
|
if metricValue, ok := metric.GetField("value"); ok {
|
|
if floatValue, err := intToFloat64(metricValue); err == nil {
|
|
value = floatValue
|
|
} else {
|
|
return fmt.Errorf("metric %s with value '%v' cannot be casted to float64", name, metricValue)
|
|
}
|
|
}
|
|
if s.config.GroupAsNameSpace && metric.HasMeta("group") {
|
|
g, _ := metric.GetMeta("group")
|
|
opts.Namespace = strings.ToLower(g)
|
|
}
|
|
|
|
if len(labels) > 0 {
|
|
new := prometheus.NewGaugeVec(opts, labels)
|
|
new.WithLabelValues(labelValues...).Set(value)
|
|
s.labelMetrics[name] = new
|
|
prometheus.Register(new)
|
|
} else {
|
|
new := prometheus.NewGauge(opts)
|
|
new.Set(value)
|
|
s.nodeMetrics[name] = new
|
|
prometheus.Register(new)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *PrometheusSink) updateMetric(metric lp.CCMessage) error {
|
|
var value float64 = 0.0
|
|
name := metric.Name()
|
|
labelValues := getLabelValue(metric)
|
|
|
|
if metricValue, ok := metric.GetField("value"); ok {
|
|
if floatValue, err := intToFloat64(metricValue); err == nil {
|
|
value = floatValue
|
|
} else {
|
|
return fmt.Errorf("metric %s with value '%v' cannot be casted to float64", name, metricValue)
|
|
}
|
|
}
|
|
|
|
if len(labelValues) > 0 {
|
|
if _, ok := s.labelMetrics[name]; !ok {
|
|
err := s.newMetric(metric)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
s.labelMetrics[name].WithLabelValues(labelValues...).Set(value)
|
|
} else {
|
|
if _, ok := s.labelMetrics[name]; !ok {
|
|
err := s.newMetric(metric)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
s.nodeMetrics[name].Set(value)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *PrometheusSink) Write(m lp.CCMessage) error {
|
|
msg, err := s.mp.ProcessMessage(m)
|
|
if err == nil && msg != nil {
|
|
err = s.updateMetric(m)
|
|
}
|
|
return err
|
|
}
|
|
|
|
func (s *PrometheusSink) Flush() error {
|
|
return nil
|
|
}
|
|
|
|
func (s *PrometheusSink) Close() {
|
|
cclog.ComponentDebug(s.name, "CLOSE")
|
|
s.promServer.Shutdown(context.Background())
|
|
s.promWg.Wait()
|
|
}
|
|
|
|
func NewPrometheusSink(name string, config json.RawMessage) (Sink, error) {
|
|
s := new(PrometheusSink)
|
|
s.name = "PrometheusSink"
|
|
if len(config) > 0 {
|
|
d := json.NewDecoder(bytes.NewReader(config))
|
|
d.DisallowUnknownFields()
|
|
if err := d.Decode(&s.config); err != nil {
|
|
cclog.ComponentError(s.name, "Error reading config:", err.Error())
|
|
return nil, err
|
|
}
|
|
}
|
|
if len(s.config.Port) == 0 {
|
|
err := errors.New("not all configuration variables set required by PrometheusSink")
|
|
cclog.ComponentError(s.name, err.Error())
|
|
return nil, err
|
|
}
|
|
p, err := mp.NewMessageProcessor()
|
|
if err != nil {
|
|
return nil, fmt.Errorf("initialization of message processor failed: %v", err.Error())
|
|
}
|
|
s.mp = p
|
|
if len(s.config.MessageProcessor) > 0 {
|
|
err = p.FromConfigJSON(s.config.MessageProcessor)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed parsing JSON for message processor: %v", err.Error())
|
|
}
|
|
}
|
|
for _, k := range s.config.MetaAsTags {
|
|
s.mp.AddMoveMetaToTags("true", k, k)
|
|
}
|
|
s.labelMetrics = make(map[string]*prometheus.GaugeVec)
|
|
s.nodeMetrics = make(map[string]prometheus.Gauge)
|
|
s.promWg.Add(1)
|
|
go func() {
|
|
router := mux.NewRouter()
|
|
// Prometheus endpoint
|
|
router.Path("/" + s.config.Path).Handler(promhttp.Handler())
|
|
|
|
url := fmt.Sprintf("%s:%s", s.config.Host, s.config.Port)
|
|
cclog.ComponentDebug(s.name, "Serving Prometheus metrics at", fmt.Sprintf("%s:%s/%s", s.config.Host, s.config.Port, s.config.Path))
|
|
s.promServer = &http.Server{Addr: url, Handler: router}
|
|
err := s.promServer.ListenAndServe()
|
|
if err != nil && err.Error() != "http: Server closed" {
|
|
cclog.ComponentError(s.name, err.Error())
|
|
}
|
|
s.promWg.Done()
|
|
}()
|
|
return s, nil
|
|
}
|