feat: make job logic re-runnable

Author: zeidlitz

config.yaml.example

@@ -5,7 +5,6 @@
 USER_ID: "user-1234"
 DATASET_ID: "aa-Dataset-abc"
 DATASET_FOLDER: "DATASET_ABC"
-JOB_EXPECTED_NR_FILES: 0
 CLIENT_ACCESS_TOKEN: "youraccesstoken"
 
 MAIL_ADDRESS: "myemail@example.com"

helpers/helpers.go

@@ -12,7 +12,6 @@ import (
 
 	"github.com/NBISweden/sda-bpctl/cmd"
 	"github.com/NBISweden/sda-bpctl/internal/config"
-	"github.com/NBISweden/sda-bpctl/internal/models"
 	"github.com/spf13/cobra"
 )
 
@@ -146,16 +145,3 @@ func GetFileIDsPath(dataDirectory string, datasetFolder string) string {
 func GetStableIDsPath(dataDirectory string, datasetFolder string) string {
 	return fmt.Sprintf("%s/%s-stableIDs.txt", dataDirectory, datasetFolder)
 }
-
-func GetPathsForAccessionIDs(files []models.FileInfo, datasetFolder string) []string {
-	var paths []string
-	for _, f := range files {
-		if f.Status == "verified" &&
-			strings.Contains(f.InboxPath, datasetFolder) &&
-			!strings.Contains(f.InboxPath, "PRIVATE") {
-			paths = append(paths, f.InboxPath)
-		}
-	}
-	slog.Info("files found for accession id creation", "files_found", len(paths))
-	return paths
-}

helpers/templates/job.template.yaml

@@ -95,7 +95,7 @@ spec:
         - name: xml-data
           secret:
             defaultMode: 288
-            secretName: {{ .JobName }}-xml-secret
+            secretName: job-xml-secret
         {{- end }}
         - name: tls
           secret:

internal/accession/accession.go

@@ -14,6 +14,7 @@ import (
 	"github.com/NBISweden/sda-bpctl/helpers"
 	"github.com/NBISweden/sda-bpctl/internal/client"
 	"github.com/NBISweden/sda-bpctl/internal/config"
+	"github.com/NBISweden/sda-bpctl/internal/models"
 	"github.com/spf13/cobra"
 )
 
@@ -51,17 +52,12 @@ var accessionCmd = &cobra.Command{
 		}
 		defer file.Close() //nolint:errcheck
 
-		files, err := api.GetUsersFilesWithPrefix()
-		if err != nil {
-			return err
-		}
-
-		paths := helpers.GetPathsForAccessionIDs(files, datasetFolder)
+		files, err := api.GetFilesWithStatus("verified")
 		if dryRun {
 			slog.Info("dry run enabled, no accession ids will be created")
 			return nil
 		}
-		accessionIDs, err := postAccessionIDs(api, paths, userID)
+		accessionIDs, err := postAccessionIDs(api, files, userID)
 
 		for _, accessionID := range accessionIDs {
 			if _, err := file.WriteString(accessionID + "\n"); err != nil {
@@ -86,12 +82,7 @@ func init() {
 
 func Run(api client.APIClient, datasetFolder string, userID string) ([]string, error) {
 	slog.Info("starting accession")
-	files, err := api.GetUsersFilesWithPrefix()
-	if err != nil {
-		return nil, err
-	}
-
-	paths := helpers.GetPathsForAccessionIDs(files, datasetFolder)
+	paths, err := api.GetFilesWithStatus("verified")
 	accessionIDs, err := postAccessionIDs(api, paths, userID)
 	if err != nil {
 		return nil, err
@@ -101,17 +92,17 @@ func Run(api client.APIClient, datasetFolder string, userID string) ([]string, e
 	return accessionIDs, nil
 }
 
-func postAccessionIDs(api client.APIClient, paths []string, userID string) ([]string, error) {
+func postAccessionIDs(api client.APIClient, files []models.FileInfo, userID string) ([]string, error) {
 	var accessionIDs []string
-	for _, filepath := range paths {
+	for _, file := range files {
 		accessionID, err := generateAccessionID()
 		if err != nil {
 			return accessionIDs, err
 		}
 
 		payload, err := json.Marshal(map[string]string{
 			"accession_id": accessionID,
-			"filepath":     filepath,
+			"filepath":     file.InboxPath,
 			"user":         userID,
 		})
 		if err != nil {
@@ -128,7 +119,7 @@ func postAccessionIDs(api client.APIClient, paths []string, userID string) ([]st
 		accessionIDs = append(accessionIDs, accessionID)
 	}
 
-	slog.Info("accession IDs assigned", "nr_files", len(paths))
+	slog.Info("accession IDs assigned", "nr_files", len(files))
 	return accessionIDs, nil
 }
 

internal/accession/accession_test.go

@@ -9,14 +9,15 @@ import (
 	"os"
 	"path/filepath"
 	"testing"
+	"time"
 
-	"github.com/NBISweden/sda-bpctl/helpers"
 	"github.com/NBISweden/sda-bpctl/internal/models"
 )
 
 type mockClient struct {
-	FilesToReturn []models.FileInfo
-	Response      *http.Response
+	FilesToReturn   []models.FileInfo
+	FilesWithStatus []models.FileInfo
+	Response        *http.Response
 }
 
 func (m *mockClient) GetUsersFilesWithPrefix() ([]models.FileInfo, error) {
@@ -51,14 +52,26 @@ func (m *mockClient) PostDatasetCreate(payload []byte) ([]byte, error) {
 	return response, nil
 }
 
+func (m *mockClient) GetFilesWithStatus(status string) ([]models.FileInfo, error) {
+	return m.FilesWithStatus, nil
+}
+
+func (m *mockClient) WaitForStatus(target int, status string, interval time.Duration, timeout time.Duration) ([]models.FileInfo, error) {
+	return nil, nil
+}
+
 func newMockClient(userID string, datasetFolder string) *mockClient {
 	mock := &mockClient{
 		FilesToReturn: []models.FileInfo{
 			// Data is mocked so we expect 2 files to be included in the accession (file1.c4gh and file2.c4gh)
-			{InboxPath: fmt.Sprintf("/%s/%s/file1.c4gh", userID, datasetFolder), Status: "verified"},
-			{InboxPath: fmt.Sprintf("/%s/%s/file2.c4gh", userID, datasetFolder), Status: "verified"},
+			{InboxPath: fmt.Sprintf("/%s/%s/file1.c4gh", userID, datasetFolder), Status: "ready"},
+			{InboxPath: fmt.Sprintf("/%s/%s/file2.c4gh", userID, datasetFolder), Status: "ready"},
 			{InboxPath: fmt.Sprintf("/%s/%s/file3.c4gh", userID, datasetFolder), Status: "error"},
-			{InboxPath: fmt.Sprintf("/%s/%s/file4.c4gh", userID, "DATASET_OTHER"), Status: "verified"},
+			{InboxPath: fmt.Sprintf("/%s/%s/file4.c4gh", userID, "DATASET_OTHER"), Status: "ready"},
+		},
+		FilesWithStatus: []models.FileInfo{
+			{InboxPath: fmt.Sprintf("/%s/%s/file1.c4gh", userID, datasetFolder), Status: "ready"},
+			{InboxPath: fmt.Sprintf("/%s/%s/file2.c4gh", userID, datasetFolder), Status: "ready"},
 		},
 		Response: &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(bytes.NewBufferString("ok"))},
 	}
@@ -80,19 +93,19 @@ func TestAccession(t *testing.T) {
 	t.Run("Test Accession", func(t *testing.T) {
 
 		accessionCmd.Flag("data-directory").Value.Set(workingDirectory)
-		files, err := mock.GetUsersFiles()
+
+		files, err := mock.GetFilesWithStatus("ready")
 		if err != nil {
-			t.Error(err)
+			t.Fail()
 		}
 
-		paths := helpers.GetPathsForAccessionIDs(files, datasetFolder)
-		nrPaths := len(paths)
-		if nrPaths != expectedPaths {
-			t.Logf("recieved %d/%d paths for accessionIDs", nrPaths, expectedPaths)
+		nrFiles := len(files)
+		if nrFiles != expectedPaths {
+			t.Logf("recieved %d/%d paths for accessionIDs", nrFiles, expectedPaths)
 			t.Fail()
 		}
 
-		accessionIDs, err := postAccessionIDs(mock, paths, userID)
+		accessionIDs, err := postAccessionIDs(mock, files, userID)
 		nrAccessionIDs := len(accessionIDs)
 		if nrAccessionIDs != expectedPaths {
 			t.Logf("recieved %d/%d accessionIDs", nrAccessionIDs, expectedPaths)

internal/client/client.go

@@ -178,39 +178,42 @@ func (c *Client) doRequest(method, path string, body []byte) ([]byte, error) {
 	return responseBody, nil
 }
 
-func (c *Client) WaitForAccession(target int, interval time.Duration, timeout time.Duration) ([]string, error) {
+func (c *Client) WaitForStatus(target int, status string, interval time.Duration, timeout time.Duration) ([]models.FileInfo, error) {
 	deadline := time.Now().Add(timeout)
 	for {
-		paths, err := c.getVerifiedFilePaths()
+
+		filteredFiles, err := c.GetFilesWithStatus(status)
 		if err != nil {
 			return nil, err
 		}
 
-		if len(paths) >= target {
-			return paths, nil
+		current := len(filteredFiles)
+
+		if current >= target {
+			return filteredFiles, nil
 		}
 
 		if time.Now().After(deadline) {
-			return nil, fmt.Errorf("timeout reached, only got %d/%d files", len(paths), target)
+			return nil, fmt.Errorf("timeout reached, only got %d/%d files", current, target)
 		}
-		slog.Info(fmt.Sprintf("found %d/%d files - waiting: %s timeout: %s", len(paths), target, interval, timeout))
+		slog.Info(fmt.Sprintf("found %d/%d files - waiting: %s timeout: %s", current, target, interval, timeout))
 		time.Sleep(interval)
 	}
 }
 
-func (c *Client) getVerifiedFilePaths() ([]string, error) {
-	files, err := c.GetUsersFilesWithPrefix()
+func (c *Client) GetFilesWithStatus(status string) ([]models.FileInfo, error) {
+	allFiles, err := c.GetUsersFilesWithPrefix()
 	if err != nil {
 		return nil, err
 	}
 
-	var paths []string
-	for _, f := range files {
-		if f.Status == "verified" &&
+	var filteredFiles []models.FileInfo
+	for _, f := range allFiles {
+		if f.Status == status &&
 			strings.Contains(f.InboxPath, c.datasetFolder) &&
 			!strings.Contains(f.InboxPath, "PRIVATE") {
-			paths = append(paths, f.InboxPath)
+			filteredFiles = append(filteredFiles, f)
 		}
 	}
-	return paths, nil
+	return filteredFiles, nil
 }

internal/client/client_interface.go

@@ -2,6 +2,7 @@ package client
 
 import (
 	"github.com/NBISweden/sda-bpctl/internal/models"
+	"time"
 )
 
 type APIClient interface {
@@ -10,4 +11,6 @@ type APIClient interface {
 	PostFileIngest([]byte) ([]byte, error)
 	PostFileAccession(payload []byte) ([]byte, error)
 	PostDatasetCreate(payload []byte) ([]byte, error)
+	GetFilesWithStatus(status string) ([]models.FileInfo, error)
+	WaitForStatus(target int, status string, interval time.Duration, timeout time.Duration) ([]models.FileInfo, error)
 }

internal/config/config.go

@@ -95,7 +95,6 @@ func bindKeys(v *viper.Viper) {
 func setDefaults(v *viper.Viper) {
 	v.SetDefault("JOB_TIMEOUT", 4320)
 	v.SetDefault("JOB_POLL_RATE", 180)
-	v.SetDefault("JOB_EXPECTED_NR_FILES", 0)
 	v.SetDefault("JOB_DATA_DIRECTORY", "/data")
 
 	v.SetDefault("CLIENT_API_HOST", "https://api.bp.nbis.se")

internal/dataset/dataset.go

@@ -1,7 +1,6 @@
 package dataset
 
 import (
-	"bufio"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -57,7 +56,7 @@ var datasetCmd = &cobra.Command{
 			}
 		}
 
-		fileIDsList, err := getFileIDs(datasetFolder, api)
+		fileIDsList, err := api.GetFilesWithStatus("verified")
 		if err != nil {
 			return err
 		}
@@ -95,10 +94,9 @@ type UserFiles struct {
 	InboxPath   string `json:"inboxPath"`
 }
 
-func Run(api *client.Client, datasetFolder string, datasetID string, userID string, fileIDsList []string) error {
+func Run(api *client.Client, datasetFolder string, datasetID string, userID string) error {
 
-	// TODO: Optimize this; another query to GetUsersFilesWithPrefix is redundant since It will hold the same information as in the fileIDsList []string
-	files, err := api.GetUsersFilesWithPrefix()
+	files, err := api.GetFilesWithStatus("ready")
 	if err != nil {
 		return err
 	}
@@ -108,51 +106,33 @@ func Run(api *client.Client, datasetFolder string, datasetID string, userID stri
 		return err
 	}
 
-	err = createDataset(api, datasetID, userID, fileIDsList)
+	err = createDataset(api, datasetID, userID, files)
 	if err != nil {
 		return err
 	}
 	return nil
 }
 
-func getFileIDs(datasetFolder string, api client.APIClient) ([]string, error) {
-	var fileIDsList []string
-	filePath := helpers.GetFileIDsPath(DataDirectory, datasetFolder)
-	if _, err := os.Stat(filePath); errors.Is(err, os.ErrNotExist) {
-		files, err := api.GetUsersFilesWithPrefix()
-		if err != nil {
-			return nil, err
-		}
-		fileIDsList = helpers.GetPathsForAccessionIDs(files, datasetFolder)
-		return fileIDsList, nil
-	}
-	file, err := os.Open(filePath)
-	if err != nil {
-		return nil, err
+func createDataset(api client.APIClient, datasetID string, userID string, files []models.FileInfo) error {
+	var accessionIDs []string
+	for _, file := range files {
+		accessionIDs = append(accessionIDs, file.AccessionID)
 	}
-	defer file.Close() //nolint:errcheck
 
-	slog.Info("reading", "filePath", filePath)
-	scanner := bufio.NewScanner(file)
-	for scanner.Scan() {
-		fileIDsList = append(fileIDsList, scanner.Text())
-	}
-	return fileIDsList, nil
-}
+	nrFiles := len(accessionIDs)
 
-func createDataset(api client.APIClient, datasetID string, userID string, fileIDsList []string) error {
-	slog.Info("starting dataset")
+	slog.Info("starting dataset", "nr_files", nrFiles)
 
-	if len(fileIDsList) > 100 {
-		err := sendInChunks(fileIDsList, api, datasetID, userID)
+	if nrFiles > 100 {
+		err := sendInChunks(accessionIDs, api, datasetID, userID)
 		if err != nil {
 			return err
 		}
 	}
 
-	if len(fileIDsList) <= 100 {
+	if len(files) <= 100 {
 		payload := Payload{
-			AccessionIDs: fileIDsList,
+			AccessionIDs: accessionIDs,
 			DatasetID:    datasetID,
 			User:         userID,
 		}