mirror of
https://github.com/ClusterCockpit/cc-metric-collector.git
synced 2024-11-10 12:37:25 +01:00
be20f956c2
* InfiniBandCollector: Scale raw readings from octets to bytes * Fix clock frequency coming from LikwidCollector and update docs * Build DEB package for Ubuntu 20.04 for releases * Fix memstat collector with numa_stats option * Remove useless prints from MemstatCollector * Replace ioutils with os and io (#87) * Use lower case for error strings in RocmSmiCollector * move maybe-usable-by-other-cc-components to pkg. Fix all files to use the new paths (#88) * Add collector for monitoring the execution of cc-metric-collector itself (#81) * Add collector to monitor execution of cc-metric-collector itself * Register SelfCollector * Fix import paths for moved packages
123 lines
2.9 KiB
Go
123 lines
2.9 KiB
Go
package receivers
|
|
|
|
import (
|
|
"bufio"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"net/http"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
|
|
cclog "github.com/ClusterCockpit/cc-metric-collector/pkg/ccLogger"
|
|
lp "github.com/ClusterCockpit/cc-metric-collector/pkg/ccMetric"
|
|
)
|
|
|
|
type PrometheusReceiverConfig struct {
|
|
defaultReceiverConfig
|
|
Addr string `json:"address"`
|
|
Port string `json:"port"`
|
|
Path string `json:"path"`
|
|
Interval string `json:"interval"`
|
|
SSL bool `json:"ssl"`
|
|
}
|
|
|
|
type PrometheusReceiver struct {
|
|
receiver
|
|
meta map[string]string
|
|
config PrometheusReceiverConfig
|
|
interval time.Duration
|
|
done chan bool
|
|
wg sync.WaitGroup
|
|
ticker *time.Ticker
|
|
uri string
|
|
}
|
|
|
|
func (r *PrometheusReceiver) Start() {
|
|
cclog.ComponentDebug(r.name, "START", r.uri)
|
|
r.wg.Add(1)
|
|
|
|
r.ticker = time.NewTicker(r.interval)
|
|
go func() {
|
|
for {
|
|
select {
|
|
case <-r.done:
|
|
r.wg.Done()
|
|
return
|
|
case t := <-r.ticker.C:
|
|
resp, err := http.Get(r.uri)
|
|
if err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
scanner := bufio.NewScanner(resp.Body)
|
|
for scanner.Scan() {
|
|
line := scanner.Text()
|
|
if strings.HasPrefix(line, "#") {
|
|
continue
|
|
}
|
|
lineSplit := strings.Fields(line)
|
|
// separate metric name from tags (labels in Prometheus)
|
|
tags := map[string]string{}
|
|
name := lineSplit[0]
|
|
if sindex := strings.Index(name, "{"); sindex >= 0 {
|
|
eindex := strings.Index(name, "}")
|
|
for _, kv := range strings.Split(name[sindex+1:eindex], ",") {
|
|
eq := strings.Index(kv, "=")
|
|
tags[kv[0:eq]] = strings.Trim(kv[eq+1:], "\"")
|
|
}
|
|
name = lineSplit[0][0:sindex]
|
|
}
|
|
value, err := strconv.ParseFloat(lineSplit[1], 64)
|
|
if err == nil {
|
|
y, err := lp.New(name, tags, r.meta, map[string]interface{}{"value": value}, t)
|
|
if err == nil {
|
|
r.sink <- y
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}()
|
|
}
|
|
|
|
func (r *PrometheusReceiver) Close() {
|
|
cclog.ComponentDebug(r.name, "CLOSE")
|
|
r.done <- true
|
|
r.wg.Wait()
|
|
}
|
|
|
|
func NewPrometheusReceiver(name string, config json.RawMessage) (Receiver, error) {
|
|
r := new(PrometheusReceiver)
|
|
r.name = fmt.Sprintf("PrometheusReceiver(%s)", name)
|
|
if len(config) > 0 {
|
|
err := json.Unmarshal(config, &r.config)
|
|
if err != nil {
|
|
cclog.ComponentError(r.name, "Error reading config:", err.Error())
|
|
return nil, err
|
|
}
|
|
}
|
|
if len(r.config.Addr) == 0 ||
|
|
len(r.config.Port) == 0 ||
|
|
len(r.config.Interval) == 0 {
|
|
return nil, errors.New("not all configuration variables set required by PrometheusReceiver (address and port)")
|
|
}
|
|
if len(r.config.Interval) > 0 {
|
|
t, err := time.ParseDuration(r.config.Interval)
|
|
if err == nil {
|
|
r.interval = t
|
|
}
|
|
}
|
|
r.meta = map[string]string{"source": r.name}
|
|
proto := "http"
|
|
if r.config.SSL {
|
|
proto = "https"
|
|
}
|
|
r.uri = fmt.Sprintf("%s://%s:%s/%s", proto, r.config.Addr, r.config.Port, r.config.Path)
|
|
return r, nil
|
|
}
|