scheduler: Collect rpc stats

Author: fihuer

scheduler.go

@@ -17,8 +17,8 @@ package main
 
 import (
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/common/log"
 	"io/ioutil"
-	"log"
 	"os/exec"
 	"regexp"
 	"strconv"
@@ -33,23 +33,29 @@ import (
 
 // Basic metrics for the scheduler
 type SchedulerMetrics struct {
-	threads                           float64
-	queue_size                        float64
-	dbd_queue_size                    float64
-	last_cycle                        float64
-	mean_cycle                        float64
-	cycle_per_minute                  float64
-	backfill_last_cycle               float64
-	backfill_mean_cycle               float64
-	backfill_depth_mean               float64
+	threads                   	  float64
+	queue_size                	  float64
+	dbd_queue_size            	  float64
+	last_cycle                	  float64
+	mean_cycle                	  float64
+	cycle_per_minute          	  float64
+	backfill_last_cycle       	  float64
+	backfill_mean_cycle       	  float64
+	backfill_depth_mean       	  float64
 	total_backfilled_jobs_since_start float64
 	total_backfilled_jobs_since_cycle float64
 	total_backfilled_heterogeneous    float64
+	rpc_stats_count           map[string]float64
+	rpc_stats_avg_time        map[string]float64
+	rpc_stats_total_time      map[string]float64
+	user_rpc_stats_count      map[string]float64
+	user_rpc_stats_avg_time   map[string]float64
+	user_rpc_stats_total_time map[string]float64
 }
 
 // Execute the sdiag command and return its output
 func SchedulerData() []byte {
-	cmd := exec.Command("sdiag")
+	cmd := exec.Command("/usr/bin/sdiag")
 	stdout, err := cmd.StdoutPipe()
 	if err != nil {
 		log.Fatal(err)
@@ -121,9 +127,87 @@ func ParseSchedulerMetrics(input []byte) *SchedulerMetrics {
 			}
 		}
 	}
+	rpc_stats := ParseRpcStats(lines)
+	sm.rpc_stats_count = rpc_stats[0]
+	sm.rpc_stats_avg_time = rpc_stats[1]
+	sm.rpc_stats_total_time = rpc_stats[2]
+	sm.user_rpc_stats_count = rpc_stats[3]
+	sm.user_rpc_stats_avg_time = rpc_stats[4]
+	sm.user_rpc_stats_total_time = rpc_stats[5]
 	return &sm
 }
 
+// Helper function to split a single line from the sdiag output
+func SplitColonValueToFloat(input string) float64 {
+	str := strings.Split(input, ":")
+	if len(str) == 1 {
+		return 0
+	} else {
+		rvalue := strings.TrimSpace(str[1])
+		flt, _ := strconv.ParseFloat(rvalue, 64)
+		return flt
+	}
+}
+
+// Helper function to return RPC stats from sdiag output
+func ParseRpcStats(lines []string) []map[string]float64 {
+	var in_rpc bool
+	var in_rpc_per_user bool
+	var count_stats map[string]float64
+	var avg_stats map[string]float64
+	var total_stats map[string]float64
+	var user_count_stats map[string]float64
+	var user_avg_stats map[string]float64
+	var user_total_stats map[string]float64
+
+	count_stats = make(map[string]float64)
+	avg_stats = make(map[string]float64)
+	total_stats = make(map[string]float64)
+	user_count_stats = make(map[string]float64)
+	user_avg_stats = make(map[string]float64)
+	user_total_stats = make(map[string]float64)
+
+	in_rpc = false
+	in_rpc_per_user = false
+
+	stat_line_re := regexp.MustCompile(`^\s*([A-Za-z0-9_]*).*count:([0-9]*)\s*ave_time:([0-9]*)\s\s*total_time:([0-9]*)\s*$`)
+
+	for _, line := range lines {
+		if strings.Contains(line, "Remote Procedure Call statistics by message type") {
+			in_rpc = true
+			in_rpc_per_user = false
+		} else if strings.Contains(line, "Remote Procedure Call statistics by user") {
+			in_rpc = false
+			in_rpc_per_user = true
+		}
+		if in_rpc || in_rpc_per_user {
+			re_match := stat_line_re.FindAllStringSubmatch(line, -1)
+			if re_match != nil {
+				re_match_first := re_match[0]
+				if in_rpc {
+					count_stats[re_match_first[1]], _ = strconv.ParseFloat(re_match_first[2], 64)
+					avg_stats[re_match_first[1]], _ = strconv.ParseFloat(re_match_first[3], 64)
+					total_stats[re_match_first[1]], _ = strconv.ParseFloat(re_match_first[4], 64)
+				} else if in_rpc_per_user {
+					user_count_stats[re_match_first[1]], _ = strconv.ParseFloat(re_match_first[2], 64)
+					user_avg_stats[re_match_first[1]], _ = strconv.ParseFloat(re_match_first[3], 64)
+					user_total_stats[re_match_first[1]], _ = strconv.ParseFloat(re_match_first[4], 64)
+				}
+			}
+		}
+	}
+
+	rpc_stats_final := []map[string]float64{
+		count_stats,
+		avg_stats,
+		total_stats,
+		user_count_stats,
+		user_avg_stats,
+		user_total_stats,
+	}
+	return rpc_stats_final
+}
+
 // Returns the scheduler metrics
 func SchedulerGetMetrics() *SchedulerMetrics {
 	return ParseSchedulerMetrics(SchedulerData())
@@ -137,18 +221,24 @@ func SchedulerGetMetrics() *SchedulerMetrics {
 
 // Collector strcture
 type SchedulerCollector struct {
-	threads                           *prometheus.Desc
-	queue_size                        *prometheus.Desc
-	dbd_queue_size                    *prometheus.Desc
-	last_cycle                        *prometheus.Desc
-	mean_cycle                        *prometheus.Desc
-	cycle_per_minute                  *prometheus.Desc
-	backfill_last_cycle               *prometheus.Desc
-	backfill_mean_cycle               *prometheus.Desc
-	backfill_depth_mean               *prometheus.Desc
+	threads                   	  *prometheus.Desc
+	queue_size                	  *prometheus.Desc
+	dbd_queue_size            	  *prometheus.Desc
+	last_cycle                	  *prometheus.Desc
+	mean_cycle                	  *prometheus.Desc
+	cycle_per_minute          	  *prometheus.Desc
+	backfill_last_cycle       	  *prometheus.Desc
+	backfill_mean_cycle       	  *prometheus.Desc
+	backfill_depth_mean       	  *prometheus.Desc
 	total_backfilled_jobs_since_start *prometheus.Desc
 	total_backfilled_jobs_since_cycle *prometheus.Desc
 	total_backfilled_heterogeneous    *prometheus.Desc
+	rpc_stats_count           *prometheus.Desc
+	rpc_stats_avg_time        *prometheus.Desc
+	rpc_stats_total_time      *prometheus.Desc
+	user_rpc_stats_count      *prometheus.Desc
+	user_rpc_stats_avg_time   *prometheus.Desc
+	user_rpc_stats_total_time *prometheus.Desc
 }
 
 // Send all metric descriptions
@@ -165,6 +255,12 @@ func (c *SchedulerCollector) Describe(ch chan<- *prometheus.Desc) {
 	ch <- c.total_backfilled_jobs_since_start
 	ch <- c.total_backfilled_jobs_since_cycle
 	ch <- c.total_backfilled_heterogeneous
+	ch <- c.rpc_stats_count
+	ch <- c.rpc_stats_avg_time
+	ch <- c.rpc_stats_total_time
+	ch <- c.user_rpc_stats_count
+	ch <- c.user_rpc_stats_avg_time
+	ch <- c.user_rpc_stats_total_time
 }
 
 // Send the values of all metrics
@@ -182,10 +278,33 @@ func (sc *SchedulerCollector) Collect(ch chan<- prometheus.Metric) {
 	ch <- prometheus.MustNewConstMetric(sc.total_backfilled_jobs_since_start, prometheus.GaugeValue, sm.total_backfilled_jobs_since_start)
 	ch <- prometheus.MustNewConstMetric(sc.total_backfilled_jobs_since_cycle, prometheus.GaugeValue, sm.total_backfilled_jobs_since_cycle)
 	ch <- prometheus.MustNewConstMetric(sc.total_backfilled_heterogeneous, prometheus.GaugeValue, sm.total_backfilled_heterogeneous)
+	for rpc_type, value := range sm.rpc_stats_count {
+		ch <- prometheus.MustNewConstMetric(sc.rpc_stats_count, prometheus.GaugeValue, value, rpc_type)
+	}
+	for rpc_type, value := range sm.rpc_stats_avg_time {
+		ch <- prometheus.MustNewConstMetric(sc.rpc_stats_avg_time, prometheus.GaugeValue, value, rpc_type)
+	}
+	for rpc_type, value := range sm.rpc_stats_total_time {
+		ch <- prometheus.MustNewConstMetric(sc.rpc_stats_total_time, prometheus.GaugeValue, value, rpc_type)
+	}
+	for user, value := range sm.user_rpc_stats_count {
+		ch <- prometheus.MustNewConstMetric(sc.user_rpc_stats_count, prometheus.GaugeValue, value, user)
+	}
+	for user, value := range sm.user_rpc_stats_avg_time {
+		ch <- prometheus.MustNewConstMetric(sc.user_rpc_stats_avg_time, prometheus.GaugeValue, value, user)
+	}
+	for user, value := range sm.user_rpc_stats_total_time {
+		ch <- prometheus.MustNewConstMetric(sc.user_rpc_stats_total_time, prometheus.GaugeValue, value, user)
+	}
+
 }
 
 // Returns the Slurm scheduler collector, used to register with the prometheus client
 func NewSchedulerCollector() *SchedulerCollector {
+	rpc_stats_labels := make([]string, 0, 1)
+	rpc_stats_labels = append(rpc_stats_labels, "operation")
+	user_rpc_stats_labels := make([]string, 0, 1)
+	user_rpc_stats_labels = append(user_rpc_stats_labels, "user")
 	return &SchedulerCollector{
 		threads: prometheus.NewDesc(
 			"slurm_scheduler_threads",
@@ -247,5 +366,35 @@ func NewSchedulerCollector() *SchedulerCollector {
 			"Information provided by the Slurm sdiag command, number of heterogeneous job components started thanks to backfilling since last Slurm start",
 			nil,
 			nil),
+		rpc_stats_count: prometheus.NewDesc(
+			"slurm_rpc_stats",
+			"Information provided by the Slurm sdiag command, rpc count statistic",
+			rpc_stats_labels,
+			nil),
+		rpc_stats_avg_time: prometheus.NewDesc(
+			"slurm_rpc_stats_avg_time",
+			"Information provided by the Slurm sdiag command, rpc average time statistic",
+			rpc_stats_labels,
+			nil),
+		rpc_stats_total_time: prometheus.NewDesc(
+			"slurm_rpc_stats_total_time",
+			"Information provided by the Slurm sdiag command, rpc total time statistic",
+			rpc_stats_labels,
+			nil),
+		user_rpc_stats_count: prometheus.NewDesc(
+			"slurm_user_rpc_stats",
+			"Information provided by the Slurm sdiag command, rpc count statistic per user",
+			user_rpc_stats_labels,
+			nil),
+		user_rpc_stats_avg_time: prometheus.NewDesc(
+			"slurm_user_rpc_stats_avg_time",
+			"Information provided by the Slurm sdiag command, rpc average time statistic per user",
+			user_rpc_stats_labels,
+			nil),
+		user_rpc_stats_total_time: prometheus.NewDesc(
+			"slurm_user_rpc_stats_total_time",
+			"Information provided by the Slurm sdiag command, rpc total time statistic per user",
+			user_rpc_stats_labels,
+			nil),
 	}
 }