Minor corrections in SLURM fetcher and TSDB updater (#280)

* fix: Minor corrections in fetcher and updater

* When pulling historical SLURM data, use running state as well to update jobs incrementally.

* Use `lookback_delta` query parameter in TSDB queries to limit the artificial look back metrics which can over estimate energy consumption

* refactor: Use text/template to avoid escaping chars like `<`

---------

Signed-off-by: Mahendra Paipuri <[email protected]>

Author: mahendrapaipuri

pkg/api/resource/slurm/cli.go

@@ -487,21 +487,6 @@ func parseSacctMgrCmdOutput(sacctMgrOutput string, currentTime string) ([]models
 
 // runSacctCmd executes sacct command and return output.
 func (s *slurmScheduler) runSacctCmd(ctx context.Context, start, end time.Time) ([]byte, error) {
-	// If we are fetching historical data, do not use RUNNING state as it can report
-	// same job twice once when it was still in running state and once it is in completed
-	// state.
-	// endTimeParsed, _ := time.Parse(base.DatetimeLayout, endTime)
-	var states []string
-	// When fetching current jobs, endTime should be very close to current time. Here we
-	// assume that if current time is more than 5 sec than end time, we are fetching
-	// historical data
-	if time.Now().In(end.Location()).Sub(end) > 5*time.Second {
-		// Strip RUNNING state from slice
-		states = slurmStates[:len(slurmStates)-1]
-	} else {
-		states = slurmStates
-	}
-
 	// sacct path
 	sacctPath := filepath.Join(s.cluster.CLI.Path, "sacct")
 
@@ -515,7 +500,7 @@ func (s *slurmScheduler) runSacctCmd(ctx context.Context, start, end time.Time)
 	args := []string{
 		"-D", "-X", "--noheader", "--allusers", "--parsable2",
 		"--format", strings.Join(sacctFields, ","),
-		"--state", strings.Join(states, ","),
+		"--state", strings.Join(slurmStates, ","),
 		"--starttime", start.Format(base.DatetimeLayout),
 		"--endtime", end.Format(base.DatetimeLayout),
 	}

pkg/api/resource/slurm/manager.go

@@ -138,8 +138,6 @@ func (s *slurmScheduler) FetchUsersProjects(
 
 // Get jobs from slurm sacct command.
 func (s *slurmScheduler) fetchFromSacct(ctx context.Context, start time.Time, end time.Time) ([]models.Unit, error) {
-	// startTime := start.Format(base.DatetimeLayout)
-	// endTime := end.Format(base.DatetimeLayout)
 	// Execute sacct command between start and end times
 	sacctOutput, err := s.runSacctCmd(ctx, start, end)
 	if err != nil {

pkg/api/updater/tsdb/tsdb.go

@@ -5,12 +5,12 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"html/template"
 	"log/slog"
 	"maps"
 	"math"
 	"strings"
 	"sync"
+	"text/template"
 	"time"
 
 	"github.com/mahendrapaipuri/ceems/pkg/api/helper"

pkg/tsdb/tsdb.go

@@ -322,6 +322,14 @@ func (t *TSDB) Query(ctx context.Context, query string, queryTime time.Time) (Me
 		"time":  []string{queryTime.UTC().Format(time.RFC3339Nano)},
 	}
 
+	// Get current scrape interval to use as lookback_delta
+	// This query parameter is undocumented on Prometheus. If we use
+	// default value of 5m, we tend to have metrics 5m **after** compute
+	// unit has finished which gives over estimation of energy
+	if scrapeInterval := t.Settings(ctx).ScrapeInterval; scrapeInterval > 0 {
+		values.Add("lookback_delta", scrapeInterval.String())
+	}
+
 	// Create a new POST request
 	req, err := http.NewRequestWithContext(
 		ctx,
@@ -449,6 +457,14 @@ func (t *TSDB) RangeQuery(
 		"step":  []string{step},
 	}
 
+	// Get current scrape interval to use as lookback_delta
+	// This query parameter is undocumented on Prometheus. If we use
+	// default value of 5m, we tend to have metrics 5m **after** compute
+	// unit has finished which gives over estimation of energy
+	if scrapeInterval := t.Settings(ctx).ScrapeInterval; scrapeInterval > 0 {
+		values.Add("lookback_delta", scrapeInterval.String())
+	}
+
 	// Create a new POST request
 	req, err := http.NewRequestWithContext(
 		ctx,