Merge branch 'slurm-refactor'

Author: ipatix

cmd/cc-slurm-adapter/daemon.go

@@ -106,7 +106,6 @@ func DaemonMain() error {
 			profiler.Begin()
 
 			trace.Info("Job Event timer triggered (%d events queued)", len(jobEvents))
-			slurmApi.ClearJobCache()
 			jobEventsProcess()
 			if len(jobEvents) > 0 {
 				jobEventTimer.Reset(queryDelay)
@@ -124,7 +123,6 @@ func DaemonMain() error {
 				pollEventTicker.Reset(pollEventInterval)
 			}
 
-			slurmApi.ClearJobCache()
 			err = ccApi.CacheUpdate()
 			if err != nil {
 				trace.Error("Unable to update cc-backend cache. Trying later...")
@@ -270,10 +268,10 @@ func jobEventsProcess() {
 	newJobEvents := make([]prep.SlurmctldEnv, 0)
 
 	// map[cluster]jobId
-	clusterQueries := make(map[string][]uint32, 0)
+	clusterQueries := make(map[string][]int64, 0)
 
 	for _, jobEvent := range jobEvents {
-		jobEventId, err := strconv.ParseUint(jobEvent.SLURM_JOB_ID, 10, 32)
+		jobEventId, err := strconv.ParseInt(jobEvent.SLURM_JOB_ID, 10, 64)
 		if err != nil {
 			trace.Warn("SLURM_JOB_ID contains non-integer value: %v", err)
 			continue
@@ -285,10 +283,12 @@ func jobEventsProcess() {
 			continue
 		}
 
-		clusterQueries[jobEventCluster] = append(clusterQueries[jobEventCluster], uint32(jobEventId))
+		clusterQueries[jobEventCluster] = append(clusterQueries[jobEventCluster], jobEventId)
 	}
 
 	for cluster, jobIds := range clusterQueries {
+		// Strictly speaking, QueryJobs does an sacct call for the job list, which becomes
+		// redundant once we do the squeue call on the job list later. This could be optimized.
 		jobs, err := slurmApi.QueryJobs(cluster, jobIds)
 		if err != nil {
 			jobEventSacctAttempts += 1
@@ -302,11 +302,9 @@ func jobEventsProcess() {
 			}
 		}
 
-		for _, job := range jobs {
-			err := ccApi.SyncJob(job, false)
-			if err != nil {
-				trace.Warn("Syncing job (%s, %d) via PrEp hook failed (we will try again later during regular poll): %v", cluster, job.GetJobId(), err)
-			}
+		err = ccApi.SyncJobs(cluster, jobs, false)
+		if err != nil {
+			trace.Warn("Syncing job (%s, %v) via PrEp hook failed (we will try again later during regular poll): %v", cluster, jobIdsOfJobs(jobs), err)
 		}
 	}
 
@@ -343,12 +341,10 @@ func processSlurmSacctPoll() {
 			return
 		}
 
-		for _, job := range jobs {
-			err = ccApi.SyncJob(job, false)
-			if err != nil {
-				trace.Error("Syncing job to ClusterCockpit failed (%s). Trying later...", err)
-				return
-			}
+		err = ccApi.SyncJobs(cluster, jobs, false)
+		if err != nil {
+			trace.Error("Syncing job (%s, %v) to ClusterCockpit failed (%s). Trying later...", cluster, jobIdsOfJobs(jobs), err)
+			return
 		}
 
 		if len(jobs) > 0 {
@@ -379,7 +375,7 @@ func processSlurmSqueuePoll() {
 		// Check if there are any stale jobs in cc-backend, which are no longer known to Slurm.
 		// This should usually not happen, but in the past Slurm would occasionally lie to use and we would miss
 		// job stops.
-		jobIdsToQuery := make([]uint32, 0)
+		jobIdsToQuery := make([]int64, 0)
 
 		for jobId, cachedJobState := range ccApi.JobCache[cluster] {
 			if !cachedJobState.Running {
@@ -397,31 +393,39 @@ func processSlurmSqueuePoll() {
 				continue
 			}
 
-			jobIdsToQuery = append(jobIdsToQuery, uint32(jobId))
+			jobIdsToQuery = append(jobIdsToQuery, jobId)
 		}
 
 		if len(jobIdsToQuery) == 0 {
 			continue
 		}
 
 		trace.Warn("Detected stale jobs in cc-backend (%s, %v). Trying to synchronize...", cluster, jobIdsToQuery)
-		saJobs, err := slurmApi.QueryJobs(cluster, jobIdsToQuery)
+		jobs, err := slurmApi.QueryJobs(cluster, jobIdsToQuery)
 		if err != nil {
 			trace.Error("Failed to query cc-backend's stale job from Slurm: %v", err)
 			continue
 		}
 
-		for _, job := range saJobs {
-			trace.Warn("Stale job state is: %s", job.GetState())
+		for _, job := range jobs {
+			trace.Warn("Queuing sync of stale job (%s, %d), which is in state '%s'", job.GetCluster(), job.GetJobId(), job.GetState())
+		}
 
-			err = ccApi.SyncJob(job, false)
-			if err != nil {
-				trace.Error("Failed to sync cc-backend's stale job from Slurm: %v", err)
-			}
+		err = ccApi.SyncJobs(cluster, jobs, false)
+		if err != nil {
+			trace.Error("Failed to sync cc-backend's stale job from Slurm: %v", err)
 		}
 	}
 }
 
+func jobIdsOfJobs(jobs []slurm_common.Job) []int64 {
+	result := make([]int64, len(jobs))
+	for i, job := range jobs {
+		result[i] = job.GetJobId()
+	}
+	return result
+}
+
 func lastRunGet() time.Time {
 	statInfo, err := os.Stat(config.Config.LastRunPath)
 	if errors.Is(err, os.ErrNotExist) {

go.mod

@@ -5,7 +5,7 @@ go 1.24.0
 toolchain go1.24.4
 
 require (
-	github.com/ClusterCockpit/cc-lib v1.0.2
+	github.com/ClusterCockpit/cc-lib/v2 v2.4.0
 	github.com/nats-io/nats.go v1.48.0
 )
 

go.sum

@@ -1,5 +1,5 @@
-github.com/ClusterCockpit/cc-lib v1.0.2 h1:ZWn3oZkXgxrr3zSigBdlOOfayZ4Om4xL20DhmritPPg=
-github.com/ClusterCockpit/cc-lib v1.0.2/go.mod h1:UGdOvXEnjFqlnPSxtvtFwO6BtXYW6NnXFoud9FtN93k=
+github.com/ClusterCockpit/cc-lib/v2 v2.4.0 h1:OnZlvqSatg7yCQ2NtSR7AddpUVSiuSMZ8scF1a7nfOk=
+github.com/ClusterCockpit/cc-lib/v2 v2.4.0/go.mod h1:JuxMAuEOaLLNEnnL9U3ejha8kMvsSatLdKPZEgJw6iw=
 github.com/RaveNoX/go-jsoncommentstrip v1.0.0/go.mod h1:78ihd09MekBnJnxpICcwzCMzGrKSKYe4AqU6PDYYpjk=
 github.com/apapsch/go-jsonmerge/v2 v2.0.0 h1:axGnT1gRIfimI7gJifB699GoE/oq+F2MU7Dml6nw9rQ=
 github.com/apapsch/go-jsonmerge/v2 v2.0.0/go.mod h1:lvDnEdqiQrp0O42VQGgmlKpxL1AP2+08jFMw88y4klk=