Add Tag Fields to Jobs

CHANGELOG.md

@@ -9,53 +9,79 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 
 ## Unreleased
 
+### API
+
+#### POST /processes/:processID/execution
+
+- Added optional `tags` array to execution payload. Tags are stored with the job record and can be used to filter jobs when querying `/jobs`.
+
+#### GET /jobs
+
+- Job responses now includes the `tags` field.
+
 ## [0.2.2] - 2025-2-28
 
 ### API
+
 #### POST /processes/:processID/execution
+
 - Execution mode now determined per OGC API - Processes Requirements 25/26: honors `Prefer: respond-async` header when process supports both modes, defaults to sync otherwise
 - Returns `Preference-Applied` response header when async preference is honored
 
 #### GET /admin/resources
+
 - New endpoint to view resource utilization for local jobs (docker, subprocess) and queue status
+
 #### GET /jobs/:jobID/metadata
+
 - Added `recoveryNotice` object to metadata format
 
 ### Features
+
 - Added restart recovery for docker, subprocess, and AWS Batch jobs, including status reconciliation and log handling.
 - Introduced `lost` job status and surfaced it in UI status indicators (job list, job logs, status table).
 - Recovered jobs annotate metadata with a recovery notice and write best-effort metadata when some fields are missing.
 - Resource pool can force-reserve resources for already-running recovered docker jobs.
 
 ### Configuration
+
 - New `MAX_LOCAL_CPUS` and `MAX_LOCAL_MEMORY` environment variables (or `--max-local-cpus` and `--max-local-memory` CLI flags) to set resource limits for local job scheduling
 - Process definitions are validated against these limits at startup and when adding/updating processes via API
 - Processes without explicit resource requirements use default values
 
 ### Documentation
+
 - Added sequence diagram for local scheduler
 - Added Recovery section to DEV_GUIDE
 
 ## [0.2.1] - 2025-12-03
 
 ### API
+
 - Version information is added in landing page.
 
 ### Configuration
+
 - Repository URL is now configurable via `REPO_URL` environment variable. This URL is used for version links and metadata context references.
 
 ### Documentation
+
 - Changelog updated to new format
 
 ## [0.2.0] - 2025-12-02
 
 ### API
+
 #### GET /jobs/:jobID/logs
+
 - In response body `container_logs` key is replaced by `process_logs`
+
 #### PUT|POST /processes/:processID
+
 - Request payload schema has changed (See Process YAML Schema changes below)
 
 ### Process YAML Schema
+
 - `command` is now a first class object and moved outside of `container`
 - `config` object is added
 - `maxResources` and `envVars` are moved under `config` object
@@ -64,21 +90,21 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm
 - `host.type` valid options are changed from `local` | `aws-batch` to `docker` | `aws-batch` | `subprocess`
 
 ### Features
+
 - `subprocess` type processes now can be executed through API. They must be registered like other processes and will be called using OS subprocess calls.
 
 ### Documentation
+
 - A `CHANGELOG.md` file is added in the repo.
 - Process templates are provided for all three host types in `./process_templates` folder
 - Windows setup instructions are added in `README.md`
 
-
 ## [0.1.0] - 2023-07-07
 
 - Initial release with core API endpoints for process and job management
 
 [Unreleased]: https://github.com/Dewberry/sepex/compare/v0.2.2...HEAD
 [0.2.2]: https://github.com/Dewberry/sepex/releases/tag/v0.2.2
 [0.2.1]: https://github.com/Dewberry/sepex/releases/tag/v0.2.1
-
 [0.2.0]: https://github.com/Dewberry/sepex/releases/tag/v0.2.0
 [0.1.0]: https://github.com/Dewberry/sepex/releases/tag/v0.1.0

DEV_GUIDE.md

@@ -1,19 +1,22 @@
 ## Config
+
 - All secrets and configuration settings are handled through environment variables
 - There is an example.env provided to ease the configuration process
 - Command line flags are available for config that is only needed at startup, they take precedence over the environment variables when used.
 - Other configs are defined through env variables so that they can be modified without restarting the server.
 - Here is the resolution order:
-    - Flag, where option is available and used
-    - Environment variable
-    - Default value, where available
+  - Flag, where option is available and used
+  - Environment variable
+  - Default value, where available
 
 ## Process Specific Env
+
 - They must start with ALL CAPS process id.
 - They will be passed to jobs with process id prefix removed. This allow setting 3rd party env variables such as GDAL_NUM_CPUS etc.
 - We are parsing at the job level so as to allow dynamic updates without having to restart server
 
 ## Auth
+
 - If auth is enabled some or all routes are protected based on env variable `AUTH_LEVEL` settings.
 - The middleware validate and parse JWT to verify `X-SEPEX-User-Email` header and inject `X-SEPEX-User-Roles` header.
 - A user can use tools like Postman to set these headers themselves, but if auth is enabled, they will be checked against the token. This setup allows adding submitter info to the database when auth is not enabled.
@@ -25,9 +28,23 @@
 - Only admins can add/update/delete processes.
 
 ## Inputs
-- If `"Inputs": {}` in `/execution` payload. Nothing will be appended to process commands. This allow running processes that do not have any inputs.
+
+- If `"inputs": {}` in `/execution` payload, nothing will be appended to process commands. This allows running processes that do not have any inputs.
+- `"tags"` (optional) can be included as an array of strings in the `/execution` payload. Tags are stored with the job record and can be used to filter jobs when querying `/jobs`. If omitted, the job will be created with an empty tag list.
+
+Example payload:
+
+```JSON
+{
+  "inputs": {
+    "text": "Hello World"
+  },
+  "tags": ["example", "test"]
+}
+```
 
 ## Scope
+
 - The behavior of logging is unknown for AWS Batch processes with job definitions having number of attempts more than 1.
 
 ## Local Scheduler
@@ -54,7 +71,6 @@ Recovery rebuilds in-memory state after a restart using DB non-terminal jobs.
 
 1. Metadata for recovered jobs will be incomplete but present.
 
-
 ## Release/Versioning/Changelog
 
 The project uses an automated release workflow triggered by semver tags (e.g., `v1.0.0`, `v1.0.0-beta`). The workflow validates prerequisites, runs security scans, builds multi-platform container images, and creates GitHub releases with auto-generated release notes.
@@ -68,6 +84,7 @@ The project uses an automated release workflow triggered by semver tags (e.g., `
    - Release workflow fails if version is missing from CHANGELOG.md
 
 2. **Create and Push a Semver Tag**
+
    ```bash
    # For a regular release
    git tag v1.0.0
@@ -91,6 +108,3 @@ The project uses an automated release workflow triggered by semver tags (e.g., `
    - Go to Actions tab → Release workflow → Run workflow
    - Select the tag from the dropdown
    - This is useful for re-running a release if needed
-
-
-

api/handlers/handlers.go

@@ -39,6 +39,7 @@ type jobResponse struct {
 	ProcessID  string      `json:"processID,omitempty"`
 	Message    string      `json:"message,omitempty"`
 	Outputs    interface{} `json:"outputs,omitempty"`
+	Tags       []string    `json:"tags"`
 }
 
 type link struct {
@@ -97,6 +98,7 @@ func prepareResponse(c echo.Context, httpStatus int, renderName string, output i
 // specs: https://developer.ogc.org/api/processes/index.html#tag/Execute
 type runRequestBody struct {
 	Inputs map[string]interface{} `json:"inputs"`
+	Tags   []string               `json:"tags"`
 }
 
 // LandingPage godoc
@@ -190,6 +192,14 @@ func (rh *RESTHandler) Execution(c echo.Context) error {
 	if params.Inputs == nil {
 		return c.JSON(http.StatusBadRequest, errResponse{Message: "'inputs' is required in the body of the request"})
 	}
+	if params.Tags == nil {
+		params.Tags = []string{} // default to empty if not provided
+	}
+
+	err = utils.SanitizeTags(params.Tags)
+	if err != nil {
+		return c.JSON(http.StatusBadRequest, errResponse{Message: err.Error()})
+	}
 
 	err = p.VerifyInputs(params.Inputs)
 	if err != nil {
@@ -246,6 +256,7 @@ func (rh *RESTHandler) Execution(c echo.Context) error {
 			StorageSvc:     rh.StorageSvc,
 			DB:             rh.DB,
 			DoneChan:       rh.MessageQueue.JobDone,
+			Tags:           params.Tags,
 			ResourcePool:   rh.ResourcePool,
 			IsSync:         mode == "sync-execute",
 		}
@@ -265,6 +276,7 @@ func (rh *RESTHandler) Execution(c echo.Context) error {
 			StorageSvc:     rh.StorageSvc,
 			DB:             rh.DB,
 			DoneChan:       rh.MessageQueue.JobDone,
+			Tags:           params.Tags,
 		}
 
 	case "subprocess":
@@ -279,6 +291,7 @@ func (rh *RESTHandler) Execution(c echo.Context) error {
 			StorageSvc:     rh.StorageSvc,
 			DB:             rh.DB,
 			DoneChan:       rh.MessageQueue.JobDone,
+			Tags:           params.Tags,
 			ResourcePool:   rh.ResourcePool,
 			IsSync:         mode == "sync-execute",
 		}
@@ -304,7 +317,7 @@ func (rh *RESTHandler) Execution(c echo.Context) error {
 		c.Response().Header().Set("Preference-Applied", modeResult.PreferenceApplied)
 	}
 
-	resp := jobResponse{ProcessID: j.ProcessID(), Type: "process", JobID: jobID, Status: j.CurrentStatus()}
+	resp := jobResponse{ProcessID: j.ProcessID(), Type: "process", JobID: jobID, Status: j.CurrentStatus(), Tags: params.Tags}
 	switch mode {
 	case "sync-execute":
 		j.Run()
@@ -405,20 +418,31 @@ func (rh *RESTHandler) JobStatusHandler(c echo.Context) (err error) {
 
 	var jRcrd jobs.JobRecord
 	jobID := c.Param("jobID")
+
 	if job, ok := rh.ActiveJobs.Jobs[jobID]; ok {
+		tags := (*job).TAGS()
+		if tags == nil {
+			tags = []string{}
+		}
 		resp := jobResponse{
 			ProcessID:  (*job).ProcessID(),
 			JobID:      (*job).JobID(),
 			LastUpdate: (*job).LastUpdate(),
 			Status:     (*job).CurrentStatus(),
+			Tags:       tags,
 		}
 		return prepareResponse(c, http.StatusOK, "jobStatus", resp)
 	} else if jRcrd, ok, err = rh.DB.GetJob(jobID); ok {
+		if jRcrd.Tags == nil {
+			jRcrd.Tags = []string{}
+		}
+
 		resp := jobResponse{
 			ProcessID:  jRcrd.ProcessID,
 			JobID:      jRcrd.JobID,
 			LastUpdate: jRcrd.LastUpdate,
 			Status:     jRcrd.Status,
+			Tags:       jRcrd.Tags,
 		}
 		return prepareResponse(c, http.StatusOK, "jobStatus", resp)
 	}
@@ -617,6 +641,16 @@ func (rh *RESTHandler) ListJobsHandler(c echo.Context) error {
 	processIDs := c.QueryParam("processID") // assuming comma-separated list: "process1,process2"
 	statuses := c.QueryParam("status")
 	submitters := c.QueryParam("submitter")
+	tagsParam := c.QueryParam("tags")
+	var tagsList []string
+	if tagsParam != "" {
+		for _, t := range strings.Split(tagsParam, ",") {
+			t = strings.TrimSpace(t)
+			if t != "" {
+				tagsList = append(tagsList, t)
+			}
+		}
+	}
 
 	var processIDList []string
 	if processIDs != "" {
@@ -660,7 +694,7 @@ func (rh *RESTHandler) ListJobsHandler(c echo.Context) error {
 		offset = 0
 	}
 
-	result, err := rh.DB.GetJobs(limit, offset, processIDList, statusList, submittersList)
+	result, err := rh.DB.GetJobs(limit, offset, processIDList, statusList, submittersList, tagsList)
 	if err != nil {
 		output := errResponse{HTTPStatus: http.StatusInternalServerError, Message: err.Error()}
 		return prepareResponse(c, http.StatusNotFound, "error", output)
@@ -669,14 +703,14 @@ func (rh *RESTHandler) ListJobsHandler(c echo.Context) error {
 	links := make([]link, 0)
 	if offset != 0 {
 		lnk := link{
-			Href:  fmt.Sprintf("/jobs?offset=%v&limit=%v&processID=%v&status=%v&submitter=%v", offset-limit, limit, processIDs, statuses, submitters),
+			Href:  fmt.Sprintf("/jobs?offset=%v&limit=%v&processID=%v&status=%v&submitter=%v&tags=%v", offset-limit, limit, processIDs, statuses, submitters, tagsParam),
 			Title: "prev",
 		}
 		links = append(links, lnk)
 	}
 	if limit == len(result) {
 		lnk := link{
-			Href:  fmt.Sprintf("/jobs?offset=%v&limit=%v&processID=%v&status=%v&submitter=%v", offset+limit, limit, processIDs, statuses, submitters),
+			Href:  fmt.Sprintf("/jobs?offset=%v&limit=%v&processID=%v&status=%v&submitter=%v&tags=%v", offset+limit, limit, processIDs, statuses, submitters, tagsParam),
 			Title: "next",
 		}
 		links = append(links, lnk)

api/jobs/aws_batch_jobs.go

@@ -38,7 +38,7 @@ type AWSBatchJob struct {
 	UpdateTime     time.Time
 	Status         string `json:"status"`
 	// results       interface{}
-
+	Tags    []string `json:"tags"`
 	logger  *log.Logger
 	logFile *os.File
 
@@ -72,6 +72,10 @@ func (j *AWSBatchJob) ProcessID() string {
 	return j.ProcessName
 }
 
+func (j *AWSBatchJob) TAGS() []string {
+	return j.Tags
+}
+
 func (j *AWSBatchJob) SUBMITTER() string {
 	return j.Submitter
 }
@@ -273,7 +277,7 @@ func (j *AWSBatchJob) Create() error {
 	j.batchContext = batchContext
 
 	// At this point job is ready to be added to database
-	err = j.DB.addJob(j.UUID, "accepted", "", "aws-batch", j.AWSBatchID, j.ProcessName, j.Submitter, time.Now())
+	err = j.DB.addJob(j.UUID, "accepted", "", "aws-batch", j.AWSBatchID, j.ProcessName, j.Submitter, j.Tags, time.Now())
 	if err != nil {
 		j.ctxCancel()
 		return err

api/jobs/database.go

@@ -8,15 +8,15 @@ import (
 
 // Database interface abstracts database operations
 type Database interface {
-	addJob(jid, status, mode, host, hostJobID, processID, submitter string, updated time.Time) error
+	addJob(jid, status, mode, host, hostJobID, processID, submitter string, tags []string, updated time.Time) error
 	updateJobRecord(jid, status string, now time.Time) error
 
 	updateJobHostId(jid, hostJobID string) error
 	GetNonTerminalJobs() ([]JobRecord, error)
 
 	GetJob(jid string) (JobRecord, bool, error)
 	CheckJobExist(jid string) (bool, error)
-	GetJobs(limit, offset int, processIDs, statuses, submitters []string) ([]JobRecord, error)
+	GetJobs(limit, offset int, processIDs, statuses, submitters, tags []string) ([]JobRecord, error)
 	Close() error
 }