Add job delete support

go.mod

@@ -34,6 +34,7 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/gophercloud/gophercloud v1.14.0
 	github.com/hashicorp/go-retryablehttp v0.7.7
+	github.com/jackc/pgconn v1.14.3
 	github.com/jackc/pgtype v1.14.3
 	github.com/jackc/pgx/v4 v4.18.3
 	github.com/julienschmidt/httprouter v1.3.0
@@ -165,7 +166,6 @@ require (
 	github.com/hashicorp/go-version v1.7.0 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
-	github.com/jackc/pgconn v1.14.3 // indirect
 	github.com/jackc/pgio v1.0.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgproto3/v2 v2.3.3 // indirect

internal/cloudapi/v2/errors.go

@@ -78,6 +78,8 @@ const (
 	ErrorTenantNotInContext                       ServiceErrorCode = 1020
 	ErrorGettingComposeList                       ServiceErrorCode = 1021
 	ErrorArtifactNotFound                         ServiceErrorCode = 1022
+	ErrorDeletingJob                              ServiceErrorCode = 1023
+	ErrorDeletingArtifacts                        ServiceErrorCode = 1024
 
 	// Errors contained within this file
 	ErrorUnspecified          ServiceErrorCode = 10000
@@ -163,6 +165,8 @@ func getServiceErrors() serviceErrors {
 		serviceError{ErrorGettingJobType, http.StatusInternalServerError, "Unable to get job type of existing job"},
 		serviceError{ErrorTenantNotInContext, http.StatusInternalServerError, "Unable to retrieve tenant from request context"},
 		serviceError{ErrorGettingComposeList, http.StatusInternalServerError, "Unable to get list of composes"},
+		serviceError{ErrorDeletingJob, http.StatusInternalServerError, "Unable to delete job"},
+		serviceError{ErrorDeletingArtifacts, http.StatusInternalServerError, "Unable to delete job artifacts"},
 
 		serviceError{ErrorUnspecified, http.StatusInternalServerError, "Unspecified internal error "},
 		serviceError{ErrorNotHTTPError, http.StatusInternalServerError, "Error is not an instance of HTTPError"},

internal/cloudapi/v2/handler.go

@@ -307,13 +307,13 @@ func (h *apiHandlers) targetResultToUploadStatus(jobId uuid.UUID, t *target.Targ
 
 // GetComposeList returns a list of the root job UUIDs
 func (h *apiHandlers) GetComposeList(ctx echo.Context) error {
-	jobs, err := h.server.workers.AllRootJobIDs()
+	jobs, err := h.server.workers.AllRootJobIDs(ctx.Request().Context())
 	if err != nil {
 		return HTTPErrorWithInternal(ErrorGettingComposeList, err)
 	}
 
 	// Gather up the details of each job
-	var stats []ComposeStatus
+	stats := []ComposeStatus{}
 	for _, jid := range jobs {
 		s, err := h.getJobIDComposeStatus(jid)
 		if err != nil {
@@ -329,6 +329,34 @@ func (h *apiHandlers) GetComposeList(ctx echo.Context) error {
 	return ctx.JSON(http.StatusOK, stats)
 }
 
+// DeleteCompose deletes a compose by UUID
+func (h *apiHandlers) DeleteCompose(ctx echo.Context, jobId uuid.UUID) error {
+	return h.server.EnsureJobChannel(h.deleteComposeImpl)(ctx, jobId)
+}
+
+func (h *apiHandlers) deleteComposeImpl(ctx echo.Context, jobId uuid.UUID) error {
+	_, err := h.server.workers.JobType(jobId)
+	if err != nil {
+		return HTTPError(ErrorComposeNotFound)
+	}
+
+	err = h.server.workers.DeleteJob(ctx.Request().Context(), jobId)
+	if err != nil {
+		return HTTPErrorWithInternal(ErrorDeletingJob, err)
+	}
+
+	err = h.server.workers.CleanupArtifacts()
+	if err != nil {
+		return HTTPErrorWithInternal(ErrorDeletingArtifacts, err)
+	}
+
+	return ctx.JSON(http.StatusOK, ComposeDeleteStatus{
+		Href: fmt.Sprintf("/api/image-builder-composer/v2/composes/delete/%v", jobId),
+		Id:   jobId.String(),
+		Kind: "ComposeDeleteStatus",
+	})
+}
+
 func (h *apiHandlers) GetComposeStatus(ctx echo.Context, jobId uuid.UUID) error {
 	return h.server.EnsureJobChannel(h.getComposeStatusImpl)(ctx, jobId)
 }

internal/cloudapi/v2/openapi.v2.yml

@@ -133,6 +133,60 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/Error'
+    delete:
+      operationId: deleteCompose
+      summary: Delete a compose
+      security:
+        - Bearer: []
+      parameters:
+        - in: path
+          name: id
+          schema:
+            type: string
+            format: uuid
+            example: '123e4567-e89b-12d3-a456-426655440000'
+          required: true
+          description: ID of compose to delete
+      description: |-
+        Delete a compose and all of its independent jobs.
+      responses:
+        '200':
+          description: compose delete status
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ComposeDeleteStatus'
+        '400':
+          description: Invalid compose id
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Error'
+        '401':
+          description: Auth token is invalid
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Error'
+        '403':
+          description: Unauthorized to perform operation
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Error'
+        '404':
+          description: Unknown compose id
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Error'
+        '500':
+          description: Unexpected error occurred
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Error'
+
 
   /composes/{id}/metadata:
     get:
@@ -805,6 +859,10 @@ components:
         - failure
         - pending
       example: success
+
+    ComposeDeleteStatus:
+      $ref: '#/components/schemas/ObjectReference'
+
     ComposeLogs:
       allOf:
       - $ref: '#/components/schemas/ObjectReference'

internal/cloudapi/v2/v2_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"net/http"
 	"os"
+	"path/filepath"
 	"sync"
 	"testing"
 	"time"
@@ -241,11 +242,19 @@ func mockSearch(t *testing.T, workerServer *worker.Server, wg *sync.WaitGroup, f
 }
 
 func newV2Server(t *testing.T, dir string, enableJWT bool, fail bool) (*v2.Server, *worker.Server, jobqueue.JobQueue, context.CancelFunc) {
-	q, err := fsjobqueue.New(dir)
+	jobsDir := filepath.Join(dir, "jobs")
+	err := os.Mkdir(jobsDir, 0755)
 	require.NoError(t, err)
+	q, err := fsjobqueue.New(jobsDir)
+	require.NoError(t, err)
+
+	artifactsDir := filepath.Join(dir, "artifacts")
+	err = os.Mkdir(artifactsDir, 0755)
+	require.NoError(t, err)
+
 	workerServer := worker.NewServer(nil, q,
 		worker.Config{
-			ArtifactsDir:         t.TempDir(),
+			ArtifactsDir:         artifactsDir,
 			BasePath:             "/api/worker/v1",
 			JWTEnabled:           enableJWT,
 			TenantProviderFields: []string{"rh-org-id", "account_id"},
@@ -1878,6 +1887,10 @@ func TestComposesRoute(t *testing.T) {
 	srv, _, _, cancel := newV2Server(t, t.TempDir(), false, false)
 	defer cancel()
 
+	// List empty root composes
+	test.TestRoute(t, srv.Handler("/api/image-builder-composer/v2"), false, "GET", "/api/image-builder-composer/v2/composes/", ``,
+		http.StatusOK, `[]`)
+
 	// Make a compose so it has something to list
 	reply := test.TestRouteWithReply(t, srv.Handler("/api/image-builder-composer/v2"), false, "POST", "/api/image-builder-composer/v2/compose", fmt.Sprintf(`
 	{
@@ -2127,3 +2140,65 @@ func TestComposeRequestMetadata(t *testing.T) {
 		"request": %s
 	}`, jobId, request))
 }
+
+func TestComposesDeleteRoute(t *testing.T) {
+	srv, wrksrv, _, cancel := newV2Server(t, t.TempDir(), false, false)
+	defer cancel()
+
+	// Make a compose so it has something to list and delete
+	reply := test.TestRouteWithReply(t, srv.Handler("/api/image-builder-composer/v2"), false, "POST", "/api/image-builder-composer/v2/compose", fmt.Sprintf(`
+	{
+		"distribution": "%s",
+		"image_request":{
+			"architecture": "%s",
+			"image_type": "%s",
+			"repositories": [{
+				"baseurl": "somerepo.org",
+				"rhsm": false
+			}],
+			"upload_options": {
+				"region": "eu-central-1",
+				"snapshot_name": "name",
+				"share_with_accounts": ["123456789012","234567890123"]
+			}
+		}
+	}`, test_distro.TestDistro1Name, test_distro.TestArch3Name, string(v2.ImageTypesAws)), http.StatusCreated, `
+	{
+		"href": "/api/image-builder-composer/v2/compose",
+		"kind": "ComposeId"
+	}`, "id")
+
+	// Extract the compose ID to use to test the list response
+	var composeReply v2.ComposeId
+	err := json.Unmarshal(reply, &composeReply)
+	require.NoError(t, err)
+	jobID := composeReply.Id
+
+	_, token, jobType, _, _, err := wrksrv.RequestJob(context.Background(), test_distro.TestArch3Name, []string{worker.JobTypeOSBuild}, []string{""}, uuid.Nil)
+	require.NoError(t, err)
+	require.Equal(t, worker.JobTypeOSBuild, jobType)
+	res, err := json.Marshal(&worker.OSBuildJobResult{
+		Success:       true,
+		OSBuildOutput: &osbuild.Result{Success: true},
+	})
+	require.NoError(t, err)
+	err = wrksrv.FinishJob(token, res)
+	require.NoError(t, err)
+
+	// List root composes
+	test.TestRoute(t, srv.Handler("/api/image-builder-composer/v2"), false, "GET", "/api/image-builder-composer/v2/composes/", ``,
+		http.StatusOK, fmt.Sprintf(`[{"href":"/api/image-builder-composer/v2/composes/%[1]s", "id":"%[1]s", "image_status":{"status":"success"}, "kind":"ComposeStatus", "status":"success"}]`,
+			jobID.String()))
+
+	// Delete the compose
+	test.TestRoute(t, srv.Handler("/api/image-builder-composer/v2"), false, "DELETE", fmt.Sprintf("/api/image-builder-composer/v2/composes/%s", jobID.String()), ``,
+		http.StatusOK, fmt.Sprintf(`
+	{
+		"id": "%s",
+		"kind": "ComposeDeleteStatus"
+	}`, jobID.String()), "href")
+
+	// List root composes (should now be none)
+	test.TestRoute(t, srv.Handler("/api/image-builder-composer/v2"), false, "GET", "/api/image-builder-composer/v2/composes/", ``,
+		http.StatusOK, `[]`)
+}

internal/jobqueue/fsjobqueue/fsjobqueue.go

@@ -705,7 +705,7 @@ func jobMatchesCriteria(j *job, acceptedJobTypes []string, acceptedChannels []st
 
 // AllRootJobIDs Return a list of all the top level(root) job uuids
 // This only includes jobs without any Dependents set
-func (q *fsJobQueue) AllRootJobIDs() ([]uuid.UUID, error) {
+func (q *fsJobQueue) AllRootJobIDs(_ context.Context) ([]uuid.UUID, error) {
 	ids, err := q.db.List()
 	if err != nil {
 		return nil, err
@@ -726,3 +726,48 @@ func (q *fsJobQueue) AllRootJobIDs() ([]uuid.UUID, error) {
 
 	return jobIDs, nil
 }
+
+// DeleteJob will delete a job and all of its dependencies
+// If a dependency has multiple depenents it will only delete the parent job from
+// the dependants list and then re-save the job instead of removing it.
+//
+// This assumes that the jobs have been created correctly, and that they have
+// no dependency loops. Shared Dependants are ok, but a job cannot have a dependancy
+// on any of its parents (this should never happen).
+func (q *fsJobQueue) DeleteJob(_ context.Context, id uuid.UUID) error {
+	// Start it off with an empty parent
+	return q.deleteJob(uuid.UUID{}, id)
+}
+
+// deleteJob will delete jobs as far down the list as possible
+// missing dependencies are ignored, it deletes as much as it can.
+// A missing parent (the first call) will be returned as an error
+func (q *fsJobQueue) deleteJob(parent, id uuid.UUID) error {
+	var j job
+	_, err := q.db.Read(id.String(), &j)
+	if err != nil {
+		return err
+	}
+
+	// Delete the parent uuid from the Dependents list
+	var deps []uuid.UUID
+	for _, d := range j.Dependents {
+		if d == parent {
+			continue
+		}
+		deps = append(deps, d)
+	}
+	j.Dependents = deps
+
+	// This job can only be deleted when the Dependents list is empty
+	// Otherwise it needs to be saved with the new Dependents list
+	if len(j.Dependents) > 0 {
+		return q.db.Write(id.String(), j)
+	}
+	// Recursively delete the dependencies of this job
+	for _, dj := range j.Dependencies {
+		_ = q.deleteJob(id, dj)
+	}
+
+	return q.db.Delete(id.String())
+}