Merge pull request #63 from dmatch01/generalize-appwrapper-items

Generalize appwrapper items

Author: dmatch01

pkg/apis/controller/v1alpha1/appwrapper.go

@@ -57,13 +57,21 @@ type AppWrapperSpec struct {
 // SchedSpec specifies the parameters for scheduling.
 SchedSpec SchedulingSpecTemplate `json:"schedulingSpec,omitempty" protobuf:"bytes,2,opt,name=schedulingSpec"`
 }
+// a collection of AppWrapperResource
+type AppWrapperResourceList struct {
+metav1.TypeMeta `json:",inline"`
+metav1.ListMeta `json:"metadata"`
+Items []AppWrapperResource
+GenericItems[]AppWrapperGenericResource
+}
 
 // AppWrapperService is App Wrapper service definition
 type AppWrapperService struct {
 Spec v1.ServiceSpec `json:"spec"`
 }
 
 // AppWrapperResource is App Wrapper aggregation resource
+//todo: To be depricated
 type AppWrapperResource struct {
 metav1.TypeMeta `json:",inline"`
 metav1.ObjectMeta `json:"metadata"`
@@ -88,13 +96,43 @@ type AppWrapperResource struct {
 //The template for the resource; it is now a raw text because we don't know for what resource
 //it should be instantiated
 Template runtime.RawExtension `json:"template"`
+
 }
 
-// a collection of AppWrapperResource
-type AppWrapperResourceList struct {
-metav1.TypeMeta `json:",inline"`
-metav1.ListMeta `json:"metadata"`
-Items []AppWrapperResource
+// AppWrapperResource is App Wrapper aggregation resource
+type AppWrapperGenericResource struct {
+metav1.TypeMeta `json:",inline"`
+metav1.ObjectMeta `json:"metadata"`
+// Replicas is the number of desired replicas
+DesiredAvailable int32 `json:"replicas,omitempty" protobuf:"bytes,2,opt,name=desiredavailable"`
+
+// The minimal available pods to run for this AppWrapper; the default value is nil
+MinAvailable *int32 `json:"minavailable,omitempty" protobuf:"bytes,3,opt,name=minavailable"`
+
+// The number of allocated replicas from this resource type
+Allocated int32 `json:"allocated"`
+
+// The priority of this resource
+Priority float64 `json:"priority"`
+
+// The increasing rate of priority value for this resource
+PrioritySlope float64 `json:"priorityslope"`
+
+//The template for the resource; it is now a raw text because we don't know for what resource
+//it should be instantiated
+GenericTemplate runtime.RawExtension `json:"generictemplate"`
+
+//Optional section that specifies resource requirements for non-standard k8s resources, follows same format as that
+// of standard k8s resources
+CustomPodResources []CustomPodResourceTemplate `json:"custompodresources,omitempty"`
+}
+
+type CustomPodResourceTemplate struct {
+Replicas int `json:"replicas"`
+//todo: replace with
+//Containers []Container Contain v1.ResourceRequirements
+Requests v1.ResourceList `json:"requests"`
+Limits v1.ResourceList `json:"limits"`
 }
 
 // App Wrapper resources type

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

@@ -42,6 +42,7 @@ import (
 "k8s.io/apimachinery/pkg/runtime/serializer/json"
 
 "github.com/IBM/multi-cluster-app-dispatcher/pkg/controller/queuejobresources"
+"github.com/IBM/multi-cluster-app-dispatcher/pkg/controller/queuejobresources/genericresource"
 resconfigmap "github.com/IBM/multi-cluster-app-dispatcher/pkg/controller/queuejobresources/configmap" // ConfigMap
 resdeployment "github.com/IBM/multi-cluster-app-dispatcher/pkg/controller/queuejobresources/deployment"
 resnamespace "github.com/IBM/multi-cluster-app-dispatcher/pkg/controller/queuejobresources/namespace" // NP
@@ -89,6 +90,9 @@ type XController struct {
 // controllers for these resources
 qjobResControls map[arbv1.ResourceType]queuejobresources.Interface
 
+// Captures all available resources in the cluster
+genericresources *genericresource.GenericResources
+
 clients *kubernetes.Clientset
 arbclients *clientset.Clientset
 
@@ -193,6 +197,8 @@ func NewJobController(config *rest.Config, serverOption *options.ServerOption) *
 }
 cc.metricsAdapter = adapter.New(config, cc.cache)
 
+cc.genericresources = genericresource.NewAppWrapperGenericResource(config)
+
 cc.qjobResControls = map[arbv1.ResourceType]queuejobresources.Interface{}
 RegisterAllQueueJobResourceTypes(&cc.qjobRegisteredResources)
 
@@ -447,12 +453,18 @@ func GetPodTemplate(qjobRes *arbv1.AppWrapperResource) (*v1.PodTemplateSpec, err
 }
 
 func (qjm *XController) GetAggregatedResources(cqj *arbv1.AppWrapper) *clusterstateapi.Resource {
+//todo: deprecate resource controllers
 allocated := clusterstateapi.EmptyResource()
  for _, resctrl := range qjm.qjobResControls {
  qjv := resctrl.GetAggregatedResources(cqj)
  allocated = allocated.Add(qjv)
  }
 
+for _, genericItem := range cqj.Spec.AggrResources.GenericItems {
+qjv, _ := genericresource.GetResources(&genericItem)
+allocated = allocated.Add(qjv)
+}
+
  return allocated
 }
 
@@ -483,6 +495,11 @@ func (qjm *XController) getAggregatedAvailableResourcesPriority(targetpr float64
 qjv := resctrl.GetAggregatedResources(value)
 preemptable = preemptable.Add(qjv)
 }
+for _, genericItem := range value.Spec.AggrResources.GenericItems {
+qjv, _ := genericresource.GetResources(&genericItem)
+preemptable = preemptable.Add(qjv)
+}
+
 continue
 } else if value.Status.State == arbv1.AppWrapperStateEnqueued {
 // Don't count the resources that can run but not yet realized (job orchestration pending or partially running).
@@ -491,6 +508,11 @@ func (qjm *XController) getAggregatedAvailableResourcesPriority(targetpr float64
 pending = pending.Add(qjv)
 glog.V(10).Infof("[getAggAvaiResPri] Subtract all resources %+v in resctrlType=%T for job %s which can-run is set to: %v but state is still pending.", qjv, resctrl, value.Name, value.Status.CanRun)
 }
+for _, genericItem := range value.Spec.AggrResources.GenericItems {
+qjv, _ := genericresource.GetResources(&genericItem)
+pending = pending.Add(qjv)
+glog.V(10).Infof("[getAggAvaiResPri] Subtract all resources %+v in resctrlType=%T for job %s which can-run is set to: %v but state is still pending.", qjv, genericItem, value.Name, value.Status.CanRun)
+}
 continue
 } else if value.Status.State == arbv1.AppWrapperStateActive {
 if value.Status.Pending > 0 {
@@ -500,6 +522,11 @@ func (qjm *XController) getAggregatedAvailableResourcesPriority(targetpr float64
 pending = pending.Add(qjv)
 glog.V(10).Infof("[getAggAvaiResPri] Subtract all resources %+v in resctrlType=%T for job %s which can-run is set to: %v and status set to: %s but %v pod(s) are pending.", qjv, resctrl, value.Name, value.Status.CanRun, value.Status.State, value.Status.Pending)
 }
+for _, genericItem := range value.Spec.AggrResources.GenericItems {
+qjv, _ := genericresource.GetResources(&genericItem)
+pending = pending.Add(qjv)
+glog.V(10).Infof("[getAggAvaiResPri] Subtract all resources %+v in resctrlType=%T for job %s which can-run is set to: %v and status set to: %s but %v pod(s) are pending.", qjv, genericItem, value.Name, value.Status.CanRun, value.Status.State, value.Status.Pending)
+}
 } else {
 // TODO: Hack to handle race condition when Running jobs have not yet updated the pod counts
 // This hack uses the golang struct implied behavior of defining the object without a value. In this case
@@ -510,6 +537,11 @@ func (qjm *XController) getAggregatedAvailableResourcesPriority(targetpr float64
 pending = pending.Add(qjv)
 glog.V(10).Infof("[getAggAvaiResPri] Subtract all resources %+v in resctrlType=%T for job %s which can-run is set to: %v and status set to: %s but no pod counts in the state have been defined.", qjv, resctrl, value.Name, value.Status.CanRun, value.Status.State)
 }
+for _, genericItem := range value.Spec.AggrResources.GenericItems {
+qjv, _ := genericresource.GetResources(&genericItem)
+pending = pending.Add(qjv)
+glog.V(10).Infof("[getAggAvaiResPri] Subtract all resources %+v in resctrlType=%T for job %s which can-run is set to: %v and status set to: %s but no pod counts in the state have been defined.", qjv, genericItem, value.Name, value.Status.CanRun, value.Status.State)
+}
 }
 }
 continue
@@ -1116,6 +1148,16 @@ func (cc *XController) manageQueueJob(qj *arbv1.AppWrapper) error {
 break
 }
 }
+// Handle generic resources
+for _, ar := range qj.Spec.AggrResources.GenericItems {
+glog.V(10).Infof("[worker-manageQJ] before dispatch Generic.SyncQueueJob %s &qj=%p Version=%s Status=%+v", qj.Name, qj, qj.ResourceVersion, qj.Status)
+_, err00 := cc.genericresources.SyncQueueJob(qj, &ar)
+if err00 != nil {
+glog.Errorf("[worker-manageQJ] Error dispatching job=%s Status=%+v err=%+v", qj.Name, qj.Status, err00)
+dispatched = false
+}
+}
+
 if dispatched { // set QueueJobStateRunning if all resources are successfully dispatched
 qj.Status.QueueJobState = arbv1.QueueJobStateDispatched
 glog.V(4).Infof("[worker-manageQJ] %s 4Delay=%.6f seconds AllResourceDispatchedToEtcd Version=%s Status=%+v",