scheduler_operand.go

package storageoscluster

import (
	"context"
	"fmt"
	"os"
	"time"

	"github.com/ondat/operator-toolkit/declarative"
	"github.com/ondat/operator-toolkit/declarative/kubectl"
	"github.com/ondat/operator-toolkit/declarative/kustomize"
	"github.com/ondat/operator-toolkit/declarative/transform"
	eventv1 "github.com/ondat/operator-toolkit/event/v1"
	"github.com/ondat/operator-toolkit/operator/v1/operand"
	appsv1 "k8s.io/api/apps/v1"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
	kustomizetypes "sigs.k8s.io/kustomize/api/types"
	"sigs.k8s.io/kustomize/kyaml/filesys"

	storageoscomv1 "github.com/storageos/operator/api/v1"
	"github.com/storageos/operator/internal/image"
	stransform "github.com/storageos/operator/internal/transform"
)

const (
	// schedulerPackage contains the resource manifests for scheduler operand.
	schedulerPackage = "scheduler"

	// Scheduler container name.
	kubeSchedulerContainer = "storageos-scheduler"

	// Kustomize image name for container image.
	kImageKubeScheduler = "kube-scheduler"

	// Related image environment variable.
	kubeSchedulerEnvVar = "RELATED_IMAGE_KUBE_SCHEDULER"

	// Official scheduler image.
	officialSchedulerImage = "registry.k8s.io/kube-scheduler"
)

type SchedulerOperand struct {
	name            string
	client          client.Client
	kubeVersion     string
	requires        []string
	cleanupRequires []string
	requeueStrategy operand.RequeueStrategy
	fs              filesys.FileSystem
	kubectlClient   kubectl.KubectlClient
}

var _ operand.Operand = &SchedulerOperand{}

func (c *SchedulerOperand) Name() string                             { return c.name }
func (c *SchedulerOperand) Requires() []string                       { return c.requires }
func (c *SchedulerOperand) CleanupRequires() []string                { return c.cleanupRequires }
func (c *SchedulerOperand) RequeueStrategy() operand.RequeueStrategy { return c.requeueStrategy }

func (c *SchedulerOperand) ReadyCheck(ctx context.Context, obj client.Object) (bool, error) {
	ctx, span, _, log := instrumentation.Start(ctx, "SchedulerOperand.ReadyCheck")
	defer span.End()

	ctx, cancel := context.WithTimeout(ctx, time.Minute)
	defer cancel()

	cluster, ok := obj.(*storageoscomv1.StorageOSCluster)
	if !ok {
		err := fmt.Errorf("failed to convert %v to StorageOSCluster", obj)
		span.RecordError(err)
		return false, err
	}

	if cluster.Spec.DisableScheduler {
		return true, nil
	}

	// Get the deployment object and check status of the replicas.
	schedulerDep := &appsv1.Deployment{}
	key := client.ObjectKey{Name: "storageos-scheduler", Namespace: obj.GetNamespace()}
	if err := c.client.Get(ctx, key, schedulerDep); err != nil {
		return false, err
	}

	if schedulerDep.Status.AvailableReplicas > 0 {
		log.Info("Found available replicas more than 0", "availableReplicas", schedulerDep.Status.AvailableReplicas)
		return true, nil
	}

	log.Info("scheduler not ready")
	return false, nil
}

func (c *SchedulerOperand) PostReady(ctx context.Context, obj client.Object) error { return nil }

func (c *SchedulerOperand) Ensure(ctx context.Context, obj client.Object, ownerRef metav1.OwnerReference) (eventv1.ReconcilerEvent, error) {
	ctx, span, _, log := instrumentation.Start(ctx, "SchedulerOperand.Ensure")
	defer span.End()

	cluster, ok := obj.(*storageoscomv1.StorageOSCluster)
	if !ok {
		err := fmt.Errorf("failed to convert %v to StorageOSCluster", obj)
		span.RecordError(err)
		return nil, err
	}

	b, err := getSchedulerBuilder(c.fs, cluster, c.kubectlClient, c.kubeVersion)
	if err != nil {
		span.RecordError(err)
		return nil, err
	}

	if cluster.Spec.DisableScheduler {
		log.Info("scheduler feature is disabled")
		err = b.Delete(ctx)
		if err != nil {
			span.RecordError(err)
		}

		return nil, err
	}

	err = b.Apply(ctx)
	if err != nil {
		span.RecordError(err)
	}

	return nil, err
}

func (c *SchedulerOperand) Delete(ctx context.Context, obj client.Object) (eventv1.ReconcilerEvent, error) {
	ctx, span, _, _ := instrumentation.Start(ctx, "SchedulerOperand.Delete")
	defer span.End()

	cluster, ok := obj.(*storageoscomv1.StorageOSCluster)
	if !ok {
		err := fmt.Errorf("failed to convert %v to StorageOSCluster", obj)
		span.RecordError(err)
		return nil, err
	}

	b, err := getSchedulerBuilder(c.fs, cluster, c.kubectlClient, c.kubeVersion)
	if err != nil {
		span.RecordError(err)
		return nil, err
	}

	return nil, b.Delete(ctx)
}

func getSchedulerBuilder(fs filesys.FileSystem, cluster *storageoscomv1.StorageOSCluster, kcl kubectl.KubectlClient, kubeVersion string) (*declarative.Builder, error) {
	// Get image name.
	images := []kustomizetypes.Image{}

	// Get the images from the cluster spec. These overwrite the default images
	// set by the operator related images environment variables.
	if img := image.GetKustomizeImage(kImageKubeScheduler, cluster.Spec.Images.KubeSchedulerContainer, os.Getenv(kubeSchedulerEnvVar)); img != nil {
		images = append(images, *img)
	} else {
		versionImage := image.GetKustomizeImage(kImageKubeScheduler, fmt.Sprintf("%s:%s", officialSchedulerImage, kubeVersion), "")
		images = append(images, *versionImage)
	}

	// Create kubescheduler config transforms.
	configTransforms := []transform.TransformFunc{}

	// Add leader election resource lock namespace.
	rnsTF := stransform.SetKubeSchedulerLeaderElectionRNamespaceFunc(cluster.Namespace)

	configTransforms = append(configTransforms, rnsTF)

	// Create deployment transforms.
	deploymentTransforms := []transform.TransformFunc{}

	// Set resource requests and limits.
	if cluster.Spec.ContainerResources.KubeSchedulerContainer.Limits != nil || cluster.Spec.ContainerResources.KubeSchedulerContainer.Requests != nil {
		deploymentTransforms = append(deploymentTransforms, stransform.SetPodTemplateContainerResourceFunc(kubeSchedulerContainer, cluster.Spec.ContainerResources.KubeSchedulerContainer))
	}

	return declarative.NewBuilder(schedulerPackage, fs,
		declarative.WithManifestTransform(transform.ManifestTransform{
			"scheduler/config.yaml":     configTransforms,
			"scheduler/deployment.yaml": deploymentTransforms,
		}),
		declarative.WithKustomizeMutationFunc([]kustomize.MutateFunc{
			kustomize.AddNamespace(cluster.GetNamespace()),
			kustomize.AddImages(images),
		}),
		declarative.WithKubectlClient(kcl),
	)
}

func NewSchedulerOperand(
	name string,
	client client.Client,
	kubeVersion string,
	requires []string,
	cleanupRequires []string,
	requeueStrategy operand.RequeueStrategy,
	fs filesys.FileSystem,
	kcl kubectl.KubectlClient,
) *SchedulerOperand {
	return &SchedulerOperand{
		name:            name,
		client:          client,
		kubeVersion:     kubeVersion,
		requires:        requires,
		cleanupRequires: cleanupRequires,
		requeueStrategy: requeueStrategy,
		fs:              fs,
		kubectlClient:   kcl,
	}
}