func (r *Reconciler) removeFinalizer(ctx reconcileRequestContext) error {
	var err error
	ctx.ProcessingJob.ObjectMeta.Finalizers = controllers.RemoveString(ctx.ProcessingJob.ObjectMeta.Finalizers, controllers.SageMakerResourceFinalizerName)
	if err = r.Update(ctx, ctx.ProcessingJob); err != nil {
		return errors.Wrap(err, "Failed to remove finalizer")
	}
	ctx.Log.Info("Finalizer has been removed")
	return nil
}
Removes the finalizer held by our controller.
package processingjob
import (
	"context"
	"fmt"
	"time"
	"github.com/go-logr/logr"
	"github.com/pkg/errors"
	apierrs "k8s.io/apimachinery/pkg/api/errors"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/predicate"
	processingjobv1 "github.com/aws/amazon-sagemaker-operator-for-k8s/api/v1/processingjob"
	"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/aws"
	"github.com/aws/aws-sdk-go/service/sagemaker"
)
const (
	ReconcilingProcessingJobStatus = "Reconciling"
	MaxProcessingJobNameLength = 63
)
type Reconciler struct {
	client.Client
	Log                   logr.Logger
	PollInterval          time.Duration
	createSageMakerClient clientwrapper.SageMakerClientWrapperProvider
	awsConfigLoader       controllers.AWSConfigLoader
}
func NewProcessingJobReconciler(client client.Client, log logr.Logger, pollInterval time.Duration) *Reconciler {
	return &Reconciler{
		Client:       client,
		Log:          log,
		PollInterval: pollInterval,
		createSageMakerClient: func(cfg aws.Config) clientwrapper.SageMakerClientWrapper {
			sess := controllers.CreateNewAWSSessionFromConfig(cfg)
			return clientwrapper.NewSageMakerClientWrapper(sagemaker.New(sess))
		},
		awsConfigLoader: controllers.NewAWSConfigLoader(),
	}
}
func (r *Reconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
	ctx := reconcileRequestContext{
		Context:       context.Background(),
		Log:           r.Log.WithValues("processingjob", req.NamespacedName),
		ProcessingJob: new(processingjobv1.ProcessingJob),
	}
	ctx.Log.Info("Getting resource")
	if err := r.Get(ctx, req.NamespacedName, ctx.ProcessingJob); err != nil {
		ctx.Log.Info("Unable to fetch ProcessingJob job", "reason", err)
		if apierrs.IsNotFound(err) {
			return controllers.NoRequeue()
		}
		return controllers.RequeueImmediately()
	}
	if err := r.reconcileProcessingJob(ctx); err != nil {
		ctx.Log.Info("Got error while reconciling, will retry", "err", err)
		return controllers.RequeueImmediately()
	}
	switch ctx.ProcessingJob.Status.ProcessingJobStatus {
	case string(sagemaker.ProcessingJobStatusCompleted):
		fallthrough
	case string(sagemaker.ProcessingJobStatusFailed):
		fallthrough
	case string(sagemaker.ProcessingJobStatusStopped):
		return controllers.NoRequeue()
	default:
		return controllers.RequeueAfterInterval(r.PollInterval, nil)
	}
}
type reconcileRequestContext struct {
	context.Context
	Log             logr.Logger
	SageMakerClient clientwrapper.SageMakerClientWrapper
	ProcessingJob *processingjobv1.ProcessingJob
	ProcessingJobDescription *sagemaker.DescribeProcessingJobOutput
	ProcessingJobName string
}
func (r *Reconciler) reconcileProcessingJob(ctx reconcileRequestContext) error {
	var err error
	if ctx.ProcessingJob.Status.ProcessingJobStatus == "" {
		if err = r.updateStatus(ctx, controllers.InitializingJobStatus); err != nil {
			return err
		}
	}
	if err = r.initializeContext(&ctx); err != nil {
		return r.updateStatusAndReturnError(ctx, errors.Wrap(err, "Unable to initialize operator"))
	}
	if !controllers.HasDeletionTimestamp(ctx.ProcessingJob.ObjectMeta) {
		if !controllers.ContainsString(ctx.ProcessingJob.ObjectMeta.GetFinalizers(), controllers.SageMakerResourceFinalizerName) {
			ctx.ProcessingJob.ObjectMeta.Finalizers = append(ctx.ProcessingJob.ObjectMeta.Finalizers, controllers.SageMakerResourceFinalizerName)
			if err := r.Update(ctx, ctx.ProcessingJob); err != nil {
				return errors.Wrap(err, "Failed to add finalizer")
			}
			ctx.Log.Info("Finalizer added")
		}
	}
	if ctx.ProcessingJobDescription, err = ctx.SageMakerClient.DescribeProcessingJob(ctx, ctx.ProcessingJobName); err != nil {
		return r.updateStatusAndReturnError(ctx, errors.Wrap(err, "Unable to describe SageMaker processing job"))
	}
	if ctx.ProcessingJobDescription == nil {
		if controllers.HasDeletionTimestamp(ctx.ProcessingJob.ObjectMeta) {
			return r.removeFinalizer(ctx)
		}
		if err = r.createProcessingJob(ctx); err != nil {
			return r.updateStatusAndReturnError(ctx, errors.Wrap(err, "Unable to create processing job"))
		}
		if ctx.ProcessingJobDescription, err = ctx.SageMakerClient.DescribeProcessingJob(ctx, ctx.ProcessingJobName); err != nil {
			return r.updateStatusAndReturnError(ctx, errors.Wrap(err, "Unable to describe SageMaker processing job"))
		}
	}
	switch *ctx.ProcessingJobDescription.ProcessingJobStatus {
	case sagemaker.ProcessingJobStatusInProgress:
		if controllers.HasDeletionTimestamp(ctx.ProcessingJob.ObjectMeta) {
			if _, err := ctx.SageMakerClient.StopProcessingJob(ctx, ctx.ProcessingJobName); err != nil && !clientwrapper.IsStopTrainingJob404Error(err) {
				return r.updateStatusAndReturnError(ctx, errors.Wrap(err, "Unable to delete processing job"))
			}
			if ctx.ProcessingJobDescription, err = ctx.SageMakerClient.DescribeProcessingJob(ctx, ctx.ProcessingJobName); err != nil {
				return r.updateStatusAndReturnError(ctx, errors.Wrap(err, "Unable to describe SageMaker processing job"))
			}
		}
		break
	case sagemaker.ProcessingJobStatusCompleted:
		fallthrough
	case sagemaker.ProcessingJobStatusStopped, sagemaker.ProcessingJobStatusFailed:
		if controllers.HasDeletionTimestamp(ctx.ProcessingJob.ObjectMeta) {
			return r.removeFinalizer(ctx)
		}
		break
	case sagemaker.ProcessingJobStatusStopping:
		break
	default:
		unknownStateError := errors.New(fmt.Sprintf("Unknown Processing Job Status: %s", *ctx.ProcessingJobDescription.ProcessingJobStatus))
		return r.updateStatusAndReturnError(ctx, unknownStateError)
	}
	status := *ctx.ProcessingJobDescription.ProcessingJobStatus
	additional := controllers.GetOrDefault(ctx.ProcessingJobDescription.FailureReason, "")
	if err = r.updateStatusWithAdditional(ctx, status, additional); err != nil {
		return err
	}
	return nil
}
func (r *Reconciler) initializeContext(ctx *reconcileRequestContext) error {
	ctx.ProcessingJobName = controllers.GetGeneratedJobName(ctx.ProcessingJob.ObjectMeta.GetUID(), ctx.ProcessingJob.ObjectMeta.GetName(), MaxProcessingJobNameLength)
	ctx.Log.Info("ProcessingJob", "Using job name: ", ctx.ProcessingJobName)
	awsConfig, err := r.awsConfigLoader.LoadAWSConfigWithOverrides(ctx.ProcessingJob.Spec.Region, ctx.ProcessingJob.Spec.SageMakerEndpoint)
	if err != nil {
		ctx.Log.Error(err, "Error loading AWS config")
		return err
	}
	ctx.SageMakerClient = r.createSageMakerClient(awsConfig)
	ctx.Log.Info("Loaded AWS config")
	return nil
}
func (r *Reconciler) createProcessingJob(ctx reconcileRequestContext) error {
	var createProcessingJobInput sagemaker.CreateProcessingJobInput
	createProcessingJobInput = sdkutil.CreateCreateProcessingJobInputFromSpec(ctx.ProcessingJob.Spec, &ctx.ProcessingJobName)
	ctx.Log.Info("Creating ProcessingJob in SageMaker", "input", createProcessingJobInput)
	if _, err := ctx.SageMakerClient.CreateProcessingJob(ctx, &createProcessingJobInput); err != nil {
		return errors.Wrap(err, "Unable to create Processing Job")
	}
	return nil
}