(airavata-custos) 01/01: Usage monitor's job query within a defined range to avoid wide query issues

[la...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lahirujayathilake pushed a commit to branch bug-fix
in repository https://gitbox.apache.org/repos/asf/airavata-custos.git

commit b72f7f4de0f9a067da9888bb2b98f5f02daa4ee3
Author: lahiruj <la...@gmail.com>
AuthorDate: Fri Oct 2 12:58:20 2026 -0400

    Usage monitor's job query within a defined range to avoid wide query issues
---
 .../Usage-Monitor/internal/smonitor/smonitor.go    | 18 +++---
 .../internal/smonitor/smonitor_test.go             | 66 +++++++++++++++++++++-
 2 files changed, 73 insertions(+), 11 deletions(-)

diff --git a/connectors/SLURM/Usage-Monitor/internal/smonitor/smonitor.go b/connectors/SLURM/Usage-Monitor/internal/smonitor/smonitor.go
index f857db293..fe02586c6 100644
--- a/connectors/SLURM/Usage-Monitor/internal/smonitor/smonitor.go
+++ b/connectors/SLURM/Usage-Monitor/internal/smonitor/smonitor.go
@@ -40,6 +40,12 @@ const monitorInterval = 30 * time.Second
 // depends on the cluster's slurmdbd commit lag, so it is configurable.
 const defaultPollOverlap = 15 * time.Minute
 
+// maxPollStep is the value of the next poll step keeping the gap not too wide.
+const maxPollStep = 24 * time.Hour
+
+// startupCatchUp is the value how far back the poller scans for jobs.
+const startupCatchUp = 24 * time.Hour
+
 type jobLister interface {
 	ListJobs(filter client.JobFilter) ([]client.JobInfo, error)
 }
@@ -63,7 +69,7 @@ func NewSlurmMonitor(slurmClient *client.Client, eventBus *events.Bus, coreServi
 		coreService:     coreService,
 		clusterId:       clusterId,
 		pollOverlap:     pollOverlap,
-		lastMonitorTime: 1, // initialize to 1 to avoid issues with zero value
+		lastMonitorTime: time.Now().Add(-startupCatchUp).Unix(),
 	}
 }
 
@@ -71,7 +77,7 @@ func (m *SlurmMonitor) StartMonitor(ctx context.Context) {
 	ticker := time.NewTicker(monitorInterval)
 	defer ticker.Stop()
 
-	slog.Info("Starting SLURM usage monitor", "interval", monitorInterval)
+	slog.Info("Starting SLURM usage monitor", "interval", monitorInterval, "first_poll_from", time.Unix(m.lastMonitorTime, 0).UTC())
 	for {
 		select {
 		case <-ctx.Done():
@@ -99,13 +105,9 @@ func (m *SlurmMonitor) poll() {
 		return
 	}
 
-	windowStart := m.lastMonitorTime - int64(m.pollOverlap.Seconds())
-	if windowStart < 1 {
-		windowStart = 1
-	}
 	jobFilter := client.JobFilter{
-		StartTime: windowStart,
-		EndTime:   time.Now().Unix(),
+		StartTime: m.lastMonitorTime - int64(m.pollOverlap.Seconds()),
+		EndTime:   min(time.Now().Unix(), m.lastMonitorTime+int64(maxPollStep.Seconds())),
 	}
 
 	jobs, err := m.slurmClient.ListJobs(jobFilter)
diff --git a/connectors/SLURM/Usage-Monitor/internal/smonitor/smonitor_test.go b/connectors/SLURM/Usage-Monitor/internal/smonitor/smonitor_test.go
index 75fd320a7..4605fd45c 100644
--- a/connectors/SLURM/Usage-Monitor/internal/smonitor/smonitor_test.go
+++ b/connectors/SLURM/Usage-Monitor/internal/smonitor/smonitor_test.go
@@ -19,6 +19,7 @@ package smonitor
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -28,11 +29,14 @@ import (
 )
 
 type fakeJobLister struct {
-	jobs []client.JobInfo
+	jobs    []client.JobInfo
+	err     error              // returned by every query when set
+	filters []client.JobFilter // every query it was asked for, in order
 }
 
 func (f *fakeJobLister) ListJobs(filter client.JobFilter) ([]client.JobInfo, error) {
-	return f.jobs, nil
+	f.filters = append(f.filters, filter)
+	return f.jobs, f.err
 }
 
 func fixtureJob(jobID int64, user, partition string) client.JobInfo {
@@ -95,7 +99,63 @@ func newTestMonitor(core *service.CoreServiceMock, jobs ...client.JobInfo) *Slur
 		coreService:     core,
 		clusterId:       "cl-1",
 		pollOverlap:     defaultPollOverlap,
-		lastMonitorTime: 1,
+		lastMonitorTime: time.Now().Unix(),
+	}
+}
+
+// maxQuerySeconds is the longest time range one job query may cover.
+const maxQuerySeconds = int64((maxPollStep + defaultPollOverlap) / time.Second)
+
+// Make sure the first poll after a start asks only for the last day of jobs.
+func TestPollFirstQueryAfterAStartCoversOneStep(t *testing.T) {
+	lister := &fakeJobLister{}
+	m := NewSlurmMonitor(nil, nil, newMockCore(nil, nil), "cl-1", 0)
+	m.slurmClient = lister
+
+	m.poll()
+
+	query := lister.filters[0]
+	if seconds := query.EndTime - query.StartTime; seconds > maxQuerySeconds {
+		t.Errorf("first query covers %d seconds, want at most %d", seconds, maxQuerySeconds)
+	}
+}
+
+// Make sure a long gap is polled one step at a time, with no time skipped.
+func TestPollWalksALongGapInSteps(t *testing.T) {
+	lister := &fakeJobLister{}
+	m := newTestMonitor(newMockCore(nil, nil))
+	m.slurmClient = lister
+	m.lastMonitorTime = time.Now().Add(-60 * time.Hour).Unix()
+
+	before := time.Now().Unix()
+	for range 3 {
+		m.poll()
+	}
+
+	for i, query := range lister.filters {
+		if seconds := query.EndTime - query.StartTime; seconds > maxQuerySeconds {
+			t.Errorf("query %d covers %d seconds, want at most %d", i, seconds, maxQuerySeconds)
+		}
+		if i > 0 && query.StartTime > lister.filters[i-1].EndTime {
+			t.Errorf("query %d starts at %d, after query %d ended at %d", i, query.StartTime, i-1, lister.filters[i-1].EndTime)
+		}
+	}
+	if last := lister.filters[2].EndTime; last < before {
+		t.Errorf("after three polls the queries reach %d, want now (%d)", last, before)
+	}
+}
+
+// Make sure a failed query does not move the window forward.
+func TestPollKeepsTheWindowAfterAFailedQuery(t *testing.T) {
+	lister := &fakeJobLister{err: errors.New("cluster unreachable")}
+	m := newTestMonitor(newMockCore(nil, nil))
+	m.slurmClient = lister
+
+	m.poll()
+	m.poll()
+
+	if first, second := lister.filters[0].StartTime, lister.filters[1].StartTime; second != first {
+		t.Errorf("second query starts at %d, want the same start as the failed one (%d)", second, first)
 	}
 }