Add InfluxDBv2 as metric data repo

.env

@@ -0,0 +1,11 @@
+
+export CCMETRICSTORE_URL="http://localhost:8081"
+export CCMETRICSTORE_JWT="eyJ0eXAiOiJKV1QiLCJhbGciOiJFZERTQSJ9.eyJ1c2VyIjoiYWRtaW4iLCJyb2xlcyI6WyJST0xFX0FETUlOIiwiUk9MRV9BTkFMWVNUIiwiUk9MRV9VU0VSIl19.d-3_3FZTsadPjDEdsWrrQ7nS0edMAR4zjl-eK7rJU3HziNBfI9PDHDIpJVHTNN5E5SlLGLFXctWyKAkwhXL-Dw"
+
+export INFLUXDB_V2_TOKEN="egLfcf7fx0FESqFYU3RpAAbj"
+
+export JWT_PUBLIC_KEY="kzfYrYy+TzpanWZHJ5qSdMj5uKUWgq74BWhQG6copP0="
+export JWT_PRIVATE_KEY="dtPC/6dWJFKZK7KZ78CvWuynylOmjBFyMsUWArwmodOTN9itjL5POlqdZkcnmpJ0yPm4pRaCrvgFaFAbpyik/Q=="
+export SESSION_KEY="67d829bf61dc5f87a73fd814e2c9f629"
+
+export LDAP_ADMIN_PASSWORD="mashup"

metricdata/archive.go

@@ -18,8 +18,6 @@ import (
 	"github.com/ClusterCockpit/cc-jobarchive/schema"
 )
 
-var JobArchivePath string = "./var/job-archive"
-
 // For a given job, return the path of the `data.json`/`meta.json` file.
 // TODO: Implement Issue ClusterCockpit/ClusterCockpit#97
 func getPath(job *model.Job, file string) (string, error) {

metricdata/cc-metric-store.go

@@ -46,11 +46,11 @@ type ApiStatsData struct {
 	Max     schema.Float `json:"max"`
 }
 
-func (ccms *CCMetricStore) Init() error {
-	ccms.url = os.Getenv("CCMETRICSTORE_URL")
+func (ccms *CCMetricStore) Init(url string) error {
+	ccms.url = url // os.Getenv("CCMETRICSTORE_URL")
 	ccms.jwt = os.Getenv("CCMETRICSTORE_JWT")
-	if ccms.url == "" || ccms.jwt == "" {
-		return errors.New("environment variables 'CCMETRICSTORE_URL' or 'CCMETRICSTORE_JWT' not set")
+	if ccms.jwt == "" {
+		return errors.New("environment variable 'CCMETRICSTORE_JWT' not set")
 	}
 
 	return nil

metricdata/influxdb-v2.go

@@ -0,0 +1,143 @@
+package metricdata
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/ClusterCockpit/cc-jobarchive/config"
+	"github.com/ClusterCockpit/cc-jobarchive/graph/model"
+	"github.com/ClusterCockpit/cc-jobarchive/schema"
+	influxdb2 "github.com/influxdata/influxdb-client-go/v2"
+	influxdb2Api "github.com/influxdata/influxdb-client-go/v2/api"
+)
+
+type InfluxDBv2DataRepository struct {
+	client              influxdb2.Client
+	queryClient         influxdb2Api.QueryAPI
+	bucket, measurement string
+}
+
+func (idb *InfluxDBv2DataRepository) Init(url string) error {
+	token := os.Getenv("INFLUXDB_V2_TOKEN")
+	if token == "" {
+		return errors.New("warning: environment variable 'INFLUXDB_V2_TOKEN' not set")
+	}
+
+	idb.client = influxdb2.NewClient(url, token)
+	idb.queryClient = idb.client.QueryAPI("ClusterCockpit")
+	idb.bucket = "ClusterCockpit/data"
+	idb.measurement = "data"
+	return nil
+}
+
+func (idb *InfluxDBv2DataRepository) formatTime(t time.Time) string {
+	return fmt.Sprintf("%d-%02d-%02dT%02d:%02d:%02dZ",
+		t.Year(), t.Month(), t.Day(), t.Hour(), t.Minute(), t.Second())
+}
+
+func (idb *InfluxDBv2DataRepository) LoadData(job *model.Job, metrics []string, ctx context.Context) (schema.JobData, error) {
+	fieldsConds := make([]string, 0, len(metrics))
+	for _, m := range metrics {
+		fieldsConds = append(fieldsConds, fmt.Sprintf(`r._field == "%s"`, m))
+	}
+	fieldsCond := strings.Join(fieldsConds, " or ")
+
+	hostsConds := make([]string, 0, len(job.Nodes))
+	for _, h := range job.Nodes {
+		hostsConds = append(hostsConds, fmt.Sprintf(`r.host == "%s"`, h))
+	}
+	hostsCond := strings.Join(hostsConds, " or ")
+
+	query := fmt.Sprintf(`from(bucket: "%s")
+		|> range(start: %s, stop: %s)
+		|> filter(fn: (r) => r._measurement == "%s" and (%s) and (%s))
+		|> drop(columns: ["_start", "_stop", "_measurement"])`, idb.bucket,
+		idb.formatTime(job.StartTime), idb.formatTime(job.StartTime.Add(time.Duration(job.Duration)).Add(1*time.Second)),
+		idb.measurement, hostsCond, fieldsCond)
+	rows, err := idb.queryClient.Query(ctx, query)
+	if err != nil {
+		return nil, err
+	}
+
+	jobData := make(schema.JobData)
+
+	var currentSeries *schema.MetricSeries = nil
+	for rows.Next() {
+		row := rows.Record()
+		if currentSeries == nil || rows.TableChanged() {
+			field, host := row.Field(), row.ValueByKey("host").(string)
+			jobMetric, ok := jobData[field]
+			if !ok {
+				mc := config.GetMetricConfig(job.ClusterID, field)
+				jobMetric = &schema.JobMetric{
+					Scope:    "node", // TODO: FIXME: Whatever...
+					Unit:     mc.Unit,
+					Timestep: mc.Sampletime,
+					Series:   make([]*schema.MetricSeries, 0, len(job.Nodes)),
+				}
+				jobData[field] = jobMetric
+			}
+
+			currentSeries = &schema.MetricSeries{
+				NodeID:     host,
+				Statistics: nil,
+				Data:       make([]schema.Float, 0),
+			}
+			jobMetric.Series = append(jobMetric.Series, currentSeries)
+		}
+
+		val := row.Value().(float64)
+		currentSeries.Data = append(currentSeries.Data, schema.Float(val))
+	}
+
+	return jobData, idb.addStats(job, jobData, metrics, hostsCond, ctx)
+}
+
+func (idb *InfluxDBv2DataRepository) addStats(job *model.Job, jobData schema.JobData, metrics []string, hostsCond string, ctx context.Context) error {
+	for _, metric := range metrics {
+		query := fmt.Sprintf(`
+			data = from(bucket: "%s")
+				|> range(start: %s, stop: %s)
+				|> filter(fn: (r) => r._measurement == "%s" and r._field == "%s" and (%s))
+
+			union(tables: [
+					data |> mean(column: "_value") |> set(key: "_field", value: "avg")
+					data |>  min(column: "_value") |> set(key: "_field", value: "min")
+					data |>  max(column: "_value") |> set(key: "_field", value: "max")
+				])
+				|> pivot(rowKey: ["host"], columnKey: ["_field"], valueColumn: "_value")
+				|> group()`, idb.bucket,
+			idb.formatTime(job.StartTime), idb.formatTime(job.StartTime.Add(time.Duration(job.Duration)).Add(1*time.Second)),
+			idb.measurement, metric, hostsCond)
+		rows, err := idb.queryClient.Query(ctx, query)
+		if err != nil {
+			return err
+		}
+
+		jobMetric := jobData[metric]
+		for rows.Next() {
+			row := rows.Record()
+			host := row.ValueByKey("host").(string)
+			avg, min, max := row.ValueByKey("avg").(float64),
+				row.ValueByKey("min").(float64),
+				row.ValueByKey("max").(float64)
+
+			for _, s := range jobMetric.Series {
+				if s.NodeID == host {
+					s.Statistics = &schema.MetricStatistics{
+						Avg: avg,
+						Min: min,
+						Max: max,
+					}
+					break
+				}
+			}
+		}
+	}
+
+	return nil
+}

metricdata/metricdata.go

@@ -4,25 +4,55 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"log"
 
+	"github.com/ClusterCockpit/cc-jobarchive/config"
 	"github.com/ClusterCockpit/cc-jobarchive/graph/model"
 	"github.com/ClusterCockpit/cc-jobarchive/schema"
 )
 
-var runningJobs *CCMetricStore
+type MetricDataRepository interface {
+	Init(url string) error
+	LoadData(job *model.Job, metrics []string, ctx context.Context) (schema.JobData, error)
+}
 
-func init() {
-	runningJobs = &CCMetricStore{}
-	if err := runningJobs.Init(); err != nil {
-		log.Fatalln(err)
+var metricDataRepos map[string]MetricDataRepository = map[string]MetricDataRepository{}
+
+var JobArchivePath string
+
+func Init(jobArchivePath string) error {
+	JobArchivePath = jobArchivePath
+	for _, cluster := range config.Clusters {
+		if cluster.MetricDataRepository != nil {
+			switch cluster.MetricDataRepository.Kind {
+			case "cc-metric-store":
+				ccms := &CCMetricStore{}
+				if err := ccms.Init(cluster.MetricDataRepository.Url); err != nil {
+					return err
+				}
+				metricDataRepos[cluster.ClusterID] = ccms
+			case "influxdb-v2":
+				idb := &InfluxDBv2DataRepository{}
+				if err := idb.Init(cluster.MetricDataRepository.Url); err != nil {
+					return err
+				}
+				metricDataRepos[cluster.ClusterID] = idb
+			default:
+				return fmt.Errorf("unkown metric data repository '%s' for cluster '%s'", cluster.MetricDataRepository.Kind, cluster.ClusterID)
+			}
+		}
 	}
+	return nil
 }
 
 // Fetches the metric data for a job.
 func LoadData(job *model.Job, metrics []string, ctx context.Context) (schema.JobData, error) {
 	if job.State == model.JobStateRunning {
-		return runningJobs.LoadData(job, metrics, ctx)
+		repo, ok := metricDataRepos[job.ClusterID]
+		if !ok {
+			return nil, fmt.Errorf("no metric data repository configured for '%s'", job.ClusterID)
+		}
+
+		return repo.LoadData(job, metrics, ctx)
 	}
 
 	if job.State != model.JobStateCompleted {