Preserving new fields in job struct

pixlise · Jun 14, 2024 · 1e71fe8 · 1e71fe8
1 parent 5221eb4
commit 1e71fe8
Show file tree

Hide file tree

Showing 7 changed files with 36 additions and 9 deletions.
diff --git a/api/coreg/import.go b/api/coreg/import.go
@@ -39,7 +39,7 @@ func StartCoregImport(triggerUrl string, hctx wsHelpers.HandlerContext) (string,
 
 	// Start an image coreg import job (this is a Lambda function)
 	// Once it completes, we have the data we need, so we can treat it as a "normal" image importing task
-	jobStatus, err := job.AddJob("coreg", hctx.SessUser.User.Id, protos.JobStatus_JT_IMPORT_IMAGE, triggerUrl, uint32(hctx.Svcs.Config.ImportJobMaxTimeSec), hctx.Svcs.MongoDB, hctx.Svcs.IDGen, hctx.Svcs.TimeStamper, hctx.Svcs.Log, i.sendUpdate)
+	jobStatus, err := job.AddJob("coreg", hctx.SessUser.User.Id, protos.JobStatus_JT_IMPORT_IMAGE, triggerUrl, fmt.Sprintf("Import: %v", path.Base(triggerUrl)), []string{}, uint32(hctx.Svcs.Config.ImportJobMaxTimeSec), hctx.Svcs.MongoDB, hctx.Svcs.IDGen, hctx.Svcs.TimeStamper, hctx.Svcs.Log, i.sendUpdate)
 	jobId := ""
 	if jobStatus != nil {
 		jobId = jobStatus.JobId

diff --git a/api/job/jobWatcher.go b/api/job/jobWatcher.go
@@ -22,11 +22,14 @@ var activeJobLock = sync.Mutex{}
 // Expected to be called by API to create the initial record of a job. It can then trigger it however it needs to
 // (eg AWS lambda or running PIQUANT nodes) and this sticks around monitoring the DB entry for changes, calling
 // the sendUpdate callback function on change. Returns the snapshot of the "added" job that was saved
+
 func AddJob(
 	idPrefix string,
 	requestorUserId string,
 	jobType protos.JobStatus_JobType,
 	jobItemId string,
+	jobName string,
+	elementList []string, // optional, only set if it's a quant!
 	jobTimeoutSec uint32,
 	db *mongo.Database,
 	idgen idgen.IDGenerator,
@@ -50,6 +53,8 @@ func AddJob(
 		OtherLogFiles:    []string{},
 		JobType:          jobType,
 		JobItemId:        jobItemId,
+		Name:             jobName,
+		Elements:         elementList,
 	}
 
 	if _, ok := activeJobs[jobId]; ok {

diff --git a/api/job/statusChanges.go b/api/job/statusChanges.go
@@ -34,10 +34,11 @@ func UpdateJob(jobId string, status protos.JobStatus_Status, message string, log
 		logger.Errorf("Failed to read existing job status when writing UpdateJob %v: %v", jobId, err)
 	} else {
 		jobStatus.StartUnixTimeSec = existingStatus.StartUnixTimeSec
-
 		jobStatus.JobType = existingStatus.JobType
 		jobStatus.JobItemId = existingStatus.JobItemId
 		jobStatus.RequestorUserId = existingStatus.RequestorUserId
+		jobStatus.Name = existingStatus.Name
+		jobStatus.Elements = existingStatus.Elements
 	}
 
 	replaceResult, err := coll.ReplaceOne(ctx, filter, jobStatus, opt)
@@ -90,10 +91,11 @@ func CompleteJob(jobId string, success bool, message string, outputFilePath stri
 	} else {
 		jobStatus.LogId = existingStatus.LogId
 		jobStatus.StartUnixTimeSec = existingStatus.StartUnixTimeSec
-
 		jobStatus.JobType = existingStatus.JobType
 		jobStatus.JobItemId = existingStatus.JobItemId
 		jobStatus.RequestorUserId = existingStatus.RequestorUserId
+		jobStatus.Name = existingStatus.Name
+		jobStatus.Elements = existingStatus.Elements
 	}
 
 	replaceResult, err := coll.ReplaceOne(ctx, filter, jobStatus, opt)

diff --git a/api/quantification/create.go b/api/quantification/create.go
@@ -64,7 +64,7 @@ func CreateJob(createParams *protos.QuantCreateParams, requestorUserId string, s
 		// Make the name and ID the same, and start with something that stands out
 		jobId = fmt.Sprintf("cmd-%v-%s", createParams.Command, svcs.IDGen.GenObjectID())
 	} else {
-		jobStatus, err = job.AddJob("quant", requestorUserId, protos.JobStatus_JT_RUN_QUANT, "", uint32(svcs.Config.ImportJobMaxTimeSec), svcs.MongoDB, svcs.IDGen, svcs.TimeStamper, svcs.Log, sendUpdate)
+		jobStatus, err = job.AddJob("quant", requestorUserId, protos.JobStatus_JT_RUN_QUANT, "", createParams.Name, createParams.Elements, uint32(svcs.Config.ImportJobMaxTimeSec), svcs.MongoDB, svcs.IDGen, svcs.TimeStamper, svcs.Log, sendUpdate)
 		if jobStatus != nil {
 			jobId = jobStatus.JobId
 		}

diff --git a/api/ws/handlers/scan.go b/api/ws/handlers/scan.go
@@ -305,7 +305,7 @@ func HandleScanTriggerReImportReq(req *protos.ScanTriggerReImportReq, hctx wsHel
 		hctx.Svcs.MongoDB,
 	}
 
-	jobStatus, err := job.AddJob("reimport", hctx.SessUser.User.Id, protos.JobStatus_JT_REIMPORT_SCAN, req.ScanId, uint32(hctx.Svcs.Config.ImportJobMaxTimeSec), hctx.Svcs.MongoDB, hctx.Svcs.IDGen, hctx.Svcs.TimeStamper, hctx.Svcs.Log, i.sendReimportUpdate)
+	jobStatus, err := job.AddJob("reimport", hctx.SessUser.User.Id, protos.JobStatus_JT_REIMPORT_SCAN, req.ScanId, fmt.Sprintf("Reimport: %v", req.ScanId), []string{}, uint32(hctx.Svcs.Config.ImportJobMaxTimeSec), hctx.Svcs.MongoDB, hctx.Svcs.IDGen, hctx.Svcs.TimeStamper, hctx.Svcs.Log, i.sendReimportUpdate)
 	jobId := ""
 	if jobStatus != nil {
 		jobId = jobStatus.JobId
@@ -544,7 +544,7 @@ func HandleScanUploadReq(req *protos.ScanUploadReq, hctx wsHelpers.HandlerContex
 	}
 
 	// Add a job watcher for this
-	jobStatus, err := job.AddJob("import", hctx.SessUser.User.Id, protos.JobStatus_JT_IMPORT_SCAN, datasetID, uint32(hctx.Svcs.Config.ImportJobMaxTimeSec), hctx.Svcs.MongoDB, hctx.Svcs.IDGen, hctx.Svcs.TimeStamper, hctx.Svcs.Log, i.sendImportUpdate)
+	jobStatus, err := job.AddJob("import", hctx.SessUser.User.Id, protos.JobStatus_JT_IMPORT_SCAN, datasetID, fmt.Sprintf("Import: %v", datasetID), []string{}, uint32(hctx.Svcs.Config.ImportJobMaxTimeSec), hctx.Svcs.MongoDB, hctx.Svcs.IDGen, hctx.Svcs.TimeStamper, hctx.Svcs.Log, i.sendImportUpdate)
 	jobId := ""
 	if jobStatus != nil {
 		jobId = jobStatus.JobId

diff --git a/data-formats b/data-formats
diff --git a/generated-protos/job.pb.go b/generated-protos/job.pb.go