fix(controller): add statefulset gc for podgroup.

installer/helm/chart/volcano/templates/controllers.yaml

@@ -72,10 +72,10 @@ rules:
     resources: ["networkpolicies"]
     verbs: ["get", "create", "delete"]
   - apiGroups: ["apps"]
-    resources: ["daemonsets", "statefulsets"]
+    resources: ["daemonsets"]
     verbs: ["get"]
   - apiGroups: ["apps"]
-    resources: ["replicasets"]
+    resources: ["replicasets", "statefulsets"]
     verbs: ["get", "list", "watch"]
   - apiGroups: ["batch"]
     resources: ["jobs"]

installer/volcano-development.yaml

@@ -4304,10 +4304,10 @@ rules:
     resources: ["networkpolicies"]
     verbs: ["get", "create", "delete"]
   - apiGroups: ["apps"]
-    resources: ["daemonsets", "statefulsets"]
+    resources: ["daemonsets"]
     verbs: ["get"]
   - apiGroups: ["apps"]
-    resources: ["replicasets"]
+    resources: ["replicasets", "statefulsets"]
     verbs: ["get", "list", "watch"]
   - apiGroups: ["batch"]
     resources: ["jobs"]

pkg/controllers/podgroup/pg_controller.go

@@ -51,6 +51,7 @@ type pgcontroller struct {
 	podInformer coreinformers.PodInformer
 	pgInformer  schedulinginformer.PodGroupInformer
 	rsInformer  appinformers.ReplicaSetInformer
+	stsInformer appinformers.StatefulSetInformer
 
 	informerFactory   informers.SharedInformerFactory
 	vcInformerFactory vcinformer.SharedInformerFactory
@@ -64,7 +65,8 @@ type pgcontroller struct {
 	pgSynced func() bool
 
 	// A store of replicaset
-	rsSynced func() bool
+	rsSynced  func() bool
+	stsSynced func() bool
 
 	queue workqueue.RateLimitingInterface
 
@@ -112,6 +114,13 @@ func (pg *pgcontroller) Initialize(opt *framework.ControllerOption) error {
 			AddFunc:    pg.addReplicaSet,
 			UpdateFunc: pg.updateReplicaSet,
 		})
+
+		pg.stsInformer = pg.informerFactory.Apps().V1().StatefulSets()
+		pg.stsSynced = pg.stsInformer.Informer().HasSynced
+		pg.stsInformer.Informer().AddEventHandler(cache.ResourceEventHandlerFuncs{
+			AddFunc:    pg.addStatefulSet,
+			UpdateFunc: pg.updateStatefulSet,
+		})
 	}
 	return nil
 }

pkg/controllers/podgroup/pg_controller_handler.go

@@ -105,6 +105,47 @@ func (pg *pgcontroller) updateReplicaSet(oldObj, newObj interface{}) {
 	pg.addReplicaSet(newObj)
 }
 
+func (pg *pgcontroller) addStatefulSet(obj interface{}) {
+	sts, ok := obj.(*appsv1.StatefulSet)
+	if !ok {
+		klog.Errorf("Failed to convert %v to appsv1.StatefulSet", obj)
+		return
+	}
+
+	if *sts.Spec.Replicas == 0 {
+		pgName := batchv1alpha1.PodgroupNamePrefix + string(sts.UID)
+		err := pg.vcClient.SchedulingV1beta1().PodGroups(sts.Namespace).Delete(context.TODO(), pgName, metav1.DeleteOptions{})
+		if err != nil && !apierrors.IsNotFound(err) {
+			klog.Errorf("Failed to delete PodGroup <%s/%s>: %v", sts.Namespace, pgName, err)
+		}
+	}
+
+	// In the rolling upgrade scenario, the addStatefulSet(replicas=0) event may be received before
+	// the updateStatefulSet(replicas=1) event, and after the addPod event for the new created pod.
+	// In this event, need to create PodGroup for the pod.
+	if *sts.Spec.Replicas > 0 {
+		selector := metav1.LabelSelector{MatchLabels: sts.Spec.Selector.MatchLabels}
+		podList, err := pg.kubeClient.CoreV1().Pods(sts.Namespace).List(context.TODO(),
+			metav1.ListOptions{LabelSelector: metav1.FormatLabelSelector(&selector)})
+		if err != nil {
+			klog.Errorf("Failed to list pods for StatefulSet <%s/%s>: %v", sts.Namespace, sts.Name, err)
+			return
+		}
+		if podList != nil && len(podList.Items) > 0 {
+			pod := podList.Items[0]
+			klog.V(4).Infof("Try to create podgroup for pod %s/%s", pod.Namespace, pod.Name)
+			err := pg.createNormalPodPGIfNotExist(&pod)
+			if err != nil {
+				klog.Errorf("Failed to create PodGroup for pod <%s/%s>: %v", pod.Namespace, pod.Name, err)
+			}
+		}
+	}
+}
+
+func (pg *pgcontroller) updateStatefulSet(oldObj, newObj interface{}) {
+	pg.addStatefulSet(newObj)
+}
+
 func (pg *pgcontroller) updatePodAnnotations(pod *v1.Pod, pgName string) error {
 	if pod.Annotations == nil {
 		pod.Annotations = make(map[string]string)