Extract owner idle actions into pkg/idler from member operator

pkg/idler/actions.go

@@ -0,0 +1,196 @@
+package idler
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/codeready-toolchain/toolchain-common/pkg/owners"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	"k8s.io/apimachinery/pkg/types"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+// SupportedScaleResources maps Camel kinds that must be idled via the scale subresource
+// (rather than a direct dynamic client patch of spec.replicas).
+var SupportedScaleResources = map[schema.GroupVersionKind]schema.GroupVersionResource{
+	schema.GroupVersion{Group: "camel.apache.org", Version: "v1"}.WithKind("Integration"):          schema.GroupVersion{Group: "camel.apache.org", Version: "v1"}.WithResource("integrations"),
+	schema.GroupVersion{Group: "camel.apache.org", Version: "v1alpha1"}.WithKind("KameletBinding"): schema.GroupVersion{Group: "camel.apache.org", Version: "v1alpha1"}.WithResource("kameletbindings"),
+}
+
+func (i *Idler) scaleToZero(ctx context.Context, objectWithGVR *owners.ObjectWithGVR) error {
+	object := objectWithGVR.Object
+	logger := log.FromContext(ctx).WithValues("kind", object.GetObjectKind().GroupVersionKind().Kind, "name", object.GetName())
+	logger.Info("Scaling controller owner to zero")
+
+	patch := []byte(`{"spec":{"replicas":0}}`)
+	if scaleGVR, ok := SupportedScaleResources[object.GetObjectKind().GroupVersionKind()]; ok {
+		logger.Info("Scaling controller owner to zero using the scale subresource")
+		_, err := i.scalesClient.Scales(object.GetNamespace()).Patch(ctx, scaleGVR, object.GetName(), types.MergePatchType, patch, metav1.PatchOptions{})
+		if err != nil {
+			return err
+		}
+		logger.Info("Controller owner scaled to zero using the scale subresource")
+		return nil
+	}
+
+	_, err := i.dynamicClient.
+		Resource(*objectWithGVR.GVR).
+		Namespace(object.GetNamespace()).
+		Patch(ctx, object.GetName(), types.MergePatchType, patch, metav1.PatchOptions{})
+	if err != nil {
+		return err
+	}
+
+	logger.Info("Controller owner scaled to zero")
+	return nil
+}
+
+// idleAAP idles AAP instance if not already idled
+func (i *Idler) idleAAP(ctx context.Context, objectWithGVR *owners.ObjectWithGVR) error {
+	return i.idleBySpecBool(ctx, objectWithGVR, "idle_aap", "AAP")
+}
+
+// idleClaw idles a Claw instance if not already idled
+func (i *Idler) idleClaw(ctx context.Context, objectWithGVR *owners.ObjectWithGVR) error {
+	return i.idleBySpecBool(ctx, objectWithGVR, "idle", "Claw")
+}
+
+// idleBySpecBool sets the given spec bool field to true when the resource is not already idled.
+func (i *Idler) idleBySpecBool(ctx context.Context, objectWithGVR *owners.ObjectWithGVR, field, label string) error {
+	name := objectWithGVR.Object.GetName()
+	logger := log.FromContext(ctx).WithValues("name", name)
+	idled, _, err := unstructured.NestedBool(objectWithGVR.Object.UnstructuredContent(), "spec", field)
+	if err != nil {
+		logger.Error(err, fmt.Sprintf("Failed to parse %s CR to get the spec.%s field", label, field))
+	}
+	if idled {
+		logger.Info(fmt.Sprintf("%s CR is already idled", label))
+		return nil
+	}
+	logger.Info(fmt.Sprintf("Idling %s", label))
+
+	patch := fmt.Appendf(nil, `{"spec":{%q:true}}`, field)
+	_, err = i.dynamicClient.
+		Resource(*objectWithGVR.GVR).
+		Namespace(objectWithGVR.Object.GetNamespace()).
+		Patch(ctx, name, types.MergePatchType, patch, metav1.PatchOptions{})
+	if err != nil {
+		return err
+	}
+
+	logger.Info(fmt.Sprintf("%s idled", label), "name", name)
+	return nil
+}
+
+func (i *Idler) deleteResource(ctx context.Context, objectWithGVR *owners.ObjectWithGVR) error {
+	logger := log.FromContext(ctx)
+	object := objectWithGVR.Object
+	logger.Info("Deleting controller owner",
+		"kind", object.GetObjectKind().GroupVersionKind().Kind, "name", object.GetName())
+	// see https://github.com/kubernetes/kubernetes/issues/20902#issuecomment-321484735
+	// also, this may be needed for the e2e tests if the call to `client.Delete` comes too quickly after creating the job,
+	// which may leave the job's pod running but orphan, hence causing a test failure (and making the test flaky)
+	propagationPolicy := metav1.DeletePropagationBackground
+
+	err := i.dynamicClient.
+		Resource(*objectWithGVR.GVR).
+		Namespace(object.GetNamespace()).
+		Delete(ctx, object.GetName(), metav1.DeleteOptions{PropagationPolicy: &propagationPolicy})
+	if err != nil {
+		return err
+	}
+
+	logger.Info("Controller owner deleted",
+		"kind", object.GetObjectKind().GroupVersionKind().Kind, "name", object.GetName())
+	return nil
+}
+
+func (i *Idler) scaleDeploymentConfigToZero(ctx context.Context, objectWithGVR *owners.ObjectWithGVR) error {
+	logger := log.FromContext(ctx)
+	object := objectWithGVR.Object
+	logger.Info("Scaling DeploymentConfig to zero", "name", object.GetName())
+	patch := []byte(`{"spec":{"replicas":0,"paused":false}}`)
+	_, err := i.dynamicClient.
+		Resource(*objectWithGVR.GVR).
+		Namespace(object.GetNamespace()).
+		Patch(ctx, object.GetName(), types.MergePatchType, patch, metav1.PatchOptions{})
+	if err != nil {
+		return err
+	}
+	log.FromContext(ctx).Info("DeploymentConfig scaled to zero", "name", object.GetName())
+	return nil
+}
+
+func (i *Idler) stopVirtualMachine(ctx context.Context, objectWithGVR *owners.ObjectWithGVR) error {
+	logger := log.FromContext(ctx)
+	object := objectWithGVR.Object
+	logger.Info("Stopping VirtualMachine", "name", object.GetName())
+	err := i.restClient.Put().
+		AbsPath(fmt.Sprintf(vmSubresourceURLFmt, "v1")).
+		Namespace(object.GetNamespace()).
+		Resource("virtualmachines").
+		Name(object.GetName()).
+		SubResource("stop").
+		Do(ctx).
+		Error()
+	if err != nil {
+		return err
+	}
+
+	logger.Info("VirtualMachine stopped", "name", object.GetName())
+	return nil
+}
+
+// idleServingRuntime idles ServingRuntime by deleting InferenceService objects that exist for longer than the timeout.
+// timeoutSeconds <= 0 deletes nothing (on-demand callers pass 0 when no cutoff applies).
+func (i *Idler) idleServingRuntime(ctx context.Context, objectWithGVR *owners.ObjectWithGVR, timeoutSeconds int32) error {
+	logger := log.FromContext(ctx)
+	namespace := objectWithGVR.Object.GetNamespace()
+
+	logger.Info("Idling ServingRuntime by deleting old InferenceService objects", "name", objectWithGVR.Object.GetName())
+
+	if timeoutSeconds <= 0 {
+		logger.Info("Skipping InferenceService cleanup because timeoutSeconds is not positive", "timeoutSeconds", timeoutSeconds)
+		return nil
+	}
+
+	inferenceServiceGVR := schema.GroupVersionResource{
+		Group:    "serving.kserve.io",
+		Version:  "v1beta1",
+		Resource: "inferenceservices",
+	}
+
+	inferenceServiceList, err := i.dynamicClient.
+		Resource(inferenceServiceGVR).
+		Namespace(namespace).
+		List(ctx, metav1.ListOptions{})
+	if err != nil {
+		return fmt.Errorf("failed to list InferenceService objects: %w", err)
+	}
+
+	cutoffTime := time.Now().Add(-time.Duration(timeoutSeconds) * time.Second)
+	var deletionErrors []error
+
+	for _, inferenceService := range inferenceServiceList.Items {
+		creationTime := inferenceService.GetCreationTimestamp().Time
+		if creationTime.Before(cutoffTime) {
+			logger.Info("Deleting old InferenceService", "name", inferenceService.GetName(), "age", time.Since(creationTime))
+
+			err := i.dynamicClient.
+				Resource(inferenceServiceGVR).
+				Namespace(namespace).
+				Delete(ctx, inferenceService.GetName(), metav1.DeleteOptions{})
+			if err != nil {
+				deletionErrors = append(deletionErrors, err)
+			} else {
+				logger.Info("InferenceService deleted", "name", inferenceService.GetName())
+			}
+		}
+	}
+
+	return errors.Join(deletionErrors...)
+}

pkg/idler/idler.go

@@ -0,0 +1,143 @@
+package idler
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/codeready-toolchain/toolchain-common/pkg/owners"
+	"github.com/go-logr/logr"
+	"github.com/redhat-cop/operator-utils/pkg/util"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/discovery"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/scale"
+	"sigs.k8s.io/controller-runtime/pkg/log"
+)
+
+const vmSubresourceURLFmt = "/apis/subresources.kubevirt.io/%s"
+
+// ErrUnsupportedKind is returned by IdleOwner when the owner kind is not in the idle matrix.
+var ErrUnsupportedKind = errors.New("unsupported idle owner kind")
+
+// Idler applies owner-idle actions (scale/stop/patch/delete) for known workload kinds.
+type Idler struct {
+	ownerFetcher  *owners.OwnerFetcher
+	dynamicClient dynamic.Interface
+	scalesClient  scale.ScalesGetter
+	restClient    rest.Interface
+}
+
+// Options configures idle actions that need caller-supplied parameters.
+type Options struct {
+	// TimeoutSeconds is the ServingRuntime InferenceService age cutoff.
+	// InferenceServices older than now-TimeoutSeconds are deleted.
+	// 0 means delete nothing.
+	TimeoutSeconds int32
+}
+
+// New creates an Idler with the clients required for the full owner-idle kind matrix.
+func New(discoveryClient discovery.ServerResourcesInterface, dynamicClient dynamic.Interface, scalesClient scale.ScalesGetter, restClient rest.Interface) *Idler {
+	return &Idler{
+		ownerFetcher:  owners.NewOwnerFetcher(discoveryClient, dynamicClient),
+		dynamicClient: dynamicClient,
+		scalesClient:  scalesClient,
+		restClient:    restClient,
+	}
+}
+
+// OwnerFetcher returns the OwnerFetcher used to walk controller owner chains.
+func (i *Idler) OwnerFetcher() *owners.OwnerFetcher {
+	return i.ownerFetcher
+}
+
+// IdleOwner applies the kind-specific idle action for a single known owner.
+// Unknown kinds return ErrUnsupportedKind so callers can skip them.
+func (i *Idler) IdleOwner(ctx context.Context, ownerWithGVR *owners.ObjectWithGVR, opts Options) error {
+	ownerKind := ownerWithGVR.Object.GetObjectKind().GroupVersionKind().Kind
+	switch ownerKind {
+	case "Deployment", "ReplicaSet", "Integration", "KameletBinding", "StatefulSet", "ReplicationController":
+		return i.scaleToZero(ctx, ownerWithGVR)
+	case "DaemonSet", "Job", "DataVolume", "PersistentVolumeClaim":
+		return i.deleteResource(ctx, ownerWithGVR)
+	case "DeploymentConfig":
+		return i.scaleDeploymentConfigToZero(ctx, ownerWithGVR)
+	case "VirtualMachine":
+		return i.stopVirtualMachine(ctx, ownerWithGVR)
+	case "AnsibleAutomationPlatform":
+		return i.idleAAP(ctx, ownerWithGVR)
+	case "Claw":
+		return i.idleClaw(ctx, ownerWithGVR)
+	case "ServingRuntime":
+		return i.idleServingRuntime(ctx, ownerWithGVR, opts.TimeoutSeconds)
+	default:
+		return ErrUnsupportedKind
+	}
+}
+
+// IdleFromPod walks the pod's owner chain and idles up to two known owners.
+// Unlike the member-operator timeout path, this always tries the second known owner
+// when present (no 105%/110% timeout percentages). It never creates Notifications.
+// Returns the kind and name of the first known owner that was attempted.
+func (i *Idler) IdleFromPod(ctx context.Context, pod *corev1.Pod, opts Options) (string, string, error) {
+	logger := log.FromContext(ctx)
+	logger.Info("Idling owners from pod")
+
+	ownerChain, fetchErr := i.ownerFetcher.GetOwners(ctx, pod)
+	if fetchErr != nil {
+		logger.Error(fetchErr, "failed to find all owners, try to idle the workload with information that is available")
+	}
+
+	LogOwnershipChain(logger, ownerChain, pod)
+
+	var topOwnerKind, topOwnerName string
+	var errToReturn error
+	for _, ownerWithGVR := range ownerChain {
+		if util.IsBeingDeleted(ownerWithGVR.Object) {
+			continue
+		}
+		owner := ownerWithGVR.Object
+		ownerKind := owner.GetObjectKind().GroupVersionKind().Kind
+
+		err := i.IdleOwner(ctx, ownerWithGVR, opts)
+		if errors.Is(err, ErrUnsupportedKind) {
+			continue
+		}
+
+		if topOwnerKind == "" {
+			topOwnerKind = ownerKind
+			topOwnerName = owner.GetName()
+			errToReturn = err
+		} else {
+			errToReturn = errors.Join(errToReturn, err)
+			break
+		}
+	}
+
+	// If no known owner was attempted, surface the GetOwners failure instead of a silent empty success.
+	if topOwnerKind == "" && fetchErr != nil {
+		return "", "", fetchErr
+	}
+	return topOwnerKind, topOwnerName, errToReturn
+}
+
+// LogOwnershipChain logs the controller ownership chain for the given pod.
+func LogOwnershipChain(logger logr.Logger, ownerChain []*owners.ObjectWithGVR, pod *corev1.Pod) {
+	if len(ownerChain) == 0 {
+		logger.Info("No ownership chain, it's a standalone pod")
+		return
+	}
+	chain := make([]string, 0, len(ownerChain)+1)
+	for _, o := range ownerChain {
+		gvk := o.Object.GetObjectKind().GroupVersionKind()
+		if gvk.Group != "" {
+			chain = append(chain, fmt.Sprintf("%s/%s.%s", o.Object.GetName(), gvk.Kind, gvk.Group))
+		} else {
+			chain = append(chain, fmt.Sprintf("%s/%s", o.Object.GetName(), gvk.Kind))
+		}
+	}
+	chain = append(chain, fmt.Sprintf("%s/Pod", pod.Name))
+	logger.Info("Ownership chain", "chain", strings.Join(chain, " -> "))
+}