Prometheus integration complete

Co-authored-by: Matteo Nardelli <matnar@gmail.com>

Author: grussorusso

internal/config/keys.go

@@ -59,9 +59,15 @@ const LISTEN_UDP_PORT = "registry.udp.port"
 // enable metrics system
 const METRICS_ENABLED = "metrics.enabled"
 
-// HTTP port used for Prometheus scraping
+// Port used by Prometheus server
 const METRICS_PROMETHEUS_PORT = "metrics.prometheus.port"
 
+// Prometheus IP address / hostname
+const METRICS_PROMETHEUS_HOST = "metrics.prometheus.host"
+
+// Interval (in seconds) for metrics retriever
+const METRICS_RETRIEVER_INTERVAL = "metrics.retriever.interval"
+
 // Scheduling policy to use
 // Possible values: "qosaware", "default", "cloudonly"
 const SCHEDULING_POLICY = "scheduler.policy"

internal/metrics/metrics.go

@@ -18,18 +18,28 @@ var registry = prometheus.NewRegistry()
 var ScrapingHandler http.Handler = nil
 var durationBuckets = []float64{0.002, 0.005, 0.010, 0.02, 0.03, 0.05, 0.1, 0.15, 0.3, 0.6, 1.0}
 
+const (
+	COMPLETIONS    = "completed_total"
+	EXECUTION_TIME = "execution_time"
+)
+
 var (
-	CompletedInvocations = promauto.NewCounterVec(prometheus.CounterOpts{
-		Name: "completed_total",
+	metricCompletions = promauto.NewCounterVec(prometheus.CounterOpts{
+		Name: COMPLETIONS,
 		Help: "Number of completed function invocations",
 	}, []string{"node", "function"})
-	ExecutionTimes = promauto.NewHistogramVec(prometheus.HistogramOpts{
-		Name:    "execution_time",
+	metricExecutionTime = promauto.NewHistogramVec(prometheus.HistogramOpts{
+		Name:    EXECUTION_TIME,
 		Help:    "Function duration",
 		Buckets: durationBuckets,
 	}, []string{"node", "function"})
 )
 
+type RetrievedMetrics struct {
+	Completions      map[string]float64
+	AvgExecutionTime map[string]float64
+}
+
 func Init() {
 	if config.GetBool(config.METRICS_ENABLED, false) {
 		log.Println("Metrics enabled.")
@@ -39,16 +49,18 @@ func Init() {
 		return
 	}
 
-	registry.MustRegister(CompletedInvocations)
-	registry.MustRegister(ExecutionTimes)
+	registry.MustRegister(metricCompletions)
+	registry.MustRegister(metricExecutionTime)
 
 	ScrapingHandler = promhttp.HandlerFor(registry, promhttp.HandlerOpts{
 		EnableOpenMetrics: true})
+
+	go MetricsRetriever()
 }
 
 func AddCompletedInvocation(funcName string) {
-	CompletedInvocations.With(prometheus.Labels{"function": funcName, "node": node.NodeIdentifier}).Inc()
+	metricCompletions.With(prometheus.Labels{"function": funcName, "node": node.NodeIdentifier}).Inc()
 }
 func AddFunctionDurationValue(funcName string, duration float64) {
-	ExecutionTimes.With(prometheus.Labels{"function": funcName, "node": node.NodeIdentifier}).Observe(duration)
+	metricExecutionTime.With(prometheus.Labels{"function": funcName, "node": node.NodeIdentifier}).Observe(duration)
 }

internal/metrics/query.go

@@ -0,0 +1,118 @@
+package metrics
+
+import (
+	"context"
+	"fmt"
+	"github.com/prometheus/common/model"
+	"github.com/serverledge-faas/serverledge/internal/config"
+	"github.com/serverledge-faas/serverledge/internal/node"
+	"log"
+	"time"
+
+	promapi "github.com/prometheus/client_golang/api"
+	v1 "github.com/prometheus/client_golang/api/prometheus/v1"
+)
+
+var retrievedMetrics RetrievedMetrics
+
+func retrieveSingleValue(query string, api v1.API, ctx context.Context) (float64, error) {
+
+	result, warnings, err := api.Query(ctx, query, time.Now())
+	if err != nil {
+		return 0.0, fmt.Errorf("Failed query : %v\n", err)
+	}
+
+	if len(warnings) > 0 {
+		log.Printf("Received warnings in the execution of the : %v\n", warnings)
+	}
+
+	vector, ok := result.(model.Vector)
+	if !ok {
+		return 0.0, fmt.Errorf("Could not convert the result of the query : %v\n", result)
+	}
+	if vector.Len() != 1 {
+		return 0.0, fmt.Errorf("Expected 1 result; found %d\n", vector.Len())
+	}
+
+	sample := vector[0]
+	return float64(sample.Value), nil
+}
+
+func retrieveByFunction(query string, api v1.API, ctx context.Context) (map[string]float64, error) {
+
+	result, warnings, err := api.Query(ctx, query, time.Now())
+	if err != nil {
+		return nil, fmt.Errorf("Failed query : %v\n", err)
+	}
+
+	if len(warnings) > 0 {
+		log.Printf("Received warnings in the execution of the : %v\n", warnings)
+	}
+
+	functionValues := make(map[string]float64)
+	if vector, ok := result.(model.Vector); ok {
+		for _, sample := range vector {
+			value := float64(sample.Value)
+			functionName, found := sample.Metric[model.LabelName("function")]
+			if !found {
+				log.Printf("Could not find the function name in the result : %v\n", sample)
+				continue
+			} else {
+				functionValues[string(functionName)] = value
+			}
+		}
+	} else {
+		return nil, fmt.Errorf("Unexpected Result %v\n", result)
+	}
+
+	return functionValues, nil
+}
+
+func MetricsRetriever() {
+	prometheusHost := config.GetString(config.METRICS_PROMETHEUS_HOST, "127.0.0.1")
+	prometheusPort := config.GetInt(config.METRICS_PROMETHEUS_PORT, 9090)
+	client, err := promapi.NewClient(promapi.Config{
+		Address: fmt.Sprintf("http://%s:%d", prometheusHost, prometheusPort),
+	})
+	if err != nil {
+		log.Printf("Error in Prometheus client creation: %v\n", err)
+		return
+	}
+
+	// API of Prometheus
+	api := v1.NewAPI(client)
+	ctx := context.Background()
+
+	ticker := time.NewTicker(time.Duration(config.GetInt(config.METRICS_RETRIEVER_INTERVAL, 10)) * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ticker.C:
+
+			query := fmt.Sprintf("%s{node=\"%s\"}", COMPLETIONS, node.NodeIdentifier)
+			completionsPerFunction, err := retrieveByFunction(query, api, ctx)
+			if err != nil {
+				log.Printf("Error in retrieveByFunction: %v\n", err)
+			}
+			retrievedMetrics.Completions = completionsPerFunction
+
+			query = fmt.Sprintf("%s_sum{node=\"%s\"}/%s_count{node=\"%s\"}",
+				EXECUTION_TIME, node.NodeIdentifier, EXECUTION_TIME, node.NodeIdentifier)
+			avgFunDuration, err := retrieveByFunction(query, api, ctx)
+			if err != nil {
+				log.Printf("Error in retrieveByFunction: %v\n", err)
+			}
+			retrievedMetrics.AvgExecutionTime = avgFunDuration
+
+			fmt.Println("All queries completed")
+
+		}
+	}
+
+}
+
+func GetMetrics() RetrievedMetrics {
+	// TODO: deep copy?
+	return retrievedMetrics
+}