Repository: osbuild/osbuild-composer
Pull request #2518 (2022-04-05 12:51:23)

cloudapi: prevent dangling goroutines after the server is terminated
By merging https://github.com/osbuild/osbuild-composer/pull/2417 I introduced flakey tests, oh no.

`T.TempDir` basically checks whether something writes into the directory *after* the test finishes, see https://github.com/golang/go/issues/43547.

This could happen in `cloudapi` because we have the manifest generation goroutines: When a test finishes, there might be still one running and it might write something into the directory created by `T.TempDir` and the test will explode.

The underlying issue is that we don't wait for all manifest generation goroutines to finish when shutting down the cloudapi server. This is potentially dangerous because the goroutine might be interrupted in a weird, non-defined state.

This PR refactors `cloudapi` a bit and adds a synchronization mechanism that waits for all manifest generation goroutines to finish.

This pull request includes:

- [ ] adequate testing for the new functionality or fixed issue
- [ ] adequate documentation informing people about the change such as
  - [ ] submit a PR for the [guides](https://github.com/osbuild/guides) repository if this PR changed any behavior described there: https://www.osbuild.org/guides/

<!--
Thanks for proposing a change to osbuild-composer!

Please don't remove the above check list. These are things that each pull
request must have before it is merged. It helps maintainers to not forget
anything.

If the reason for ticking any of the boxes is ambiguous, please add a short
note explaining why.

In addition, if this pull request fixes a downstream issue, please refer to
test/README.md and add these additional items:

- [ ] 1st commit of any `rhbz#` related PR contains bug reproducer; CI reports FAIL or
- [ ] PR contains automated tests for new functionality and
- [ ] QE has approved reproducer/new tests and
- [ ] Subsequent commits provide bug fixes without modifying the reproducer; CI reports PASS and
- [ ] QE approves this PR; RHBZ status is set to `MODIFIED + Verified=Tested`

Information regarding our GitLab pipeline (Schutzbot):

CI will not be ran automatically if WIP label is applied or the PR is in DRAFT state, instead only
a link will be provided to the pipeline which can then be triggered manually if desired. To run the
CI automatically either switch the PR to ready or apply WIP+test label.

Outside contributors need manual approval from one of the osbuild-composer maintainers.

Schutzbot will only be triggered if all Tests jobs in GitHub workflow succeed.
-->



Patch:
diff --git a/cmd/osbuild-composer/composer.go b/cmd/osbuild-composer/composer.go
index 2083a5d62..b25991317 100644
--- a/cmd/osbuild-composer/composer.go
+++ b/cmd/osbuild-composer/composer.go
@@ -330,7 +330,9 @@ func (c *Composer) Start() error {
 	logrus.Info("Shutting down.")
 
 	if c.apiListener != nil {
+		// First, close all listeners and then wait for all goroutines to finish.
 		err := composerAPI.Shutdown(context.Background())
+		c.api.Shutdown()
 		if err != nil {
 			panic(err)
 		}
diff --git a/cmd/osbuild-koji/main.go b/cmd/osbuild-koji/main.go
index fb3c0728a..42da02c5d 100644
--- a/cmd/osbuild-koji/main.go
+++ b/cmd/osbuild-koji/main.go
@@ -8,7 +8,6 @@ import (
 	"time"
 
 	"github.com/google/uuid"
-	"github.com/hashicorp/go-retryablehttp"
 	"github.com/osbuild/osbuild-composer/internal/rpmmd"
 	"github.com/osbuild/osbuild-composer/internal/upload/koji"
 	"github.com/sirupsen/logrus"
@@ -42,7 +41,7 @@ func main() {
 	}
 	defer file.Close()
 
-	transport := &retryablehttp.RoundTripper{}
+	transport := koji.CreateRetryableTransport()
 	k, err := koji.NewFromPlain(server, "osbuild", "osbuildpass", transport)
 	if err != nil {
 		println(err.Error())
diff --git a/internal/cloudapi/server.go b/internal/cloudapi/server.go
index bc1f4ac77..4c515ea1b 100644
--- a/internal/cloudapi/server.go
+++ b/internal/cloudapi/server.go
@@ -23,3 +23,7 @@ func NewServer(workers *worker.Server, distros *distroregistry.Registry, config
 func (server *Server) V2(path string) http.Handler {
 	return server.v2.Handler(path)
 }
+
+func (server *Server) Shutdown() {
+	server.v2.Shutdown()
+}
diff --git a/internal/cloudapi/v2/v2.go b/internal/cloudapi/v2/handler.go
similarity index 77%
rename from internal/cloudapi/v2/v2.go
rename to internal/cloudapi/v2/handler.go
index fa5913a21..cde88bfa6 100644
--- a/internal/cloudapi/v2/v2.go
+++ b/internal/cloudapi/v2/handler.go
@@ -2,7 +2,6 @@
 package v2
 
 import (
-	"context"
 	"crypto/rand"
 	"encoding/json"
 	"fmt"
@@ -11,72 +10,27 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
-	"time"
 
 	"github.com/google/uuid"
 	"github.com/labstack/echo/v4"
-	"github.com/labstack/echo/v4/middleware"
-	"github.com/sirupsen/logrus"
 
 	"github.com/osbuild/osbuild-composer/internal/auth"
 	"github.com/osbuild/osbuild-composer/internal/blueprint"
-	"github.com/osbuild/osbuild-composer/internal/common"
 	"github.com/osbuild/osbuild-composer/internal/distro"
-	"github.com/osbuild/osbuild-composer/internal/distroregistry"
-	"github.com/osbuild/osbuild-composer/internal/jobqueue"
 	osbuild "github.com/osbuild/osbuild-composer/internal/osbuild2"
 	"github.com/osbuild/osbuild-composer/internal/ostree"
-	"github.com/osbuild/osbuild-composer/internal/prometheus"
 	"github.com/osbuild/osbuild-composer/internal/rpmmd"
 	"github.com/osbuild/osbuild-composer/internal/target"
 	"github.com/osbuild/osbuild-composer/internal/worker"
 	"github.com/osbuild/osbuild-composer/internal/worker/clienterrors"
 )
 
-// Server represents the state of the cloud Server
-type Server struct {
-	workers *worker.Server
-	distros *distroregistry.Registry
-	config  ServerConfig
-}
-
-type ServerConfig struct {
-	AWSBucket            string
-	TenantProviderFields []string
-	JWTEnabled           bool
-}
-
 type apiHandlers struct {
 	server *Server
 }
 
 type binder struct{}
 
-func NewServer(workers *worker.Server, distros *distroregistry.Registry, config ServerConfig) *Server {
-	server := &Server{
-		workers: workers,
-		distros: distros,
-		config:  config,
-	}
-	return server
-}
-
-func (server *Server) Handler(path string) http.Handler {
-	e := echo.New()
-	e.Binder = binder{}
-	e.HTTPErrorHandler = server.HTTPErrorHandler
-	e.Pre(common.OperationIDMiddleware)
-	e.Use(middleware.Recover())
-	e.Logger = common.Logger()
-
-	handler := apiHandlers{
-		server: server,
-	}
-	RegisterHandlers(e.Group(path, prometheus.MetricsMiddleware), &handler)
-
-	return e
-}
-
 func (b binder) Bind(i interface{}, ctx echo.Context) error {
 	contentType := ctx.Request().Header["Content-Type"]
 	if len(contentType) != 1 || contentType[0] != "application/json" {
@@ -484,12 +438,12 @@ func (h *apiHandlers) PostCompose(ctx echo.Context) error {
 
 	var id uuid.UUID
 	if request.Koji != nil {
-		id, err = enqueueKojiCompose(h.server.workers, uint64(request.Koji.TaskId), request.Koji.Server, request.Koji.Name, request.Koji.Version, request.Koji.Release, distribution, bp, manifestSeed, irs, channel)
+		id, err = h.server.enqueueKojiCompose(uint64(request.Koji.TaskId), request.Koji.Server, request.Koji.Name, request.Koji.Version, request.Koji.Release, distribution, bp, manifestSeed, irs, channel)
 		if err != nil {
 			return err
 		}
 	} else {
-		id, err = enqueueCompose(h.server.workers, distribution, bp, manifestSeed, irs, channel)
+		id, err = h.server.enqueueCompose(distribution, bp, manifestSeed, irs, channel)
 		if err != nil {
 			return err
 		}
@@ -507,215 +461,6 @@ func (h *apiHandlers) PostCompose(ctx echo.Context) error {
 	})
 }
 
-func enqueueCompose(workers *worker.Server, distribution distro.Distro, bp blueprint.Blueprint, manifestSeed int64, irs []imageRequest, channel string) (uuid.UUID, error) {
-	var id uuid.UUID
-	if len(irs) != 1 {
-		return id, HTTPError(ErrorInvalidNumberOfImageBuilds)
-	}
-	ir := irs[0]
-
-	depsolveJobID, err := workers.EnqueueDepsolve(&worker.DepsolveJob{
-		PackageSets:      ir.imageType.PackageSets(bp),
-		Repos:            ir.repositories,
-		ModulePlatformID: distribution.ModulePlatformID(),
-		Arch:             ir.arch.Name(),
-		Releasever:       distribution.Releasever(),
-		PackageSetsRepos: ir.packageSetsRepositories,
-	}, channel)
-	if err != nil {
-		return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
-	}
-
-	manifestJobID, err := workers.EnqueueManifestJobByID(&worker.ManifestJobByID{}, depsolveJobID, channel)
-	if err != nil {
-		return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
-	}
-
-	id, err = workers.EnqueueOSBuildAsDependency(ir.arch.Name(), &worker.OSBuildJob{
-		Targets:         []*target.Target{ir.target},
-		Exports:         ir.imageType.Exports(),
-		StreamOptimized: ir.imageType.Name() == "vmdk", // https://github.com/osbuild/osbuild/issues/528,
-		PipelineNames: &worker.PipelineNames{
-			Build:   ir.imageType.BuildPipelines(),
-			Payload: ir.imageType.PayloadPipelines(),
-		},
-	}, manifestJobID, channel)
-	if err != nil {
-		return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
-	}
-
-	ctx, cancel := context.WithTimeout(context.Background(), time.Minute*5)
-	go generateManifest(ctx, cancel, workers, depsolveJobID, manifestJobID, ir.imageType, ir.repositories, ir.imageOptions, manifestSeed, bp.Customizations)
-
-	return id, nil
-}
-
-func enqueueKojiCompose(workers *worker.Server, taskID uint64, server, name, version, release string, distribution distro.Distro, bp blueprint.Blueprint, manifestSeed int64, irs []imageRequest, channel string) (uuid.UUID, error) {
-	var id uuid.UUID
-	kojiDirectory := "osbuild-composer-koji-" + uuid.New().String()
-
-	initID, err := workers.EnqueueKojiInit(&worker.KojiInitJob{
-		Server:  server,
-		Name:    name,
-		Version: version,
-		Release: release,
-	}, channel)
-	if err != nil {
-		return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
-	}
-
-	var kojiFilenames []string
-	var buildIDs []uuid.UUID
-	for _, ir := range irs {
-		depsolveJobID, err := workers.EnqueueDepsolve(&worker.DepsolveJob{
-			PackageSets:      ir.imageType.PackageSets(bp),
-			Repos:            ir.repositories,
-			ModulePlatformID: distribution.ModulePlatformID(),
-			Arch:             ir.arch.Name(),
-			Releasever:       distribution.Releasever(),
-			PackageSetsRepos: ir.packageSetsRepositories,
-		}, channel)
-		if err != nil {
-			return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
-		}
-
-		manifestJobID, err := workers.EnqueueManifestJobByID(&worker.ManifestJobByID{}, depsolveJobID, channel)
-		if err != nil {
-			return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
-		}
-		kojiFilename := fmt.Sprintf(
-			"%s-%s-%s.%s%s",
-			name,
-			version,
-			release,
-			ir.arch.Name(),
-			splitExtension(ir.imageType.Filename()),
-		)
-		buildID, err := workers.EnqueueOSBuildKojiAsDependency(ir.arch.Name(), &worker.OSBuildKojiJob{
-			ImageName: ir.imageType.Filename(),
-			Exports:   ir.imageType.Exports(),
-			PipelineNames: &worker.PipelineNames{
-				Build:   ir.imageType.BuildPipelines(),
-				Payload: ir.imageType.PayloadPipelines(),
-			},
-			KojiServer:    server,
-			KojiDirectory: kojiDirectory,
-			KojiFilename:  kojiFilename,
-		}, manifestJobID, initID, channel)
-		if err != nil {
-			return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
-		}
-		kojiFilenames = append(kojiFilenames, kojiFilename)
-		buildIDs = append(buildIDs, buildID)
-
-		ctx, cancel := context.WithTimeout(context.Background(), time.Minute*5)
-		go generateManifest(ctx, cancel, workers, depsolveJobID, manifestJobID, ir.imageType, ir.repositories, ir.imageOptions, manifestSeed, bp.Customizations)
-	}
-	id, err = workers.EnqueueKojiFinalize(&worker.KojiFinalizeJob{
-		Server:        server,
-		Name:          name,
-		Version:       version,
-		Release:       release,
-		KojiFilenames: kojiFilenames,
-		KojiDirectory: kojiDirectory,
-		TaskID:        taskID,
-		StartTime:     uint64(time.Now().Unix()),
-	}, initID, buildIDs, channel)
-	if err != nil {
-		return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
-	}
-
-	return id, nil
-}
-
-func generateManifest(ctx context.Context, cancel context.CancelFunc, workers *worker.Server, depsolveJobID uuid.UUID, manifestJobID uuid.UUID, imageType distro.ImageType, repos []rpmmd.RepoConfig, options distro.ImageOptions, seed int64, b *blueprint.Customizations) {
-	defer cancel()
-
-	// wait until job is in a pending state
-	var token uuid.UUID
-	var dynArgs []json.RawMessage
-	var err error
-	logWithId := logrus.WithField("jobId", manifestJobID)
-	for {
-		_, token, _, _, dynArgs, err = workers.RequestJobById(ctx, "", manifestJobID)
-		if err == jobqueue.ErrNotPending {
-			logWithId.Debug("Manifest job not pending, waiting for depsolve job to finish")
-			time.Sleep(time.Millisecond * 50)
-			select {
-			case <-ctx.Done():
-				logWithId.Warning("Manifest job dependencies took longer than 5 minutes to finish, returning to avoid dangling routines")
-				break
-			default:
-				continue
-			}
-		}
-		if err != nil {
-			logWithId.Errorf("Error requesting manifest job: %v", err)
-			return
-		}
-		break
-	}
-
-	var jobResult *worker.ManifestJobByIDResult = &worker.ManifestJobByIDResult{
-		Manifest: nil,
-	}
-
-	defer func() {
-		if jobResult.JobError != nil {
-			logWithId.Errorf("Error in manifest job %v: %v", jobResult.JobError.Reason, err)
-		}
-
-		result, err := json.Marshal(jobResult)
-		if err != nil {
-			logWithId.Errorf("Error marshalling manifest job results: %v", err)
-		}
-
-		err = workers.FinishJob(token, result)
-		if err != nil {
-			logWithId.Errorf("Error finishing manifest job: %v", err)
-		}
-	}()
-
-	if len(dynArgs) == 0 {
-		reason := "No dynamic arguments"
-		jobResult.JobError = clienterrors.WorkerClientError(clienterrors.ErrorNoDynamicArgs, reason)
-		return
-	}
-
-	var depsolveResults worker.DepsolveJobResult
-	err = json.Unmarshal(dynArgs[0], &depsolveResults)
-	if err != nil {
-		reason := "Error parsing dynamic arguments"
-		jobResult.JobError = clienterrors.WorkerClientError(clienterrors.ErrorParsingDynamicArgs, reason)
-		return
-	}
-
-	_, _, err = workers.DepsolveJobStatus(depsolveJobID, &depsolveResults)
-	if err != nil {
-		reason := "Error reading depsolve status"
-		jobResult.JobError = clienterrors.WorkerClientError(clienterrors.ErrorReadingJobStatus, reason)
-		return
-	}
-
-	if jobErr := depsolveResults.JobError; jobErr != nil {
-		if jobErr.ID == clienterrors.ErrorDNFDepsolveError || jobErr.ID == clienterrors.ErrorDNFMarkingErrors {
-			jobResult.JobError = clienterrors.WorkerClientError(clienterrors.ErrorDepsolveDependency, "Error in depsolve job dependency input, bad package set requested")
-			return
-		}
-		jobResult.JobError = clienterrors.WorkerClientError(clienterrors.ErrorDepsolveDependency, "Error in depsolve job dependency")
-		return
-	}
-
-	manifest, err := imageType.Manifest(b, options, repos, depsolveResults.PackageSpecs, seed)
-	if err != nil {
-		reason := "Error generating manifest"
-		jobResult.JobError = clienterrors.WorkerClientError(clienterrors.ErrorManifestGeneration, reason)
-		return
-	}
-
-	jobResult.Manifest = manifest
-}
-
 func imageTypeFromApiImageType(it ImageTypes, arch distro.Arch) string {
 	switch it {
 	case ImageTypesAws:
diff --git a/internal/cloudapi/v2/server.go b/internal/cloudapi/v2/server.go
new file mode 100644
index 000000000..8c0833f44
--- /dev/null
+++ b/internal/cloudapi/v2/server.go
@@ -0,0 +1,294 @@
+package v2
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"sync"
+	"time"
+
+	"github.com/google/uuid"
+	"github.com/labstack/echo/v4"
+	"github.com/labstack/echo/v4/middleware"
+	"github.com/sirupsen/logrus"
+
+	"github.com/osbuild/osbuild-composer/internal/blueprint"
+	"github.com/osbuild/osbuild-composer/internal/common"
+	"github.com/osbuild/osbuild-composer/internal/distro"
+	"github.com/osbuild/osbuild-composer/internal/distroregistry"
+	"github.com/osbuild/osbuild-composer/internal/jobqueue"
+	"github.com/osbuild/osbuild-composer/internal/prometheus"
+	"github.com/osbuild/osbuild-composer/internal/rpmmd"
+	"github.com/osbuild/osbuild-composer/internal/target"
+	"github.com/osbuild/osbuild-composer/internal/worker"
+	"github.com/osbuild/osbuild-composer/internal/worker/clienterrors"
+)
+
+// Server represents the state of the cloud Server
+type Server struct {
+	workers *worker.Server
+	distros *distroregistry.Registry
+	config  ServerConfig
+
+	goroutinesCtx       context.Context
+	goroutinesCtxCancel context.CancelFunc
+	goroutinesGroup     sync.WaitGroup
+}
+
+type ServerConfig struct {
+	AWSBucket            string
+	TenantProviderFields []string
+	JWTEnabled           bool
+}
+
+func NewServer(workers *worker.Server, distros *distroregistry.Registry, config ServerConfig) *Server {
+	ctx, cancel := context.WithCancel(context.Background())
+	server := &Server{
+		workers: workers,
+		distros: distros,
+		config:  config,
+
+		goroutinesCtx:       ctx,
+		goroutinesCtxCancel: cancel,
+	}
+	return server
+}
+
+func (s *Server) Handler(path string) http.Handler {
+	e := echo.New()
+	e.Binder = binder{}
+	e.HTTPErrorHandler = s.HTTPErrorHandler
+	e.Pre(common.OperationIDMiddleware)
+	e.Use(middleware.Recover())
+	e.Logger = common.Logger()
+
+	handler := apiHandlers{
+		server: s,
+	}
+	RegisterHandlers(e.Group(path, prometheus.MetricsMiddleware), &handler)
+
+	return e
+}
+
+func (s *Server) Shutdown() {
+	s.goroutinesCtxCancel()
+	s.goroutinesGroup.Wait()
+}
+
+func (s *Server) enqueueCompose(distribution distro.Distro, bp blueprint.Blueprint, manifestSeed int64, irs []imageRequest, channel string) (uuid.UUID, error) {
+	var id uuid.UUID
+	if len(irs) != 1 {
+		return id, HTTPError(ErrorInvalidNumberOfImageBuilds)
+	}
+	ir := irs[0]
+
+	depsolveJobID, err := s.workers.EnqueueDepsolve(&worker.DepsolveJob{
+		PackageSets:      ir.imageType.PackageSets(bp),
+		Repos:            ir.repositories,
+		ModulePlatformID: distribution.ModulePlatformID(),
+		Arch:             ir.arch.Name(),
+		Releasever:       distribution.Releasever(),
+		PackageSetsRepos: ir.packageSetsRepositories,
+	}, channel)
+	if err != nil {
+		return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
+	}
+
+	manifestJobID, err := s.workers.EnqueueManifestJobByID(&worker.ManifestJobByID{}, depsolveJobID, channel)
+	if err != nil {
+		return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
+	}
+
+	id, err = s.workers.EnqueueOSBuildAsDependency(ir.arch.Name(), &worker.OSBuildJob{
+		Targets:         []*target.Target{ir.target},
+		Exports:         ir.imageType.Exports(),
+		StreamOptimized: ir.imageType.Name() == "vmdk", // https://github.com/osbuild/osbuild/issues/528,
+		PipelineNames: &worker.PipelineNames{
+			Build:   ir.imageType.BuildPipelines(),
+			Payload: ir.imageType.PayloadPipelines(),
+		},
+	}, manifestJobID, channel)
+	if err != nil {
+		return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
+	}
+
+	s.goroutinesGroup.Add(1)
+	go func() {
+		generateManifest(s.goroutinesCtx, s.workers, depsolveJobID, manifestJobID, ir.imageType, ir.repositories, ir.imageOptions, manifestSeed, bp.Customizations)
+		defer s.goroutinesGroup.Done()
+	}()
+
+	return id, nil
+}
+
+func (s *Server) enqueueKojiCompose(taskID uint64, server, name, version, release string, distribution distro.Distro, bp blueprint.Blueprint, manifestSeed int64, irs []imageRequest, channel string) (uuid.UUID, error) {
+	var id uuid.UUID
+	kojiDirectory := "osbuild-composer-koji-" + uuid.New().String()
+
+	initID, err := s.workers.EnqueueKojiInit(&worker.KojiInitJob{
+		Server:  server,
+		Name:    name,
+		Version: version,
+		Release: release,
+	}, channel)
+	if err != nil {
+		return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
+	}
+
+	var kojiFilenames []string
+	var buildIDs []uuid.UUID
+	for _, ir := range irs {
+		depsolveJobID, err := s.workers.EnqueueDepsolve(&worker.DepsolveJob{
+			PackageSets:      ir.imageType.PackageSets(bp),
+			Repos:            ir.repositories,
+			ModulePlatformID: distribution.ModulePlatformID(),
+			Arch:             ir.arch.Name(),
+			Releasever:       distribution.Releasever(),
+			PackageSetsRepos: ir.packageSetsRepositories,
+		}, channel)
+		if err != nil {
+			return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
+		}
+
+		manifestJobID, err := s.workers.EnqueueManifestJobByID(&worker.ManifestJobByID{}, depsolveJobID, channel)
+		if err != nil {
+			return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
+		}
+		kojiFilename := fmt.Sprintf(
+			"%s-%s-%s.%s%s",
+			name,
+			version,
+			release,
+			ir.arch.Name(),
+			splitExtension(ir.imageType.Filename()),
+		)
+		buildID, err := s.workers.EnqueueOSBuildKojiAsDependency(ir.arch.Name(), &worker.OSBuildKojiJob{
+			ImageName: ir.imageType.Filename(),
+			Exports:   ir.imageType.Exports(),
+			PipelineNames: &worker.PipelineNames{
+				Build:   ir.imageType.BuildPipelines(),
+				Payload: ir.imageType.PayloadPipelines(),
+			},
+			KojiServer:    server,
+			KojiDirectory: kojiDirectory,
+			KojiFilename:  kojiFilename,
+		}, manifestJobID, initID, channel)
+		if err != nil {
+			return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
+		}
+		kojiFilenames = append(kojiFilenames, kojiFilename)
+		buildIDs = append(buildIDs, buildID)
+
+		// copy the image request while passing it into the goroutine to prevent data races
+		s.goroutinesGroup.Add(1)
+		go func(ir imageRequest) {
+			generateManifest(s.goroutinesCtx, s.workers, depsolveJobID, manifestJobID, ir.imageType, ir.repositories, ir.imageOptions, manifestSeed, bp.Customizations)
+			defer s.goroutinesGroup.Done()
+		}(ir)
+	}
+	id, err = s.workers.EnqueueKojiFinalize(&worker.KojiFinalizeJob{
+		Server:        server,
+		Name:          name,
+		Version:       version,
+		Release:       release,
+		KojiFilenames: kojiFilenames,
+		KojiDirectory: kojiDirectory,
+		TaskID:        taskID,
+		StartTime:     uint64(time.Now().Unix()),
+	}, initID, buildIDs, channel)
+	if err != nil {
+		return id, HTTPErrorWithInternal(ErrorEnqueueingJob, err)
+	}
+
+	return id, nil
+}
+
+func generateManifest(ctx context.Context, workers *worker.Server, depsolveJobID uuid.UUID, manifestJobID uuid.UUID, imageType distro.ImageType, repos []rpmmd.RepoConfig, options distro.ImageOptions, seed int64, b *blueprint.Customizations) {
+	ctx, cancel := context.WithTimeout(ctx, time.Minute*5)
+	defer cancel()
+
+	// wait until job is in a pending state
+	var token uuid.UUID
+	var dynArgs []json.RawMessage
+	var err error
+	logWithId := logrus.WithField("jobId", manifestJobID)
+	for {
+		_, token, _, _, dynArgs, err = workers.RequestJobById(ctx, "", manifestJobID)
+		if err == jobqueue.ErrNotPending {
+			logWithId.Debug("Manifest job not pending, waiting for depsolve job to finish")
+			time.Sleep(time.Millisecond * 50)
+			select {
+			case <-ctx.Done():
+				logWithId.Warning("Manifest job dependencies took longer than 5 minutes to finish, or the server is shutting down, returning to avoid dangling routines")
+				break
+			default:
+				continue
+			}
+		}
+		if err != nil {
+			logWithId.Errorf("Error requesting manifest job: %v", err)
+			return
+		}
+		break
+	}
+
+	jobResult := &worker.ManifestJobByIDResult{
+		Manifest: nil,
+	}
+
+	defer func() {
+		if jobResult.JobError != nil {
+			logWithId.Errorf("Error in manifest job %v: %v", jobResult.JobError.Reason, err)
+		}
+
+		result, err := json.Marshal(jobResult)
+		if err != nil {
+			logWithId.Errorf("Error marshalling manifest job results: %v", err)
+		}
+
+		err = workers.FinishJob(token, result)
+		if err != nil {
+			logWithId.Errorf("Error finishing manifest job: %v", err)
+		}
+	}()
+
+	if len(dynArgs) == 0 {
+		reason := "No dynamic arguments"
+		jobResult.JobError = clienterrors.WorkerClientError(clienterrors.ErrorNoDynamicArgs, reason)
+		return
+	}
+
+	var depsolveResults worker.DepsolveJobResult
+	err = json.Unmarshal(dynArgs[0], &depsolveResults)
+	if err != nil {
+		reason := "Error parsing dynamic arguments"
+		jobResult.JobError = clienterrors.WorkerClientError(clienterrors.ErrorParsingDynamicArgs, reason)
+		return
+	}
+
+	_, _, err = workers.DepsolveJobStatus(depsolveJobID, &depsolveResults)
+	if err != nil {
+		reason := "Error reading depsolve status"
+		jobResult.JobError = clienterrors.WorkerClientError(clienterrors.ErrorReadingJobStatus, reason)
+		return
+	}
+
+	if jobErr := depsolveResults.JobError; jobErr != nil {
+		if jobErr.ID == clienterrors.ErrorDNFDepsolveError || jobErr.ID == clienterrors.ErrorDNFMarkingErrors {
+			jobResult.JobError = clienterrors.WorkerClientError(clienterrors.ErrorDepsolveDependency, "Error in depsolve job dependency input, bad package set requested")
+			return
+		}
+		jobResult.JobError = clienterrors.WorkerClientError(clienterrors.ErrorDepsolveDependency, "Error in depsolve job dependency")
+		return
+	}
+
+	manifest, err := imageType.Manifest(b, options, repos, depsolveResults.PackageSpecs, seed)
+	if err != nil {
+		reason := "Error generating manifest"
+		jobResult.JobError = clienterrors.WorkerClientError(clienterrors.ErrorManifestGeneration, reason)
+		return
+	}
+
+	jobResult.Manifest = manifest
+}
diff --git a/internal/upload/koji/koji.go b/internal/upload/koji/koji.go
index 5a43f8ea2..084544e7a 100644
--- a/internal/upload/koji/koji.go
+++ b/internal/upload/koji/koji.go
@@ -162,7 +162,7 @@ func NewFromPlain(server, user, password string, transport http.RoundTripper) (*
 	// The API doesn't require sessionID, sessionKey and callnum yet,
 	// so there's no need to use the custom Koji RoundTripper,
 	// let's just use the one that the called passed in.
-	rhTransport := &rh.RoundTripper{Client: rh.NewClient()}
+	rhTransport := CreateRetryableTransport()
 	loginClient, err := xmlrpc.NewClient(server, rhTransport)
 	if err != nil {
 		return nil, err
@@ -295,14 +295,14 @@ func (k *Koji) CGImport(build ImageBuild, buildRoots []BuildRoot, images []Image
 }
 
 // uploadChunk uploads a byte slice to a given filepath/filname at a given offset
-func (k *Koji) uploadChunk(chunk []byte, filepath, filename string, offset uint64) (uint, error) {
+func (k *Koji) uploadChunk(chunk []byte, filepath, filename string, offset uint64) error {
 	// We have to open-code a bastardized version of XML-RPC: We send an octet-stream, as
 	// if it was an RPC call, and get a regular XML-RPC reply back. In addition to the
 	// standard URL parameters, we also have to pass any other parameters as part of the
 	// URL, as the body can only contain the payload.
 	u, err := url.Parse(k.server)
 	if err != nil {
-		return 0, err
+		return err
 	}
 	q := u.Query()
 	q.Add("filepath", filepath)
@@ -311,49 +311,23 @@ func (k *Koji) uploadChunk(chunk []byte, filepath, filename string, offset uint6
 	q.Add("fileverify", "adler32")
 	u.RawQuery = q.Encode()
 
-	retries := uint(0)
-
-	countingCheckRetry := func(ctx context.Context, resp *http.Response, err error) (bool, error) {
-		shouldRetry, retErr := rh.DefaultRetryPolicy(ctx, resp, err)
-
-		// DefaultRetryPolicy denies retrying for any certificate related error.
-		// Override it in case the error is a timeout.
-		if !shouldRetry && err != nil {
-			if v, ok := err.(*url.Error); ok {
-				if _, ok := v.Err.(x509.UnknownAuthorityError); ok {
-					// retry if it's a timeout
-					shouldRetry = strings.Contains(strings.ToLower(v.Error()), "timeout")
-					retErr = v
-				}
-			}
-		}
-
-		if shouldRetry {
-			retries++
-		}
-
-		return shouldRetry, retErr
-	}
-
-	client := rh.NewClient()
+	client := createCustomRetryableClient()
 
 	client.HTTPClient = &http.Client{
 		Transport: k.transport,
 	}
-	client.CheckRetry = countingCheckRetry
-	client.Logger = rh.LeveledLogger(&LeveledLogrus{logrus.StandardLogger()})
 
 	respData, err := client.Post(u.String(), "application/octet-stream", bytes.NewBuffer(chunk))
 
 	if err != nil {
-		return retries, err
+		return err
 	}
 
 	defer respData.Body.Close()
 
 	body, err := ioutil.ReadAll(respData.Body)
 	if err != nil {
-		return retries, err
+		return err
 	}
 
 	var reply struct {
@@ -364,31 +338,30 @@ func (k *Koji) uploadChunk(chunk []byte, filepath, filename string, offset uint6
 	resp := xmlrpc.Response(body)
 
 	if resp.Err() != nil {
-		return retries, fmt.Errorf("xmlrpc server returned an error: %v", resp.Err())
+		return fmt.Errorf("xmlrpc server returned an error: %v", resp.Err())
 	}
 
 	err = resp.Unmarshal(&reply)
 	if err != nil {
-		return retries, fmt.Errorf("cannot unmarshal the xmlrpc response: %v", err)
+		return fmt.Errorf("cannot unmarshal the xmlrpc response: %v", err)
 	}
 
 	if reply.Size != len(chunk) {
-		return retries, fmt.Errorf("Sent a chunk of %d bytes, but server got %d bytes", len(chunk), reply.Size)
+		return fmt.Errorf("Sent a chunk of %d bytes, but server got %d bytes", len(chunk), reply.Size)
 	}
 
 	digest := fmt.Sprintf("%08x", adler32.Checksum(chunk))
 	if reply.HexDigest != digest {
-		return retries, fmt.Errorf("Sent a chunk with Adler32 digest %s, but server computed digest %s", digest, reply.HexDigest)
+		return fmt.Errorf("Sent a chunk with Adler32 digest %s, but server computed digest %s", digest, reply.HexDigest)
 	}
 
-	return retries, nil
+	return nil
 }
 
 // Upload uploads file to the temporary filepath on the kojiserver under the name filename
 // The md5sum and size of the file is returned on success.
 func (k *Koji) Upload(file io.Reader, filepath, filename string) (string, uint64, error) {
 	chunk := make([]byte, 1024*1024) // upload a megabyte at a time
-	retries := uint(0)
 	offset := uint64(0)
 	// Koji uses MD5 hashes
 	/* #nosec G401 */
@@ -401,12 +374,11 @@ func (k *Koji) Upload(file io.Reader, filepath, filename string) (string, uint64
 			}
 			return "", 0, err
 		}
-		r, err := k.uploadChunk(chunk[:n], filepath, filename, offset)
-		retries += r
+		err = k.uploadChunk(chunk[:n], filepath, filename, offset)
 		if err != nil {
-			logrus.Infof("Koji upload failed after %d retries", retries)
 			return "", 0, err
 		}
+
 		offset += uint64(n)
 
 		m, err := hash.Write(chunk[:n])
@@ -417,7 +389,6 @@ func (k *Koji) Upload(file io.Reader, filepath, filename string) (string, uint64
 			return "", 0, fmt.Errorf("sent %d bytes, but hashed %d", n, m)
 		}
 	}
-	logrus.Infof("Koji upload successful after %d retries", retries)
 	return fmt.Sprintf("%x", hash.Sum(nil)), offset, nil
 }
 
@@ -472,7 +443,7 @@ func GSSAPICredentialsFromEnv() (*GSSAPICredentials, error) {
 func CreateKojiTransport(relaxTimeout uint) http.RoundTripper {
 	// Koji for some reason needs TLS renegotiation enabled.
 	// Clone the default http rt and enable renegotiation.
-	rt := &rh.RoundTripper{Client: rh.NewClient()}
+	rt := CreateRetryableTransport()
 
 	transport := rt.Client.HTTPClient.Transport.(*http.Transport)
 
@@ -492,3 +463,33 @@ func CreateKojiTransport(relaxTimeout uint) http.RoundTripper {
 
 	return rt
 }
+
+func customCheckRetry(ctx context.Context, resp *http.Response, err error) (bool, error) {
+	shouldRetry, retErr := rh.DefaultRetryPolicy(ctx, resp, err)
+
+	// DefaultRetryPolicy denies retrying for any certificate related error.
+	// Override it in case the error is a timeout.
+	if !shouldRetry && err != nil {
+		if v, ok := err.(*url.Error); ok {
+			if _, ok := v.Err.(x509.UnknownAuthorityError); ok {
+				// retry if it's a timeout
+				return strings.Contains(strings.ToLower(v.Error()), "timeout"), v
+			}
+		}
+	}
+
+	return shouldRetry, retErr
+}
+
+func createCustomRetryableClient() *rh.Client {
+	client := rh.NewClient()
+	client.Logger = rh.LeveledLogger(&LeveledLogrus{logrus.StandardLogger()})
+	client.CheckRetry = customCheckRetry
+	return client
+}
+
+func CreateRetryableTransport() *rh.RoundTripper {
+	rt := rh.RoundTripper{}
+	rt.Client = createCustomRetryableClient()
+	return &rt
+}
diff --git a/internal/upload/koji/rh-logrus-adapter.go b/internal/upload/koji/rh-logrus-adapter.go
index 444512865..5a5c482c1 100644
--- a/internal/upload/koji/rh-logrus-adapter.go
+++ b/internal/upload/koji/rh-logrus-adapter.go
@@ -1,6 +1,8 @@
 package koji
 
 import (
+	"strings"
+
 	"github.com/sirupsen/logrus"
 )
 
@@ -8,7 +10,9 @@ type LeveledLogrus struct {
 	*logrus.Logger
 }
 
-func (l *LeveledLogrus) fields(keysAndValues ...interface{}) map[string]interface{} {
+const monitoringKeyword = "retrying"
+
+func fields(keysAndValues ...interface{}) map[string]interface{} {
 	fields := make(map[string]interface{})
 
 	for i := 0; i < len(keysAndValues)-1; i += 2 {
@@ -19,16 +23,20 @@ func (l *LeveledLogrus) fields(keysAndValues ...interface{}) map[string]interfac
 }
 
 func (l *LeveledLogrus) Error(msg string, keysAndValues ...interface{}) {
-	l.WithFields(l.fields(keysAndValues...)).Error(msg)
+	l.WithFields(fields(keysAndValues...)).Error(msg)
 }
 
 func (l *LeveledLogrus) Info(msg string, keysAndValues ...interface{}) {
-	l.WithFields(l.fields(keysAndValues...)).Info(msg)
+	l.WithFields(fields(keysAndValues...)).Info(msg)
 }
 func (l *LeveledLogrus) Debug(msg string, keysAndValues ...interface{}) {
-	l.WithFields(l.fields(keysAndValues...)).Debug(msg)
+	if strings.Contains(msg, monitoringKeyword) {
+		l.WithFields(fields(keysAndValues...)).Info(msg)
+	} else {
+		l.WithFields(fields(keysAndValues...)).Debug(msg)
+	}
 }
 
 func (l *LeveledLogrus) Warn(msg string, keysAndValues ...interface{}) {
-	l.WithFields(l.fields(keysAndValues...)).Warn(msg)
+	l.WithFields(fields(keysAndValues...)).Warn(msg)
 }


Test patch:
diff --git a/internal/cloudapi/v2/v2_test.go b/internal/cloudapi/v2/v2_test.go
index 91b4ab434..f9d90c567 100644
--- a/internal/cloudapi/v2/v2_test.go
+++ b/internal/cloudapi/v2/v2_test.go
@@ -39,6 +39,7 @@ func newV2Server(t *testing.T, dir string, depsolveChannels []string, enableJWT
 	}
 	v2Server := v2.NewServer(workerServer, distros, config)
 	require.NotNil(t, v2Server)
+	t.Cleanup(v2Server.Shutdown)
 
 	// start a routine which just completes depsolve jobs
 	depsolveContext, cancel := context.WithCancel(context.Background())