✨ Job Management (#12)

<!--
Copyright (C) 2020-2022 Arm Limited or its affiliates and Contributors.
All rights reserved.
SPDX-License-Identifier: Proprietary
-->
### Description

Added utilities for managing jobs



### Test Coverage

<!--
Please put an `x` in the correct box e.g. `[x]` to indicate the testing
coverage of this change.
-->

- [x]  This change is covered by existing or additional automated tests.
- [ ] Manual testing has been performed (and evidence provided) as
automated testing was not feasible.
- [ ] Additional tests are not required for this change (e.g.
documentation update).

Co-authored-by: acabarbaye <[email protected]>

Author: acabarbaye

changes/20230103145501.feature

@@ -0,0 +1 @@
+:sparkles: Added utilities for dealing with jobs

utils/go.mod

@@ -11,6 +11,7 @@ require (
 	github.com/stretchr/testify v1.8.1
 	go.uber.org/atomic v1.10.0
 	go.uber.org/goleak v1.2.0
+	golang.org/x/sync v0.1.0
 )
 
 require (
@@ -55,7 +56,6 @@ require (
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
 	golang.org/x/net v0.4.0 // indirect
 	golang.org/x/oauth2 v0.0.0-20221014153046-6fdb5e3db783 // indirect
-	golang.org/x/sync v0.1.0 // indirect
 	golang.org/x/sys v0.3.0 // indirect
 	golang.org/x/text v0.5.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect

utils/job/interfaces.go

@@ -0,0 +1,41 @@
+/*
+ * Copyright (C) 2020-2023 Arm Limited or its affiliates and Contributors. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+// Package job defines utilities for managing jobs.
+package job
+
+import (
+	"context"
+
+	"github.com/ARM-software/embedded-development-services-client-utils/utils/resource"
+)
+
+// Mocks are generated using `go generate ./...`
+// Add interfaces to the following command for a mock to be generated
+//go:generate mockgen -destination=../mocks/mock_$GOPACKAGE.go -package=mocks github.com/ARM-software/embedded-development-services-client-utils/utils/$GOPACKAGE IAsynchronousJob,IJobManager
+
+// IAsynchronousJob defines a typical asynchronous job.
+type IAsynchronousJob interface {
+	resource.IResource
+	// GetDone returns whether a job has terminated.
+	GetDone() bool
+	// GetError returns whether a system error occurred.
+	GetError() bool
+	// GetFailure returns whether the job has failed.
+	GetFailure() bool
+	// GetSuccess returns whether the job has been successful.
+	GetSuccess() bool
+	// GetStatus returns the state the job is in. This is for information only and should not be relied upon as likely to change. Use flags for implementing a state machine.
+	GetStatus() string
+}
+
+// IJobManager defines a manager of asynchronous jobs
+type IJobManager interface {
+	// HasJobCompleted calls the services to determine whether the job has completed.
+	HasJobCompleted(ctx context.Context, job IAsynchronousJob) (completed bool, err error)
+
+	// WaitForJobCompletion waits for a job to complete.
+	WaitForJobCompletion(ctx context.Context, job IAsynchronousJob) (err error)
+}

utils/job/manager.go

@@ -0,0 +1,156 @@
+/*
+ * Copyright (C) 2020-2023 Arm Limited or its affiliates and Contributors. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package job
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"time"
+
+	"golang.org/x/sync/errgroup"
+
+	"github.com/ARM-software/embedded-development-services-client-utils/utils/api"
+	"github.com/ARM-software/embedded-development-services-client-utils/utils/messages"
+	"github.com/ARM-software/golang-utils/utils/collection/pagination"
+	"github.com/ARM-software/golang-utils/utils/commonerrors"
+	"github.com/ARM-software/golang-utils/utils/parallelisation"
+)
+
+type Manager struct {
+	messageLoggerFactory     messages.MessageLoggerFactory
+	messagesPaginatorFactory messages.PaginatorFactory
+	backOffPeriod            time.Duration
+	fetchJobStatusFunc       func(ctx context.Context, jobName string) (IAsynchronousJob, *http.Response, error)
+}
+
+func (m *Manager) WaitForJobCompletion(ctx context.Context, job IAsynchronousJob) (err error) {
+	err = parallelisation.DetermineContextError(ctx)
+	if err != nil {
+		return
+	}
+	messageLogger, err := m.messageLoggerFactory.Create(ctx)
+	if err != nil {
+		return
+	}
+	defer func() {
+		if messageLogger != nil {
+			_ = messageLogger.Close()
+		}
+	}()
+	messagePaginator, err := m.messagesPaginatorFactory.Create(ctx)
+	if err != nil {
+		return
+	}
+	defer func() {
+		if messagePaginator != nil {
+			_ = messagePaginator.Close()
+		}
+	}()
+
+	wait, gCtx := errgroup.WithContext(ctx)
+	wait.Go(func() error {
+		return messageLogger.LogMessagesCollection(gCtx, messagePaginator)
+
+	})
+	wait.Go(func() error {
+		return m.checkForMessageStreamExhaustion(gCtx, messagePaginator, job)
+	})
+	err = wait.Wait()
+	if err != nil {
+		messageLogger.LogError(err)
+	}
+	_, err = m.HasJobCompleted(ctx, job)
+	return
+}
+
+func (m *Manager) checkForMessageStreamExhaustion(ctx context.Context, paginator pagination.IGenericStreamPaginator, job IAsynchronousJob) error {
+
+	for {
+		err := parallelisation.DetermineContextError(ctx)
+		if err != nil {
+			return err
+		}
+		completed, err := m.HasJobCompleted(ctx, job)
+		if commonerrors.Any(err, commonerrors.ErrUndefined) {
+			return err
+		}
+		if completed {
+			err = paginator.DryUp()
+			return err
+		}
+		parallelisation.SleepWithContext(ctx, m.backOffPeriod)
+	}
+}
+
+func (m *Manager) HasJobCompleted(ctx context.Context, job IAsynchronousJob) (completed bool, err error) {
+	err = parallelisation.DetermineContextError(ctx)
+	if err != nil {
+		return
+	}
+	if job == nil {
+		err = fmt.Errorf("%w: missing job", commonerrors.ErrUndefined)
+		return
+	}
+	jobName, err := job.FetchName()
+	if err != nil {
+		return
+	}
+	jobType := job.FetchType()
+	jobStatus, resp, apierr := m.fetchJobStatusFunc(ctx, jobName)
+	if resp != nil {
+		_ = resp.Body.Close()
+	}
+	err = api.CheckAPICallSuccess(ctx, fmt.Sprintf("could not fetch %v [%v]'s status", jobType, jobName), resp, apierr)
+	if err != nil {
+		return
+	}
+	if jobStatus.GetDone() {
+		completed = true
+	}
+	if jobStatus.GetError() {
+		err = fmt.Errorf("%w: %v [%v] errored: %v", commonerrors.ErrUnexpected, jobType, jobName, jobStatus.GetStatus())
+		return
+	}
+	if jobStatus.GetFailure() {
+		err = fmt.Errorf("%w: %v [%v] failed: %v", commonerrors.ErrInvalid, jobType, jobName, jobStatus.GetStatus())
+		return
+	}
+	if jobStatus.GetSuccess() {
+		return
+	}
+	if completed {
+		err = fmt.Errorf("%w: %v [%v] completed but without success: %v", commonerrors.ErrUnexpected, jobType, jobName, jobStatus.GetStatus())
+		return
+	}
+	return
+}
+
+// NewJobManager creates a new job manager.
+func NewJobManager(logger *messages.MessageLoggerFactory, backOffPeriod time.Duration,
+	fetchJobStatusFunc func(ctx context.Context, jobName string) (IAsynchronousJob, *http.Response, error),
+	fetchFirstJobMessagesPageFunc func(context.Context) (pagination.IStaticPageStream, error),
+	fetchNextJobMessagesPageFunc func(context.Context, pagination.IStaticPage) (pagination.IStaticPage, error),
+	fetchFutureJobMessagesPageFunc func(context.Context, pagination.IStaticPageStream) (pagination.IStaticPageStream, error)) (IJobManager, error) {
+	return newJobManagerFromMessageFactory(logger, backOffPeriod, fetchJobStatusFunc, messages.NewPaginatorFactory(messages.DefaultStreamExhaustionGracePeriod, backOffPeriod, fetchFirstJobMessagesPageFunc, fetchNextJobMessagesPageFunc, fetchFutureJobMessagesPageFunc))
+}
+
+func newJobManagerFromMessageFactory(logger *messages.MessageLoggerFactory, backOffPeriod time.Duration,
+	fetchJobStatusFunc func(ctx context.Context, jobName string) (IAsynchronousJob, *http.Response, error),
+	messagePaginator *messages.PaginatorFactory) (*Manager, error) {
+	if logger == nil {
+		return nil, commonerrors.ErrNoLogger
+	}
+	if messagePaginator == nil {
+		return nil, fmt.Errorf("%w: missing paginator factory", commonerrors.ErrUndefined)
+	}
+	return &Manager{
+		messageLoggerFactory:     *logger,
+		messagesPaginatorFactory: *messagePaginator,
+		backOffPeriod:            backOffPeriod,
+		fetchJobStatusFunc:       fetchJobStatusFunc,
+	}, nil
+}

utils/job/manager_test.go

@@ -0,0 +1,167 @@
+/*
+ * Copyright (C) 2020-2023 Arm Limited or its affiliates and Contributors. All rights reserved.
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package job
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/goleak"
+
+	"github.com/ARM-software/embedded-development-services-client-utils/utils/logging"
+	"github.com/ARM-software/embedded-development-services-client-utils/utils/messages"
+	"github.com/ARM-software/golang-utils/utils/commonerrors"
+)
+
+func TestManager_HasJobCompleted(t *testing.T) {
+	logger, err := logging.NewStandardClientLogger("test", nil)
+	require.NoError(t, err)
+	loggerF := messages.NewMessageLoggerFactory(logger, false, time.Nanosecond)
+
+	tests := []struct {
+		jobFunc         func() (IAsynchronousJob, error)
+		expectCompleted bool
+		expectedError   error
+	}{
+		{
+			jobFunc:         func() (IAsynchronousJob, error) { return nil, nil },
+			expectCompleted: false,
+			expectedError:   commonerrors.ErrUndefined,
+		},
+		{
+			jobFunc:         NewMockFailedAsynchronousJob,
+			expectCompleted: true,
+			expectedError:   commonerrors.ErrInvalid,
+		},
+		{
+			jobFunc:         NewMockUndoneAsynchronousJob,
+			expectCompleted: false,
+			expectedError:   nil,
+		},
+		{
+			jobFunc:         NewMockSuccessfulAsynchronousJob,
+			expectCompleted: true,
+			expectedError:   nil,
+		},
+	}
+	for i := range tests {
+		test := tests[i]
+
+		t.Run(fmt.Sprintf("#%v", i), func(t *testing.T) {
+			defer goleak.VerifyNone(t)
+			job, err := test.jobFunc()
+			factory, err := newJobManagerFromMessageFactory(loggerF, 100*time.Millisecond, func(context.Context, string) (IAsynchronousJob, *http.Response, error) {
+				return job, httptest.NewRecorder().Result(), err
+			}, messages.NewMockMessagePaginatorFactory())
+			require.NoError(t, err)
+			require.NotNil(t, factory)
+			completed, err := factory.HasJobCompleted(context.TODO(), job)
+			if test.expectedError == nil {
+				assert.NoError(t, err)
+				assert.Equal(t, test.expectCompleted, completed)
+			} else {
+				assert.Error(t, err)
+				assert.True(t, commonerrors.Any(err, test.expectedError))
+				assert.Equal(t, test.expectCompleted, completed)
+			}
+		})
+	}
+}
+
+func TestManager_checkForMessageStreamExhaustion(t *testing.T) {
+	logger, err := logging.NewStandardClientLogger("test", nil)
+	require.NoError(t, err)
+	loggerF := messages.NewMessageLoggerFactory(logger, false, time.Nanosecond)
+
+	tests := []struct {
+		jobFunc       func() (IAsynchronousJob, error)
+		expectedError error
+	}{
+		{
+			jobFunc:       func() (IAsynchronousJob, error) { return nil, nil },
+			expectedError: commonerrors.ErrUndefined,
+		},
+		{
+			jobFunc:       NewMockFailedAsynchronousJob,
+			expectedError: nil,
+		},
+		{
+			jobFunc:       NewMockSuccessfulAsynchronousJob,
+			expectedError: nil,
+		},
+	}
+	for i := range tests {
+		test := tests[i]
+
+		t.Run(fmt.Sprintf("#%v", i), func(t *testing.T) {
+			defer goleak.VerifyNone(t)
+			job, err := test.jobFunc()
+			factory, err := newJobManagerFromMessageFactory(loggerF, 100*time.Millisecond, func(context.Context, string) (IAsynchronousJob, *http.Response, error) {
+				return job, httptest.NewRecorder().Result(), err
+			}, messages.NewMockMessagePaginatorFactory())
+			require.NoError(t, err)
+			require.NotNil(t, factory)
+			messagePaginator, err := factory.messagesPaginatorFactory.Create(context.TODO())
+			assert.False(t, messagePaginator.IsRunningDry())
+
+			err = factory.checkForMessageStreamExhaustion(context.TODO(), messagePaginator, job)
+			if test.expectedError == nil {
+				assert.NoError(t, err)
+				assert.True(t, messagePaginator.IsRunningDry())
+			} else {
+				assert.Error(t, err)
+				assert.True(t, commonerrors.Any(err, test.expectedError))
+				assert.False(t, messagePaginator.IsRunningDry())
+			}
+		})
+	}
+}
+
+func TestManager_WaitForJobCompletion(t *testing.T) {
+	defer goleak.VerifyNone(t)
+	tests := []struct {
+		jobFunc       func() (IAsynchronousJob, error)
+		expectedError error
+	}{
+		{
+			jobFunc:       NewMockFailedAsynchronousJob,
+			expectedError: commonerrors.ErrInvalid,
+		},
+		{
+			jobFunc:       NewMockSuccessfulAsynchronousJob,
+			expectedError: nil,
+		},
+	}
+	for i := range tests {
+		test := tests[i]
+
+		t.Run(fmt.Sprintf("#%v", i), func(t *testing.T) {
+			// t.Parallel()
+			logger, err := logging.NewStandardClientLogger(fmt.Sprintf("test #%v", i), nil)
+			require.NoError(t, err)
+			loggerF := messages.NewMessageLoggerFactory(logger, false, time.Nanosecond)
+			job, err := test.jobFunc()
+			factory, err := newJobManagerFromMessageFactory(loggerF, time.Nanosecond, func(context.Context, string) (IAsynchronousJob, *http.Response, error) {
+				return job, httptest.NewRecorder().Result(), err
+			}, messages.NewMockMessagePaginatorFactory().UpdateRunOutTimeout(time.Nanosecond))
+			require.NoError(t, err)
+			require.NotNil(t, factory)
+			err = factory.WaitForJobCompletion(context.TODO(), job)
+			if test.expectedError == nil {
+				assert.NoError(t, err)
+			} else {
+				assert.Error(t, err)
+				assert.True(t, commonerrors.Any(err, test.expectedError))
+			}
+		})
+	}
+}