Merge from quota-management latest changes

Signed-off-by: dmatch01 <darroyo@us.ibm.com>

Author: dmatch01

CONTROLLER_VERSION

@@ -1 +1 @@
-1.29.41
+1.29.43

pkg/apis/controller/v1beta1/zz_generated.deepcopy.go

@@ -14,7 +14,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 /*
-Copyright 2019, 2021 The Multi-Cluster App Dispatcher Authors.
+Copyright 2019, 2021, 2022 The Multi-Cluster App Dispatcher Authors.
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.
@@ -420,15 +420,16 @@ func NewJobController(config *rest.Config, serverOption *options.ServerOption) *
 func (qjm *XController) PreemptQueueJobs() {
 qjobs := qjm.GetQueueJobsEligibleForPreemption()
 var updateNewJob *arbv1.AppWrapper
-for _, q := range qjobs {
-if q.Status.Running < int32(q.Spec.SchedSpec.MinAvailable) {
-newjob, e := qjm.queueJobLister.AppWrappers(q.Namespace).Get(q.Name)
+var message string
+for _, aw := range qjobs {
+if aw.Status.Running < int32(aw.Spec.SchedSpec.MinAvailable) {
+newjob, e := qjm.queueJobLister.AppWrappers(aw.Namespace).Get(aw.Name)
 if e != nil {
 continue
 }
 newjob.Status.CanRun = false
 
-message := fmt.Sprintf("Insufficient number of Running pods, minimum=%d, running=%v.", q.Spec.SchedSpec.MinAvailable, q.Status.Running)
+message = fmt.Sprintf("Insufficient number of Running pods, minimum=%d, running=%v.", aw.Spec.SchedSpec.MinAvailable, aw.Status.Running)
 cond := GenerateAppWrapperCondition(arbv1.AppWrapperCondPreemptCandidate, v1.ConditionTrue, "MinPodsNotRunning", message)
 newjob.Status.Conditions = append(newjob.Status.Conditions, cond)
 updateNewJob = newjob.DeepCopy()
@@ -437,12 +438,12 @@ func (qjm *XController) PreemptQueueJobs() {
 //ignore co-scheduler failed scheduling events. This is a temp
 //work around until co-scheduler perf issues are resolved.
 } else {
-newjob, e := qjm.queueJobLister.AppWrappers(q.Namespace).Get(q.Name)
+newjob, e := qjm.queueJobLister.AppWrappers(aw.Namespace).Get(aw.Name)
 if e != nil {
 continue
 }
 newjob.Status.CanRun = false
-message := fmt.Sprintf("Pods failed scheduling failed=%v, running=%v.", len(q.Status.PendingPodConditions), q.Status.Running)
+message = fmt.Sprintf("Pods failed scheduling failed=%v, running=%v.", len(aw.Status.PendingPodConditions), aw.Status.Running)
 index := getIndexOfMatchedCondition(newjob, arbv1.AppWrapperCondPreemptCandidate, "PodsFailedScheduling")
 if index < 0 {
 cond := GenerateAppWrapperCondition(arbv1.AppWrapperCondPreemptCandidate, v1.ConditionTrue, "PodsFailedScheduling", message)
@@ -455,9 +456,11 @@ func (qjm *XController) PreemptQueueJobs() {
 updateNewJob = newjob.DeepCopy()
 }
 if err := qjm.updateEtcd(updateNewJob, "PreemptQueueJobs - CanRun: false"); err != nil {
-klog.Errorf("Failed to update status of AppWrapper %v/%v: %v", q.Namespace, q.Name, err)
+klog.Errorf("Failed to update status of AppWrapper %v/%v: %v", aw.Namespace, aw.Name, err)
 }
-
+klog.V(4).Infof("[PreemptQueueJobs] Adding preempted AppWrapper %s/%s to backoff queue.",
+aw.Name, aw.Namespace)
+go qjm.backoff(aw, "PreemptionTriggered", string(message))
 }
 }
 func (qjm *XController) preemptAWJobs(preemptAWs []*arbv1.AppWrapper) {
@@ -1909,43 +1912,50 @@ func (cc *XController) manageQueueJob(qj *arbv1.AppWrapper, podPhaseChanges bool
 }
 
 //Cleanup function
-func (cc *XController) Cleanup(queuejob *arbv1.AppWrapper) error {
-klog.V(3).Infof("[Cleanup] begin AppWrapper %s Version=%s Status=%+v\n", queuejob.Name, queuejob.ResourceVersion, queuejob.Status)
+func (cc *XController) Cleanup(appwrapper *arbv1.AppWrapper) error {
+klog.V(3).Infof("[Cleanup] begin AppWrapper %s Version=%s Status=%+v\n", appwrapper.Name, appwrapper.ResourceVersion, appwrapper.Status)
 
 if !cc.isDispatcher {
-if queuejob.Spec.AggrResources.Items != nil {
+if appwrapper.Spec.AggrResources.Items != nil {
 // we call clean-up for each controller
-for _, ar := range queuejob.Spec.AggrResources.Items {
-cc.qjobResControls[ar.Type].Cleanup(queuejob, &ar)
+for _, ar := range appwrapper.Spec.AggrResources.Items {
+err00 := cc.qjobResControls[ar.Type].Cleanup(appwrapper, &ar)
+if err00 != nil {
+klog.Errorf("[Cleanup] Error deleting item %s from job=%s Status=%+v err=%+v.",
+ar.Type, appwrapper.Name, appwrapper.Status, err00)
+}
+}
+}
+if appwrapper.Spec.AggrResources.GenericItems != nil {
+for _, ar := range appwrapper.Spec.AggrResources.GenericItems {
+genericResourceName, gvk, err00 := cc.genericresources.Cleanup(appwrapper, &ar)
+if err00 != nil {
+klog.Errorf("[Cleanup] Error deleting generic item %s, GVK=%s.%s.%s from job=%s Status=%+v err=%+v.",
+genericResourceName, gvk.Group, gvk.Version, gvk.Kind, appwrapper.Name, appwrapper.Status, err00)
+}
 }
 }
-// if queuejob.Spec.AggrResources.GenericItems != nil {
-//	// we call clean-up for each controller
-//	for _, ar := range queuejob.Spec.AggrResources.GenericItems {
-//	cc.qjobResControls[ar.Type].Cleanup(queuejob, &ar)
-//	}
-// }
 } else {
-// klog.Infof("[Dispatcher] Cleanup: State=%s\n", queuejob.Status.State)
-//if ! queuejob.Status.CanRun && queuejob.Status.IsDispatched {
-if queuejob.Status.IsDispatched {
-queuejobKey, _ := GetQueueJobKey(queuejob)
+// klog.Infof("[Dispatcher] Cleanup: State=%s\n", appwrapper.Status.State)
+//if ! appwrapper.Status.CanRun && appwrapper.Status.IsDispatched {
+if appwrapper.Status.IsDispatched {
+queuejobKey, _ := GetQueueJobKey(appwrapper)
 if obj, ok := cc.dispatchMap[queuejobKey]; ok {
-cc.agentMap[obj].DeleteJob(queuejob)
+cc.agentMap[obj].DeleteJob(appwrapper)
 }
-queuejob.Status.IsDispatched = false
+appwrapper.Status.IsDispatched = false
 }
 }
 
 // Release quota if quota is enabled and quota manager instance exists
 if cc.serverOption.QuotaEnabled && cc.quotaManager != nil {
-cc.quotaManager.Release(queuejob)
+cc.quotaManager.Release(appwrapper)
 }
-queuejob.Status.Pending = 0
-queuejob.Status.Running = 0
-queuejob.Status.Succeeded = 0
-queuejob.Status.Failed = 0
-klog.V(10).Infof("[Cleanup] end AppWrapper %s Version=%s Status=%+v\n", queuejob.Name, queuejob.ResourceVersion, queuejob.Status)
+appwrapper.Status.Pending = 0
+appwrapper.Status.Running = 0
+appwrapper.Status.Succeeded = 0
+appwrapper.Status.Failed = 0
+klog.V(10).Infof("[Cleanup] end AppWrapper %s Version=%s Status=%+v\n", appwrapper.Name, appwrapper.ResourceVersion, appwrapper.Status)
 
 return nil
 }

pkg/controller/queuejob/queuejob_controller_ex.go

@@ -69,6 +69,122 @@ func join(strs ...string) string {
 return result
 }
 
+func (gr *GenericResources) Cleanup(aw *arbv1.AppWrapper, awr *arbv1.AppWrapperGenericResource) (genericResourceName string, groupversionkind *schema.GroupVersionKind, erro error) {
+var err error
+err = nil
+
+// Default generic source group-version-kind
+default_gvk := &schema.GroupVersionKind{
+Group: "unknown",
+Version: "unknown",
+Kind: "unknown",
+}
+// Default generic resource name
+name :=""
+
+namespaced := true
+//todo:DELETEME	dd := common.KubeClient.Discovery()
+dd := gr.clients.Discovery()
+apigroups, err := restmapper.GetAPIGroupResources(dd)
+if err != nil {
+klog.Errorf("[Cleanup] Error getting API resources, err=%#v", err)
+return name, default_gvk, err
+}
+ext := awr.GenericTemplate
+restmapper := restmapper.NewDiscoveryRESTMapper(apigroups)
+_, gvk, err := unstructured.UnstructuredJSONScheme.Decode(ext.Raw, default_gvk, nil)
+if err != nil {
+klog.Errorf("Decoding error, please check your CR! Aborting handling the resource creation, err: `%v`", err)
+return name, gvk, err
+}
+
+mapping, err := restmapper.RESTMapping(gvk.GroupKind(), gvk.Version)
+if err != nil {
+klog.Errorf("mapping error from raw object: `%v`", err)
+return name, gvk, err
+}
+
+//todo:DELETEME	restconfig := common.KubeConfig
+restconfig := gr.kubeClientConfig
+restconfig.GroupVersion = &schema.GroupVersion{
+Group: mapping.GroupVersionKind.Group,
+Version: mapping.GroupVersionKind.Version,
+}
+dclient, err := dynamic.NewForConfig(restconfig)
+if err != nil {
+klog.Errorf("[Cleanup] Error creating new dynamic client, err=%#v.", err)
+return name, gvk, err
+}
+
+_, apiresourcelist, err := dd.ServerGroupsAndResources()
+if err != nil {
+klog.Errorf("Error getting supported groups and resources, err=%#v", err)
+return name, gvk, err
+}
+rsrc := mapping.Resource
+for _, apiresourcegroup := range apiresourcelist {
+if apiresourcegroup.GroupVersion == join(mapping.GroupVersionKind.Group, "/", mapping.GroupVersionKind.Version) {
+for _, apiresource := range apiresourcegroup.APIResources {
+if apiresource.Name == mapping.Resource.Resource && apiresource.Kind == mapping.GroupVersionKind.Kind {
+rsrc = mapping.Resource
+namespaced = apiresource.Namespaced
+}
+}
+}
+}
+
+// Unmarshal generic item raw object
+var unstruct unstructured.Unstructured
+unstruct.Object = make(map[string]interface{})
+var blob interface{}
+if err = json.Unmarshal(ext.Raw, &blob); err != nil {
+klog.Errorf("[Cleanup] Error unmarshalling, err=%#v", err)
+return name, gvk, err
+}
+
+unstruct.Object = blob.(map[string]interface{}) //set object to the content of the blob after Unmarshalling
+namespace := ""
+if md, ok := unstruct.Object["metadata"]; ok {
+
+metadata := md.(map[string]interface{})
+if objectName, ok := metadata["name"]; ok {
+name = objectName.(string)
+}
+if objectns, ok := metadata["namespace"]; ok {
+namespace = objectns.(string)
+}
+}
+
+// Get the resource to see if it exists
+labelSelector := fmt.Sprintf("%s=%s, %s=%s", appwrapperJobName, aw.Name, resourceName, unstruct.GetName())
+inEtcd, err := dclient.Resource(rsrc).List(context.Background(), metav1.ListOptions{LabelSelector: labelSelector})
+if err != nil {
+return name, gvk, err
+}
+
+// Check to see if object already exists in etcd, if not, create the object.
+if inEtcd != nil || len(inEtcd.Items) > 0 {
+newName := name
+if len(newName) > 63 {
+newName = newName[:63]
+}
+
+err = deleteObject(namespaced, namespace, newName, rsrc, dclient)
+if err != nil {
+if errors.IsAlreadyExists(err) {
+klog.V(4).Infof("%v\n", err.Error())
+} else {
+klog.Errorf("[Cleanup] Error deleting the object `%v`, the error is `%v`.", newName, errors.ReasonForError(err))
+return name, gvk, err
+}
+}
+} else {
+klog.Warningf("[Cleanup] %s/%s not found using label selector: %s.\n", name, namespace, labelSelector)
+}
+
+return name, gvk, err
+}
+
 func (gr *GenericResources) SyncQueueJob(aw *arbv1.AppWrapper, awr *arbv1.AppWrapperGenericResource) (podList []*v1.Pod, err error) {
 startTime := time.Now()
 defer func() {
@@ -340,6 +456,25 @@ func createObject(namespaced bool, namespace string, name string, rsrc schema.Gr
 }
 }
 
+func deleteObject(namespaced bool, namespace string, name string, rsrc schema.GroupVersionResource, dclient dynamic.Interface) (erro error) {
+var err error
+if !namespaced {
+res := dclient.Resource(rsrc)
+err = res.Delete(context.Background(), name, metav1.DeleteOptions{})
+} else {
+res := dclient.Resource(rsrc).Namespace(namespace)
+err = res.Delete(context.Background(), name, metav1.DeleteOptions{})
+}
+
+if err != nil {
+klog.Errorf("[deleteObject] Error deleting the object `%v`, the error is `%v`.", name, errors.ReasonForError(err))
+return err
+} else {
+klog.V(4).Infof("[deleteObject] Resource `%v` deleted.\n", name)
+return nil
+}
+}
+
 func GetListOfPodResourcesFromOneGenericItem(awr *arbv1.AppWrapperGenericResource) (resource []*clusterstateapi.Resource, er error) {
 var podResourcesList []*clusterstateapi.Resource
 
@@ -380,21 +515,25 @@ func GetResources(awr *arbv1.AppWrapperGenericResource) (resource *clusterstatea
 var err error
 err = nil
 if awr.GenericTemplate.Raw != nil {
+if len(awr.CustomPodResources) > 0 {
+podresources := awr.CustomPodResources
+for _, item := range podresources {
+res := getPodResources(item)
+totalresource = totalresource.Add(res)
+}
+klog.V(4).Infof("[GetResources] Requested total allocation resource from custompodresources `%v`.\n", totalresource)
+return totalresource, err
+}
 hasContainer, replicas, containers := hasFields(awr.GenericTemplate)
 if hasContainer {
 for _, item := range containers {
 res := getContainerResources(item, replicas)
 totalresource = totalresource.Add(res)
 }
-klog.V(8).Infof("[GetResources] Requested total allocation resource from containers `%v`.\n", totalresource)
-} else {
-podresources := awr.CustomPodResources
-for _, item := range podresources {
-res := getPodResources(item)
-totalresource = totalresource.Add(res)
-}
-klog.V(8).Infof("[GetResources] Requested total allocation resource from pods `%v`.\n", totalresource)
+klog.V(4).Infof("[GetResources] Requested total allocation resource from containers `%v`.\n", totalresource)
+return totalresource, err
 }
+
 } else {
 err = fmt.Errorf("generic template raw object is not defined (nil)")
 }