func (r *Reconciler) updateStatus(ctx reconcileRequestContext, hostingAutoscalingPolicyStatus string) error {
	return r.updateStatusWithAdditional(ctx, hostingAutoscalingPolicyStatus, "")
}
If this function returns an error, the status update has failed, and the reconciler should always requeue.
This prevents the case where a terminal status fails to persist to the Kubernetes datastore yet we stop
reconciling and thus leave the job in an unfinished state.
package hostingautoscalingpolicy
import (
	"context"
	"fmt"
	"time"
	commonv1 "github.com/aws/amazon-sagemaker-operator-for-k8s/api/v1/common"
	"github.com/aws/amazon-sagemaker-operator-for-k8s/controllers"
	"github.com/aws/amazon-sagemaker-operator-for-k8s/controllers/controllertest"
	"github.com/aws/amazon-sagemaker-operator-for-k8s/controllers/sdkutil"
	"github.com/aws/amazon-sagemaker-operator-for-k8s/controllers/sdkutil/clientwrapper"
	aws "github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/service/applicationautoscaling"
	"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"
	hostingautoscalingpolicyv1 "github.com/aws/amazon-sagemaker-operator-for-k8s/api/v1/hostingautoscalingpolicy"
)
const (
	ReconcilingAutoscalingJobStatus = "Reconciling"
	CreatedAutoscalingJobStatus = "Created"
	FailedAutoscalingJobStatus = "Error"
	DeletedAutoscalingJobStatus = "Deleted"
	MaxPolicyNameLength = 256
	ScalableDimension                   = "sagemaker:variant:DesiredInstanceCount"
	PolicyType                          = "TargetTrackingScaling"
	DefaultAutoscalingPolicyName        = "SageMakerEndpointInvocationScalingPolicy"
	DefaultSuspendedStateAttributeValue = false
)
type Reconciler struct {
	client.Client
	Log                                logr.Logger
	PollInterval                       time.Duration
	createApplicationAutoscalingClient clientwrapper.ApplicationAutoscalingClientWrapperProvider
	awsConfigLoader                    controllers.AWSConfigLoader
}
func NewHostingAutoscalingPolicyReconciler(client client.Client, log logr.Logger, pollInterval time.Duration) *Reconciler {
	return &Reconciler{
		Client:       client,
		Log:          log,
		PollInterval: pollInterval,
		createApplicationAutoscalingClient: func(cfg aws.Config) clientwrapper.ApplicationAutoscalingClientWrapper {
			sess := controllers.CreateNewAWSSessionFromConfig(cfg)
			return clientwrapper.NewApplicationAutoscalingClientWrapper(applicationautoscaling.New(sess))
		},
		awsConfigLoader: controllers.NewAWSConfigLoader(),
	}
}
func (r *Reconciler) Reconcile(req ctrl.Request) (ctrl.Result, error) {
	ctx := reconcileRequestContext{
		Context:                  context.Background(),
		Log:                      r.Log.WithValues("hostingautoscalingpolicy", req.NamespacedName),
		HostingAutoscalingPolicy: new(hostingautoscalingpolicyv1.HostingAutoscalingPolicy),
	}
	ctx.Log.Info("Getting resource")
	if err := r.Get(ctx, req.NamespacedName, ctx.HostingAutoscalingPolicy); err != nil {
		ctx.Log.Info("Unable to fetch HostingAutoscalingPolicy", "reason", err)
		if apierrs.IsNotFound(err) {
			return controllers.NoRequeue()
		}
		return controllers.RequeueImmediately()
	}
	if err := r.reconcileHostingAutoscalingPolicy(ctx); err != nil {
		ctx.Log.Info("Got an error while reconciling HostingAutoscalingPolicy, will retry", "err", err)
		return controllers.RequeueImmediately()
	}
	return controllers.RequeueAfterInterval(r.PollInterval, nil)
}
type reconcileRequestContext struct {
	context.Context
	Log                          logr.Logger
	ApplicationAutoscalingClient clientwrapper.ApplicationAutoscalingClientWrapper
	HostingAutoscalingPolicy *hostingautoscalingpolicyv1.HostingAutoscalingPolicy
	PolicyName string
	ResourceIDList []string
	ScalingPolicyDescriptionList  []*applicationautoscaling.ScalingPolicy
	ScalableTargetDescriptionList []*applicationautoscaling.DescribeScalableTargetsOutput
}
func (r *Reconciler) reconcileHostingAutoscalingPolicy(ctx reconcileRequestContext) error {
	var err error
	if ctx.HostingAutoscalingPolicy.Status.HostingAutoscalingPolicyStatus == "" {
		if err = r.updateStatus(ctx, ReconcilingAutoscalingJobStatus); 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.HostingAutoscalingPolicy.ObjectMeta) {
		if !controllers.ContainsString(ctx.HostingAutoscalingPolicy.ObjectMeta.GetFinalizers(), controllers.SageMakerResourceFinalizerName) {
			ctx.HostingAutoscalingPolicy.ObjectMeta.Finalizers = append(ctx.HostingAutoscalingPolicy.ObjectMeta.Finalizers, controllers.SageMakerResourceFinalizerName)
			if err := r.Update(ctx, ctx.HostingAutoscalingPolicy); err != nil {
				return errors.Wrap(err, "Failed to add finalizer")
			}
			ctx.Log.Info("Finalizer added")
		}
	}
	if ctx.ScalableTargetDescriptionList, ctx.ScalingPolicyDescriptionList, err = r.describeAutoscalingPolicy(ctx); err != nil && !clientwrapper.IsDescribeHAP404Error(err) {
		return r.updateStatusAndReturnError(ctx, errors.Wrap(err, "Unable to describe HostingAutoscalingPolicy."))
	}
	var action controllers.ReconcileAction
	if action, err = r.determineActionForAutoscaling(ctx); err != nil {
		return r.updateStatusAndReturnError(ctx, errors.Wrap(err, "Unable to determine action for HostingAutoscalingPolicy."))
	}
	ctx.Log.Info("Determined action for AutoscalingJob", "action", action)
	if action == controllers.NeedsDelete || action == controllers.NeedsUpdate {
		if err = r.deleteAutoscalingPolicy(ctx); err != nil {
			return r.updateStatusAndReturnError(ctx, errors.Wrap(err, "Unable to delete HostingAutoscalingPolicy"))
		}
		ctx.ScalingPolicyDescriptionList = nil
		ctx.ScalableTargetDescriptionList = nil
		ctx.HostingAutoscalingPolicy.Status.ResourceIDList = []string{}
	}
	if action == controllers.NeedsCreate || action == controllers.NeedsUpdate {
		if ctx.ScalableTargetDescriptionList, ctx.ScalingPolicyDescriptionList, err = r.applyAutoscalingPolicy(ctx); err != nil {
			return r.updateStatusAndReturnError(ctx, errors.Wrap(err, "Unable to apply HostingAutoscalingPolicy"))
		}
		r.saveCurrentResourceIDsToStatus(&ctx)
	}
	status := CreatedAutoscalingJobStatus
	if ctx.ScalingPolicyDescriptionList == nil {
		status = DeletedAutoscalingJobStatus
	}
	if err = r.updateStatus(ctx, status); err != nil {
		return err
	}
	if controllers.HasDeletionTimestamp(ctx.HostingAutoscalingPolicy.ObjectMeta) {
		if err = r.removeFinalizer(ctx); err != nil {
			return err
		}
	}
	return nil
}
func (r *Reconciler) removeFinalizer(ctx reconcileRequestContext) error {
	var err error
	ctx.HostingAutoscalingPolicy.ObjectMeta.Finalizers = controllers.RemoveString(ctx.HostingAutoscalingPolicy.ObjectMeta.Finalizers, controllers.SageMakerResourceFinalizerName)
	if err = r.Update(ctx, ctx.HostingAutoscalingPolicy); err != nil {
		return errors.Wrap(err, "Failed to remove finalizer")
	}
	ctx.Log.Info("Finalizer has been removed")
	return nil
}
func (r *Reconciler) getResourceIDListfromInputSpec(ctx *reconcileRequestContext) error {
	resourceIDListfromSpec := ctx.HostingAutoscalingPolicy.Spec.ResourceID
	for _, resourceIDfromSpec := range resourceIDListfromSpec {
		resourceID := sdkutil.ConvertAutoscalingResourceToString(*resourceIDfromSpec)
		ctx.ResourceIDList = append(ctx.ResourceIDList, *resourceID)
	}
	return nil
}
func (r *Reconciler) initializeContext(ctx *reconcileRequestContext) error {
	var err error
	if ctx.HostingAutoscalingPolicy.Spec.PolicyName != nil && len(*ctx.HostingAutoscalingPolicy.Spec.PolicyName) > 0 {
		ctx.PolicyName = *ctx.HostingAutoscalingPolicy.Spec.PolicyName
	} else {
		ctx.PolicyName = DefaultAutoscalingPolicyName
		ctx.HostingAutoscalingPolicy.Spec.PolicyName = &ctx.PolicyName
		if err := r.Update(ctx, ctx.HostingAutoscalingPolicy); err != nil {
			ctx.Log.Info("Error while updating HostingAutoscalingPolicy policyName in spec")
			return err
		}
	}
	if err = r.getResourceIDListfromInputSpec(ctx); err != nil {
		ctx.Log.Error(err, "Error reading the ResourceIDs from Spec")
		return err
	}
	if ctx.HostingAutoscalingPolicy.Status.ResourceIDList == nil {
		ctx.HostingAutoscalingPolicy.Status.ResourceIDList = []string{}
	}
	if ctx.HostingAutoscalingPolicy.Spec.ScalableDimension == nil {
		namespace := sdkutil.HostingAutoscalingPolicyServiceNamespace
		ctx.HostingAutoscalingPolicy.Spec.ServiceNamespace = &namespace
	}
	if ctx.HostingAutoscalingPolicy.Spec.ScalableDimension == nil {
		dimension := ScalableDimension
		ctx.HostingAutoscalingPolicy.Spec.ScalableDimension = &dimension
	}
	if ctx.HostingAutoscalingPolicy.Spec.PolicyType == nil {
		policyType := PolicyType
		ctx.HostingAutoscalingPolicy.Spec.PolicyType = &policyType
	}
	if ctx.HostingAutoscalingPolicy.Spec.SuspendedState == nil {
		ctx.HostingAutoscalingPolicy.Spec.SuspendedState = &commonv1.HAPSuspendedState{}
	}
	if ctx.HostingAutoscalingPolicy.Spec.SuspendedState.DynamicScalingInSuspended == nil {
		ctx.HostingAutoscalingPolicy.Spec.SuspendedState.DynamicScalingInSuspended = controllertest.ToBoolPtr(DefaultSuspendedStateAttributeValue)
	}
	if ctx.HostingAutoscalingPolicy.Spec.SuspendedState.DynamicScalingOutSuspended == nil {
		ctx.HostingAutoscalingPolicy.Spec.SuspendedState.DynamicScalingOutSuspended = controllertest.ToBoolPtr(DefaultSuspendedStateAttributeValue)
	}
	if ctx.HostingAutoscalingPolicy.Spec.SuspendedState.ScheduledScalingSuspended == nil {
		ctx.HostingAutoscalingPolicy.Spec.SuspendedState.ScheduledScalingSuspended = controllertest.ToBoolPtr(DefaultSuspendedStateAttributeValue)
	}
	awsConfig, err := r.awsConfigLoader.LoadAWSConfigWithOverrides(ctx.HostingAutoscalingPolicy.Spec.Region, ctx.HostingAutoscalingPolicy.Spec.SageMakerEndpoint)
	if err != nil {
		ctx.Log.Error(err, "Error loading AWS config")
		return err
	}
	ctx.ApplicationAutoscalingClient = r.createApplicationAutoscalingClient(awsConfig)
	ctx.Log.Info("Loaded AWS config")
	return nil
}
func (r *Reconciler) saveCurrentResourceIDsToStatus(ctx *reconcileRequestContext) {
	ctx.HostingAutoscalingPolicy.Status.ResourceIDList = ctx.ResourceIDList
}
func (r *Reconciler) determineActionForAutoscaling(ctx reconcileRequestContext) (controllers.ReconcileAction, error) {
	var err error
	if controllers.HasDeletionTimestamp(ctx.HostingAutoscalingPolicy.ObjectMeta) {
		ctx.Log.Info("Object Has Deletion Timestamp")
		if len(ctx.ScalingPolicyDescriptionList) > 0 || len(ctx.ScalableTargetDescriptionList) > 0 {
			return controllers.NeedsDelete, nil
		}
		return controllers.NeedsNoop, nil
	}
	if ctx.ScalingPolicyDescriptionList == nil || len(ctx.ScalingPolicyDescriptionList) == 0 {
		return controllers.NeedsCreate, nil
	}
	var comparison sdkutil.Comparison
	if comparison, err = sdkutil.HostingAutoscalingPolicySpecMatchesDescription(ctx.ScalableTargetDescriptionList, ctx.ScalingPolicyDescriptionList, ctx.HostingAutoscalingPolicy.Spec, ctx.HostingAutoscalingPolicy.Status.ResourceIDList); err != nil {
		return controllers.NeedsNoop, err
	}
	r.Log.Info("Compared existing HAP to actual HAP to determine if HAP needs to be updated.", "differences", comparison.Differences, "is-equal", comparison.Equal)
	if comparison.Equal {
		return controllers.NeedsNoop, nil
	}
	return controllers.NeedsUpdate, nil
}
func (r *Reconciler) describeAutoscalingPolicy(ctx reconcileRequestContext) ([]*applicationautoscaling.DescribeScalableTargetsOutput, []*applicationautoscaling.ScalingPolicy, error) {
	var err error
	var scalingPolicyDescriptionList []*applicationautoscaling.ScalingPolicy
	var scalableTargetDescriptionList []*applicationautoscaling.DescribeScalableTargetsOutput
	for _, ResourceID := range ctx.ResourceIDList {
		var scalableTargetDescription *applicationautoscaling.DescribeScalableTargetsOutput
		var scalingPolicyDescription *applicationautoscaling.ScalingPolicy
		if scalableTargetDescription, err = ctx.ApplicationAutoscalingClient.DescribeScalableTargets(ctx, ResourceID); err != nil {
			return scalableTargetDescriptionList, scalingPolicyDescriptionList, errors.Wrap(err, "Unable to describe ScalableTarget")
		}
		if scalingPolicyDescription, err = ctx.ApplicationAutoscalingClient.DescribeScalingPolicies(ctx, ctx.PolicyName, ResourceID); err != nil {
			return scalableTargetDescriptionList, scalingPolicyDescriptionList, errors.Wrap(err, "Unable to describe ScalingPolicy")
		}
		if scalableTargetDescription != nil {
			scalableTargetDescriptionList = append(scalableTargetDescriptionList, scalableTargetDescription)
		}
		if scalingPolicyDescription != nil {
			scalingPolicyDescriptionList = append(scalingPolicyDescriptionList, scalingPolicyDescription)
		}
	}
	return scalableTargetDescriptionList, scalingPolicyDescriptionList, nil
}
func (r *Reconciler) deleteAutoscalingPolicy(ctx reconcileRequestContext) error {
	var deregisterScalableTargetInput applicationautoscaling.DeregisterScalableTargetInput
	var deleteScalingPolicyInput applicationautoscaling.DeleteScalingPolicyInput
	for _, ResourceID := range ctx.HostingAutoscalingPolicy.Status.ResourceIDList {
		deleteScalingPolicyInput = sdkutil.CreateDeleteScalingPolicyInput(ctx.HostingAutoscalingPolicy.Spec, ResourceID)
		if _, err := ctx.ApplicationAutoscalingClient.DeleteScalingPolicy(ctx, &deleteScalingPolicyInput); err != nil && !clientwrapper.IsDeleteHAP404Error(err) {
			return errors.Wrap(err, "Unable to DeleteScalingPolicy")
		}
	}
	for _, ResourceID := range ctx.HostingAutoscalingPolicy.Status.ResourceIDList {
		deregisterScalableTargetInput = sdkutil.CreateDeregisterScalableTargetInput(ctx.HostingAutoscalingPolicy.Spec, ResourceID)
		if _, err := ctx.ApplicationAutoscalingClient.DeregisterScalableTarget(ctx, &deregisterScalableTargetInput); err != nil && !clientwrapper.IsDeleteHAP404Error(err) {
			return errors.Wrap(err, "Unable to DeregisterScalableTarget")
		}
	}
	return nil
}
func (r *Reconciler) applyAutoscalingPolicy(ctx reconcileRequestContext) ([]*applicationautoscaling.DescribeScalableTargetsOutput, []*applicationautoscaling.ScalingPolicy, error) {
	var registerScalableTargetInputList []applicationautoscaling.RegisterScalableTargetInput
	var putScalingPolicyInputList []applicationautoscaling.PutScalingPolicyInput
	var scalableTargetDescriptionList []*applicationautoscaling.DescribeScalableTargetsOutput
	var scalingPolicyDescriptionList []*applicationautoscaling.ScalingPolicy
	if ctx.HostingAutoscalingPolicy.Spec.PolicyName == nil || len(*ctx.HostingAutoscalingPolicy.Spec.PolicyName) == 0 {
		ctx.HostingAutoscalingPolicy.Spec.PolicyName = &ctx.PolicyName
	}
	registerScalableTargetInputList = sdkutil.CreateRegisterScalableTargetInputFromSpec(ctx.HostingAutoscalingPolicy.Spec)
	for _, registerScalableTargetInput := range registerScalableTargetInputList {
		if _, err := ctx.ApplicationAutoscalingClient.RegisterScalableTarget(ctx, &registerScalableTargetInput); err != nil {
			return scalableTargetDescriptionList, scalingPolicyDescriptionList, errors.Wrap(err, "Unable to Register Target")
		}
	}
	putScalingPolicyInputList = sdkutil.CreatePutScalingPolicyInputFromSpec(ctx.HostingAutoscalingPolicy.Spec)
	for _, putScalingPolicyInput := range putScalingPolicyInputList {
		if _, err := ctx.ApplicationAutoscalingClient.PutScalingPolicy(ctx, &putScalingPolicyInput); err != nil {
			return scalableTargetDescriptionList, scalingPolicyDescriptionList, errors.Wrap(err, "Unable to Put Scaling Policy")
		}
	}
	var err error
	if scalableTargetDescriptionList, scalingPolicyDescriptionList, err = r.describeAutoscalingPolicy(ctx); err != nil {
		return scalableTargetDescriptionList, scalingPolicyDescriptionList, r.updateStatusAndReturnError(ctx, errors.Wrap(err, "Unable to describe HostingAutoscalingPolicy."))
	}
	if len(scalableTargetDescriptionList) == 0 || len(scalingPolicyDescriptionList) == 0 {
		return nil, nil, fmt.Errorf("hosting autoscaling policy was not applied, description is empty")
	}
	return scalableTargetDescriptionList, scalingPolicyDescriptionList, nil
}