func (s hpoTrainingJobSpawner) reconcileSpawnedTrainingJobDeletion(ctx context.Context, trainingJobNamespacedName types.NamespacedName) error {
	var trainingJob trainingjobv1.TrainingJob
	if err := s.K8sClient.Get(ctx, trainingJobNamespacedName, &trainingJob); err != nil {
		return controllers.IgnoreNotFound(err)
	}
	needsRemoveFinalizer := controllers.ContainsString(trainingJob.ObjectMeta.GetFinalizers(), hpoTrainingJobOwnershipFinalizer)
	needsDelete := trainingJob.ObjectMeta.GetDeletionTimestamp().IsZero()
	if needsRemoveFinalizer {
		s.Log.Info("Removing HPO ownership finalizer from TrainingJob", "trainingJobName", trainingJob.Status.SageMakerTrainingJobName)
		trainingJob.ObjectMeta.Finalizers = controllers.RemoveString(trainingJob.ObjectMeta.GetFinalizers(), hpoTrainingJobOwnershipFinalizer)
		if err := s.K8sClient.Update(ctx, &trainingJob); err != nil {
			return controllers.IgnoreNotFound(err)
		}
	}
	if needsDelete {
		s.Log.Info("Deleting TrainingJob", "trainingJobName", trainingJob.Status.SageMakerTrainingJobName)
		if err := s.K8sClient.Delete(ctx, &trainingJob); err != nil {
			return controllers.IgnoreNotFound(err)
		}
	}
	return nil
}
Delete a single training job and remove its finalizer, if present.
Returns an error if any operation failed and needs to be retried.
package hyperparametertuningjob
import (
	"context"
	"fmt"
	"sync"
	"github.com/pkg/errors"
	hpojobv1 "github.com/aws/amazon-sagemaker-operator-for-k8s/api/v1/hyperparametertuningjob"
	trainingjobv1 "github.com/aws/amazon-sagemaker-operator-for-k8s/api/v1/trainingjob"
	"github.com/aws/amazon-sagemaker-operator-for-k8s/controllers"
	"github.com/aws/amazon-sagemaker-operator-for-k8s/controllers/sdkutil"
	"github.com/aws/amazon-sagemaker-operator-for-k8s/controllers/sdkutil/clientwrapper"
	"github.com/aws/aws-sdk-go/service/sagemaker"
	"github.com/go-logr/logr"
	apierrs "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
)
const hpoTrainingJobOwnershipFinalizer = "sagemaker-operator-hpo-trainingjob"
type HPOTrainingJobSpawner interface {
	SpawnMissingTrainingJobs(ctx context.Context, hpoJob hpojobv1.HyperparameterTuningJob)
	DeleteSpawnedTrainingJobs(ctx context.Context, hpoJob hpojobv1.HyperparameterTuningJob) error
}
func NewHPOTrainingJobSpawner(k8sClient client.Client, log logr.Logger, sageMakerClient clientwrapper.SageMakerClientWrapper) HPOTrainingJobSpawner {
	return &hpoTrainingJobSpawner{
		K8sClient:       k8sClient,
		Log:             log.WithName("HPOTrainingJobSpawner"),
		SageMakerClient: sageMakerClient,
	}
}
type HPOTrainingJobSpawnerProvider func(k8sClient client.Client, log logr.Logger, sageMakerClient clientwrapper.SageMakerClientWrapper) HPOTrainingJobSpawner
type hpoTrainingJobSpawner struct {
	HPOTrainingJobSpawner
	K8sClient       client.Client
	Log             logr.Logger
	SageMakerClient clientwrapper.SageMakerClientWrapper
}
func (s hpoTrainingJobSpawner) SpawnMissingTrainingJobs(ctx context.Context, hpoJob hpojobv1.HyperparameterTuningJob) {
	hpoJobName := *hpoJob.Spec.HyperParameterTuningJobName
	k8sNamespace := hpoJob.ObjectMeta.GetNamespace()
	awsRegion := *hpoJob.Spec.Region
	sageMakerEndpoint := hpoJob.Spec.SageMakerEndpoint
	var nextToken *string = nil
	var wg sync.WaitGroup
	for {
		hpoTrainingJobOutput, err := s.SageMakerClient.ListTrainingJobsForHyperParameterTuningJob(ctx, hpoJobName, nextToken)
		if err != nil {
			s.Log.Error(err, "Error while getting training jobs")
			break
		}
		list := hpoTrainingJobOutput.TrainingJobSummaries
		s.Log.Info("Got a page of TrainingJobs to spawn", "length", len(list))
		for _, trainingJob := range list {
			wg.Add(1)
			go func(trainingJob *sagemaker.HyperParameterTrainingJobSummary) {
				defer wg.Done()
				if exists, err := s.trainingJobExistsInKubernetes(ctx, *trainingJob.TrainingJobName, k8sNamespace); exists || (err != nil) {
					return
				}
				err := s.spawnTrainingJobInKubernetes(ctx, awsRegion, sageMakerEndpoint, *trainingJob.TrainingJobName, k8sNamespace)
				if err != nil {
					s.Log.Info("Unable to spawn missing training job", "err", err)
				}
			}(trainingJob)
		}
		if hpoTrainingJobOutput.NextToken == nil {
			break
		}
		nextToken = hpoTrainingJobOutput.NextToken
	}
	wg.Wait()
}
func (s hpoTrainingJobSpawner) trainingJobExistsInKubernetes(ctx context.Context, trainingJobName, k8sNamespace string) (bool, error) {
	var trainingJob trainingjobv1.TrainingJob
	err := s.K8sClient.Get(ctx, types.NamespacedName{
		Namespace: k8sNamespace,
		Name:      trainingJobName,
	}, &trainingJob)
	if err == nil {
		return true, nil
	}
	if !apierrs.IsNotFound(err) {
		s.Log.Info("Unable to check if TrainingJob spawn needed because k8sClient.Get failed", "err", err)
		return false, err
	}
	return false, nil
}
func (s hpoTrainingJobSpawner) spawnTrainingJobInKubernetes(ctx context.Context, awsRegion string, sageMakerEndpoint *string, trainingJobName, k8sNamespace string) error {
	trainingJobSpec, err := s.getKubernetesTrainingJobSpec(ctx, trainingJobName)
	if err != nil {
		return errors.Wrap(err, "Unable to create job spec")
	}
	trainingJobSpec.Region = &awsRegion
	trainingJobSpec.SageMakerEndpoint = sageMakerEndpoint
	trainingJob := &trainingjobv1.TrainingJob{
		ObjectMeta: metav1.ObjectMeta{
			Name:       trainingJobName,
			Namespace:  k8sNamespace,
			Finalizers: []string{hpoTrainingJobOwnershipFinalizer},
		},
		Spec: *trainingJobSpec,
	}
	if err := s.K8sClient.Create(ctx, trainingJob); err != nil {
		return errors.Wrap(err, "Unable to create k8s job")
	}
	s.Log.Info("Successfully spawned TrainingJob", "trainingJobName", trainingJobName)
	return nil
}
func (s hpoTrainingJobSpawner) getKubernetesTrainingJobSpec(ctx context.Context, trainingJobName string) (*trainingjobv1.TrainingJobSpec, error) {
	response, err := s.SageMakerClient.DescribeTrainingJob(ctx, trainingJobName)
	if err != nil {
		return nil, errors.Wrap(err, "An error occured while describing the TrainingJob from SageMaker")
	} else if response == nil {
		return nil, errors.Wrap(err, "Unable to get TrainingJob description from SageMaker")
	}
	spec, err := sdkutil.CreateTrainingJobSpecFromDescription(*response)
	return &spec, err
}
func (s hpoTrainingJobSpawner) DeleteSpawnedTrainingJobs(ctx context.Context, hpoJob hpojobv1.HyperparameterTuningJob) error {
	hpoJobName := *hpoJob.Spec.HyperParameterTuningJobName
	k8sNamespace := hpoJob.ObjectMeta.GetNamespace()
	errors := s.deleteSpawnedTrainingJobsConcurrently(ctx, hpoJobName, k8sNamespace)
	if len(errors) > 0 {
		return fmt.Errorf("Error(s) occurred while deleting spawned training jobs: %+v", errors)
	}
	return nil
}
func (s hpoTrainingJobSpawner) deleteSpawnedTrainingJobsConcurrently(ctx context.Context, hpoJobName, k8sNamespace string) []error {
	var wg sync.WaitGroup
	errorsChannel := make(chan error)
	errors := []error{}
	var nextToken *string = nil
	for {
		hpoTrainingJobOutput, err := s.SageMakerClient.ListTrainingJobsForHyperParameterTuningJob(ctx, hpoJobName, nextToken)
		if err != nil {
			s.Log.Error(err, "Error while getting training jobs")
			errors = append(errors, err)
			break
		}
		list := hpoTrainingJobOutput.TrainingJobSummaries
		for _, trainingJobSummary := range list {
			wg.Add(1)
			go func(trainingJobSummary *sagemaker.HyperParameterTrainingJobSummary) {
				defer wg.Done()
				key := types.NamespacedName{
					Namespace: k8sNamespace,
					Name:      *trainingJobSummary.TrainingJobName,
				}
				if err := s.reconcileSpawnedTrainingJobDeletion(ctx, key); err != nil {
					errorsChannel <- err
				}
			}(trainingJobSummary)
		}
		if hpoTrainingJobOutput.NextToken == nil {
			break
		}
		nextToken = hpoTrainingJobOutput.NextToken
	}
	go func() {
		defer close(errorsChannel)
		wg.Wait()
	}()
	for err := range errorsChannel {
		errors = append(errors, err)
	}
	return errors
}