cc-backend/internal/metricdata/prometheus.go

448 lines
12 KiB
Go
Raw Permalink Normal View History

2022-12-08 13:51:44 +01:00
// Copyright (C) 2022 DKRZ
// All rights reserved.
// Use of this source code is governed by a MIT-style
// license that can be found in the LICENSE file.
package metricdata
import (
2023-02-15 11:50:51 +01:00
"bytes"
2022-12-08 13:51:44 +01:00
"context"
"encoding/json"
2023-02-15 11:50:51 +01:00
"errors"
2022-12-08 13:51:44 +01:00
"fmt"
"math"
2023-02-15 11:50:51 +01:00
"net/http"
"os"
2022-12-08 13:51:44 +01:00
"regexp"
2023-02-15 11:50:51 +01:00
"sort"
"strings"
2022-12-08 13:51:44 +01:00
"sync"
2023-02-15 11:50:51 +01:00
"text/template"
"time"
2022-12-08 13:51:44 +01:00
"github.com/ClusterCockpit/cc-backend/pkg/archive"
"github.com/ClusterCockpit/cc-backend/pkg/log"
2023-02-15 11:50:51 +01:00
"github.com/ClusterCockpit/cc-backend/pkg/schema"
2022-12-08 13:51:44 +01:00
promapi "github.com/prometheus/client_golang/api"
2023-02-15 11:50:51 +01:00
promv1 "github.com/prometheus/client_golang/api/prometheus/v1"
2022-12-08 13:51:44 +01:00
promcfg "github.com/prometheus/common/config"
promm "github.com/prometheus/common/model"
)
type PrometheusDataRepositoryConfig struct {
2023-02-15 11:50:51 +01:00
Url string `json:"url"`
Username string `json:"username,omitempty"`
Suffix string `json:"suffix,omitempty"`
Templates map[string]string `json:"query-templates"`
2022-12-08 13:51:44 +01:00
}
type PrometheusDataRepository struct {
2023-02-15 11:50:51 +01:00
client promapi.Client
queryClient promv1.API
suffix string
templates map[string]*template.Template
2022-12-08 13:51:44 +01:00
}
type PromQLArgs struct {
2023-02-15 11:50:51 +01:00
Nodes string
2022-12-08 13:51:44 +01:00
}
type Trie map[rune]Trie
var logOnce sync.Once
func contains(s []schema.MetricScope, str schema.MetricScope) bool {
for _, v := range s {
if v == str {
return true
}
}
return false
}
func MinMaxMean(data []schema.Float) (float64, float64, float64) {
if len(data) == 0 {
2023-02-15 11:50:51 +01:00
return 0.0, 0.0, 0.0
2022-12-08 13:51:44 +01:00
}
min := math.MaxFloat64
max := -math.MaxFloat64
var sum float64
var n float64
for _, val := range data {
if val.IsNaN() {
continue
}
sum += float64(val)
n += 1
2023-02-15 11:50:51 +01:00
if float64(val) > max {
max = float64(val)
}
if float64(val) < min {
min = float64(val)
}
2022-12-08 13:51:44 +01:00
}
return min, max, sum / n
}
// Rewritten from
// https://github.com/ermanh/trieregex/blob/master/trieregex/trieregex.py
func nodeRegex(nodes []string) string {
2023-02-15 11:50:51 +01:00
root := Trie{}
// add runes of each compute node to trie
for _, node := range nodes {
_trie := root
for _, c := range node {
if _, ok := _trie[c]; !ok {
_trie[c] = Trie{}
}
_trie = _trie[c]
}
_trie['*'] = Trie{}
}
// recursively build regex from rune trie
var trieRegex func(trie Trie, reset bool) string
trieRegex = func(trie Trie, reset bool) string {
if reset == true {
trie = root
}
if len(trie) == 0 {
return ""
}
if len(trie) == 1 {
for key, _trie := range trie {
if key == '*' {
return ""
}
return regexp.QuoteMeta(string(key)) + trieRegex(_trie, false)
}
} else {
sequences := []string{}
for key, _trie := range trie {
if key != '*' {
sequences = append(sequences, regexp.QuoteMeta(string(key))+trieRegex(_trie, false))
}
}
sort.Slice(sequences, func(i, j int) bool {
return (-len(sequences[i]) < -len(sequences[j])) || (sequences[i] < sequences[j])
})
var result string
// single edge from this tree node
if len(sequences) == 1 {
result = sequences[0]
if len(result) > 1 {
result = "(?:" + result + ")"
}
// multiple edges, each length 1
} else if s := strings.Join(sequences, ""); len(s) == len(sequences) {
// char or numeric range
if len(s)-1 == int(s[len(s)-1])-int(s[0]) {
result = fmt.Sprintf("[%c-%c]", s[0], s[len(s)-1])
// char or numeric set
} else {
result = "[" + s + "]"
}
// multiple edges of different lengths
} else {
result = "(?:" + strings.Join(sequences, "|") + ")"
}
if _, ok := trie['*']; ok {
result += "?"
}
return result
}
return ""
}
return trieRegex(root, true)
2022-12-08 13:51:44 +01:00
}
func (pdb *PrometheusDataRepository) Init(rawConfig json.RawMessage) error {
var config PrometheusDataRepositoryConfig
// parse config
if err := json.Unmarshal(rawConfig, &config); err != nil {
log.Warn("Error while unmarshaling raw json config")
2022-12-08 13:51:44 +01:00
return err
}
// support basic authentication
var rt http.RoundTripper = nil
if prom_pw := os.Getenv("PROMETHEUS_PASSWORD"); prom_pw != "" && config.Username != "" {
prom_pw := promcfg.Secret(prom_pw)
rt = promcfg.NewBasicAuthRoundTripper(config.Username, prom_pw, "", promapi.DefaultRoundTripper)
} else {
if config.Username != "" {
return errors.New("METRICDATA/PROMETHEUS > Prometheus username provided, but PROMETHEUS_PASSWORD not set.")
2022-12-08 13:51:44 +01:00
}
}
// init client
client, err := promapi.NewClient(promapi.Config{
2023-02-15 11:50:51 +01:00
Address: config.Url,
2022-12-08 13:51:44 +01:00
RoundTripper: rt,
})
if err != nil {
log.Error("Error while initializing new prometheus client")
2022-12-08 13:51:44 +01:00
return err
}
// init query client
pdb.client = client
pdb.queryClient = promv1.NewAPI(pdb.client)
// site config
pdb.suffix = config.Suffix
// init query templates
pdb.templates = make(map[string]*template.Template)
for metric, templ := range config.Templates {
pdb.templates[metric], err = template.New(metric).Parse(templ)
if err == nil {
log.Debugf("Added PromQL template for %s: %s", metric, templ)
2022-12-08 13:51:44 +01:00
} else {
log.Warnf("Failed to parse PromQL template %s for metric %s", templ, metric)
2022-12-08 13:51:44 +01:00
}
}
return nil
}
// TODO: respect scope argument
func (pdb *PrometheusDataRepository) FormatQuery(
metric string,
scope schema.MetricScope,
nodes []string,
cluster string) (string, error) {
args := PromQLArgs{}
if len(nodes) > 0 {
args.Nodes = fmt.Sprintf("(%s)%s", nodeRegex(nodes), pdb.suffix)
} else {
args.Nodes = fmt.Sprintf(".*%s", pdb.suffix)
}
buf := &bytes.Buffer{}
if templ, ok := pdb.templates[metric]; ok {
err := templ.Execute(buf, args)
if err != nil {
return "", errors.New(fmt.Sprintf("METRICDATA/PROMETHEUS > Error compiling template %v", templ))
2022-12-08 13:51:44 +01:00
} else {
query := buf.String()
log.Debugf("PromQL: %s", query)
2022-12-08 13:51:44 +01:00
return query, nil
}
} else {
return "", errors.New(fmt.Sprintf("METRICDATA/PROMETHEUS > No PromQL for metric %s configured.", metric))
2022-12-08 13:51:44 +01:00
}
}
// Convert PromAPI row to CC schema.Series
func (pdb *PrometheusDataRepository) RowToSeries(
2023-02-15 11:50:51 +01:00
from time.Time,
step int64,
steps int64,
row *promm.SampleStream) schema.Series {
ts := from.Unix()
hostname := strings.TrimSuffix(string(row.Metric["exported_instance"]), pdb.suffix)
// init array of expected length with NaN
values := make([]schema.Float, steps+1)
for i, _ := range values {
values[i] = schema.NaN
2022-12-08 13:51:44 +01:00
}
2023-02-15 11:50:51 +01:00
// copy recorded values from prom sample pair
for _, v := range row.Values {
idx := (v.Timestamp.Unix() - ts) / step
values[idx] = schema.Float(v.Value)
}
min, max, mean := MinMaxMean(values)
// output struct
return schema.Series{
Hostname: hostname,
Data: values,
2023-04-11 16:26:09 +02:00
Statistics: schema.MetricStatistics{
2023-02-15 11:50:51 +01:00
Avg: mean,
Min: min,
Max: max,
},
}
}
2022-12-08 13:51:44 +01:00
func (pdb *PrometheusDataRepository) LoadData(
job *schema.Job,
metrics []string,
scopes []schema.MetricScope,
ctx context.Context) (schema.JobData, error) {
// TODO respect requested scope
2023-02-15 11:50:51 +01:00
if len(scopes) == 0 || !contains(scopes, schema.MetricScopeNode) {
2022-12-08 13:51:44 +01:00
scopes = append(scopes, schema.MetricScopeNode)
}
jobData := make(schema.JobData)
// parse job specs
nodes := make([]string, len(job.Resources))
for i, resource := range job.Resources {
nodes[i] = resource.Hostname
}
from := job.StartTime
to := job.StartTime.Add(time.Duration(job.Duration) * time.Second)
for _, scope := range scopes {
if scope != schema.MetricScopeNode {
2023-02-15 11:50:51 +01:00
logOnce.Do(func() {
log.Infof("Scope '%s' requested, but not yet supported: Will return 'node' scope only.", scope)
})
2022-12-08 13:51:44 +01:00
continue
}
for _, metric := range metrics {
metricConfig := archive.GetMetricConfig(job.Cluster, metric)
if metricConfig == nil {
log.Warnf("Error in LoadData: Metric %s for cluster %s not configured", metric, job.Cluster)
return nil, errors.New("Prometheus config error")
2022-12-08 13:51:44 +01:00
}
query, err := pdb.FormatQuery(metric, scope, nodes, job.Cluster)
if err != nil {
log.Warn("Error while formatting prometheus query")
2022-12-08 13:51:44 +01:00
return nil, err
}
// ranged query over all job nodes
r := promv1.Range{
2023-02-15 11:50:51 +01:00
Start: from,
End: to,
Step: time.Duration(metricConfig.Timestep * 1e9),
}
2022-12-08 13:51:44 +01:00
result, warnings, err := pdb.queryClient.QueryRange(ctx, query, r)
if err != nil {
log.Errorf("Prometheus query error in LoadData: %v\nQuery: %s", err, query)
return nil, errors.New("Prometheus query error")
2022-12-08 13:51:44 +01:00
}
if len(warnings) > 0 {
log.Warnf("Warnings: %v\n", warnings)
2022-12-08 13:51:44 +01:00
}
// init data structures
if _, ok := jobData[metric]; !ok {
jobData[metric] = make(map[schema.MetricScope]*schema.JobMetric)
}
jobMetric, ok := jobData[metric][scope]
if !ok {
jobMetric = &schema.JobMetric{
Unit: metricConfig.Unit,
Timestep: metricConfig.Timestep,
Series: make([]schema.Series, 0),
}
}
step := int64(metricConfig.Timestep)
steps := int64(to.Sub(from).Seconds()) / step
// iter rows of host, metric, values
for _, row := range result.(promm.Matrix) {
jobMetric.Series = append(jobMetric.Series,
pdb.RowToSeries(from, step, steps, row))
}
2023-08-08 10:17:30 +02:00
// only add metric if at least one host returned data
if !ok && len(jobMetric.Series) > 0{
jobData[metric][scope] = jobMetric
}
2022-12-08 13:51:44 +01:00
// sort by hostname to get uniform coloring
sort.Slice(jobMetric.Series, func(i, j int) bool {
2023-02-15 11:50:51 +01:00
return (jobMetric.Series[i].Hostname < jobMetric.Series[j].Hostname)
2022-12-08 13:51:44 +01:00
})
}
}
return jobData, nil
}
// TODO change implementation to precomputed/cached stats
func (pdb *PrometheusDataRepository) LoadStats(
job *schema.Job,
metrics []string,
ctx context.Context) (map[string]map[string]schema.MetricStatistics, error) {
// map of metrics of nodes of stats
stats := map[string]map[string]schema.MetricStatistics{}
data, err := pdb.LoadData(job, metrics, []schema.MetricScope{schema.MetricScopeNode}, ctx)
if err != nil {
log.Warn("Error while loading job for stats")
2022-12-08 13:51:44 +01:00
return nil, err
}
for metric, metricData := range data {
stats[metric] = make(map[string]schema.MetricStatistics)
for _, series := range metricData[schema.MetricScopeNode].Series {
2023-04-11 16:26:09 +02:00
stats[metric][series.Hostname] = series.Statistics
2022-12-08 13:51:44 +01:00
}
}
return stats, nil
}
func (pdb *PrometheusDataRepository) LoadNodeData(
cluster string,
metrics, nodes []string,
scopes []schema.MetricScope,
from, to time.Time,
ctx context.Context) (map[string]map[string][]*schema.JobMetric, error) {
t0 := time.Now()
// Map of hosts of metrics of value slices
data := make(map[string]map[string][]*schema.JobMetric)
// query db for each metric
// TODO: scopes seems to be always empty
if len(scopes) == 0 || !contains(scopes, schema.MetricScopeNode) {
scopes = append(scopes, schema.MetricScopeNode)
}
for _, scope := range scopes {
if scope != schema.MetricScopeNode {
2023-02-15 11:50:51 +01:00
logOnce.Do(func() {
log.Infof("Note: Scope '%s' requested, but not yet supported: Will return 'node' scope only.", scope)
})
2022-12-08 13:51:44 +01:00
continue
}
for _, metric := range metrics {
metricConfig := archive.GetMetricConfig(cluster, metric)
if metricConfig == nil {
log.Warnf("Error in LoadNodeData: Metric %s for cluster %s not configured", metric, cluster)
return nil, errors.New("Prometheus config error")
2022-12-08 13:51:44 +01:00
}
query, err := pdb.FormatQuery(metric, scope, nodes, cluster)
if err != nil {
log.Warn("Error while formatting prometheus query")
2022-12-08 13:51:44 +01:00
return nil, err
}
// ranged query over all nodes
r := promv1.Range{
2023-02-15 11:50:51 +01:00
Start: from,
End: to,
Step: time.Duration(metricConfig.Timestep * 1e9),
}
2022-12-08 13:51:44 +01:00
result, warnings, err := pdb.queryClient.QueryRange(ctx, query, r)
if err != nil {
log.Errorf("Prometheus query error in LoadNodeData: %v\n", err)
return nil, errors.New("Prometheus query error")
2022-12-08 13:51:44 +01:00
}
if len(warnings) > 0 {
log.Warnf("Warnings: %v\n", warnings)
2022-12-08 13:51:44 +01:00
}
step := int64(metricConfig.Timestep)
steps := int64(to.Sub(from).Seconds()) / step
// iter rows of host, metric, values
for _, row := range result.(promm.Matrix) {
hostname := strings.TrimSuffix(string(row.Metric["exported_instance"]), pdb.suffix)
hostdata, ok := data[hostname]
if !ok {
hostdata = make(map[string][]*schema.JobMetric)
data[hostname] = hostdata
}
// output per host and metric
hostdata[metric] = append(hostdata[metric], &schema.JobMetric{
Unit: metricConfig.Unit,
Timestep: metricConfig.Timestep,
2023-02-15 11:50:51 +01:00
Series: []schema.Series{pdb.RowToSeries(from, step, steps, row)},
},
2022-12-08 13:51:44 +01:00
)
}
}
}
t1 := time.Since(t0)
log.Debugf("LoadNodeData of %v nodes took %s", len(data), t1)
2022-12-08 13:51:44 +01:00
return data, nil
}