Merge pull request #299 from ClusterCockpit/dev

feat: Add buffered channel with worker thread for job start API
ClusterCockpit · Nov 27, 2024 · ab284ed · ab284ed
2 parents 16b11db + 00a5786
commit ab284ed
Show file tree

Hide file tree

Showing 29 changed files with 637 additions and 659 deletions.
diff --git a/api/swagger.json b/api/swagger.json
diff --git a/api/swagger.yaml b/api/swagger.yaml
diff --git a/cmd/cc-backend/server.go b/cmd/cc-backend/server.go
@@ -25,6 +25,7 @@ import (
 	"github.com/ClusterCockpit/cc-backend/internal/config"
 	"github.com/ClusterCockpit/cc-backend/internal/graph"
 	"github.com/ClusterCockpit/cc-backend/internal/graph/generated"
+	"github.com/ClusterCockpit/cc-backend/internal/repository"
 	"github.com/ClusterCockpit/cc-backend/internal/routerConfig"
 	"github.com/ClusterCockpit/cc-backend/pkg/log"
 	"github.com/ClusterCockpit/cc-backend/pkg/runtimeEnv"
@@ -109,9 +110,7 @@ func serverInit() {
 
 	if !config.Keys.DisableAuthentication {
 		router.Handle("/login", authHandle.Login(
-			// On success:
-			http.RedirectHandler("/", http.StatusTemporaryRedirect),
-
+			// On success: Handled within Login()
 			// On failure:
 			func(rw http.ResponseWriter, r *http.Request, err error) {
 				rw.Header().Add("Content-Type", "text/html; charset=utf-8")
@@ -126,9 +125,7 @@ func serverInit() {
 			})).Methods(http.MethodPost)
 
 		router.Handle("/jwt-login", authHandle.Login(
-			// On success:
-			http.RedirectHandler("/", http.StatusTemporaryRedirect),
-
+			// On success: Handled within Login()
 			// On failure:
 			func(rw http.ResponseWriter, r *http.Request, err error) {
 				rw.Header().Add("Content-Type", "text/html; charset=utf-8")
@@ -164,11 +161,12 @@ func serverInit() {
 				func(rw http.ResponseWriter, r *http.Request, err error) {
 					rw.WriteHeader(http.StatusUnauthorized)
 					web.RenderTemplate(rw, "login.tmpl", &web.Page{
-						Title:   "Authentication failed - ClusterCockpit",
-						MsgType: "alert-danger",
-						Message: err.Error(),
-						Build:   buildInfo,
-						Infos:   info,
+						Title:    "Authentication failed - ClusterCockpit",
+						MsgType:  "alert-danger",
+						Message:  err.Error(),
+						Build:    buildInfo,
+						Infos:    info,
+						Redirect: r.RequestURI,
 					})
 				})
 		})
@@ -316,6 +314,9 @@ func serverShutdown() {
 	// First shut down the server gracefully (waiting for all ongoing requests)
 	server.Shutdown(context.Background())
 
+	// Then, wait for any async jobStarts still pending...
+	repository.WaitForJobStart()
+
 	// Then, wait for any async archivings still pending...
 	archiver.WaitForArchiving()
 }
diff --git a/internal/api/api_test.go b/internal/api/api_test.go
@@ -14,9 +14,9 @@ import (
 	"os"
 	"path/filepath"
 	"reflect"
-	"strconv"
 	"strings"
 	"testing"
+	"time"
 
 	"github.com/ClusterCockpit/cc-backend/internal/api"
 	"github.com/ClusterCockpit/cc-backend/internal/archiver"
@@ -200,6 +200,10 @@ func TestRestApi(t *testing.T) {
 	r.StrictSlash(true)
 	restapi.MountApiRoutes(r)
 
+	var TestJobId int64 = 123
+	var TestClusterName string = "testcluster"
+	var TestStartTime int64 = 123456789
+
 	const startJobBody string = `{
         "jobId":            123,
 		"user":             "testuser",
@@ -225,7 +229,6 @@ func TestRestApi(t *testing.T) {
 		"startTime": 123456789
 	}`
 
-	var dbid int64
 	const contextUserKey repository.ContextKey = "user"
 	contextUserValue := &schema.User{
 		Username:   "testuser",
@@ -247,13 +250,10 @@ func TestRestApi(t *testing.T) {
 			t.Fatal(response.Status, recorder.Body.String())
 		}
 
-		var res api.StartJobApiResponse
-		if err := json.Unmarshal(recorder.Body.Bytes(), &res); err != nil {
-			t.Fatal(err)
-		}
+		time.Sleep(1 * time.Second)
 
 		resolver := graph.GetResolverInstance()
-		job, err := resolver.Query().Job(ctx, strconv.Itoa(int(res.DBID)))
+		job, err := restapi.JobRepository.Find(&TestJobId, &TestClusterName, &TestStartTime)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -285,8 +285,6 @@ func TestRestApi(t *testing.T) {
 		if len(job.Tags) != 1 || job.Tags[0].Type != "testTagType" || job.Tags[0].Name != "testTagName" || job.Tags[0].Scope != "testuser" {
 			t.Fatalf("unexpected tags: %#v", job.Tags)
 		}
-
-		dbid = res.DBID
 	}); !ok {
 		return
 	}
@@ -314,8 +312,7 @@ func TestRestApi(t *testing.T) {
 		}
 
 		archiver.WaitForArchiving()
-		resolver := graph.GetResolverInstance()
-		job, err := resolver.Query().Job(ctx, strconv.Itoa(int(dbid)))
+		job, err := restapi.JobRepository.Find(&TestJobId, &TestClusterName, &TestStartTime)
 		if err != nil {
 			t.Fatal(err)
 		}
@@ -404,8 +401,10 @@ func TestRestApi(t *testing.T) {
 		t.Fatal("subtest failed")
 	}
 
+	time.Sleep(1 * time.Second)
+
 	const stopJobBodyFailed string = `{
-        "jobId":     12345,
+    "jobId":     12345,
 		"cluster":   "testcluster",
 
 		"jobState": "failed",