Fix downstream bug

after deleting dataset, the event of deleting job can be sent to LC. Signed-off-by: JimmyYang20 <[email protected]>
kubeedge · Dec 15, 2021 · 3fc0a15 · 3fc0a15
1 parent 1edba9a
commit 3fc0a15
Show file tree

Hide file tree

Showing 2 changed files with 29 additions and 19 deletions.
diff --git a/pkg/globalmanager/controllers/incrementallearning/downstream.go b/pkg/globalmanager/controllers/incrementallearning/downstream.go
@@ -62,18 +62,15 @@ func (c *Controller) syncToEdge(eventType watch.EventType, obj interface{}) erro
 	// more details at https://github.com/kubernetes/kubernetes/issues/3030
 	job.Kind = KindName
 
-	jobConditions := job.Status.Conditions
-	if len(jobConditions) == 0 {
-		return nil
-	}
-
 	dataName := job.Spec.Dataset.Name
+	// LC has dataset object on this node that may call dataset node
+	var dsNodeName string
 	ds, err := c.client.Datasets(job.Namespace).Get(context.TODO(), dataName, metav1.GetOptions{})
 	if err != nil {
-		return fmt.Errorf("dataset(%s/%s) not found", job.Namespace, dataName)
+		klog.Errorf("not found job(name=%s/%s)'s dataset, error: %v", job.Kind, job.Name, err)
+	} else {
+		dsNodeName = ds.Spec.NodeName
 	}
-	// LC has dataset object on this node that may call dataset node
-	dsNodeName := ds.Spec.NodeName
 
 	var trainNodeName string
 	var evalNodeName string
@@ -102,6 +99,15 @@ func (c *Controller) syncToEdge(eventType watch.EventType, obj interface{}) erro
 		return nil
 	}
 
+	if dsNodeName == "" {
+		return nil
+	}
+
+	jobConditions := job.Status.Conditions
+	if len(jobConditions) == 0 {
+		return nil
+	}
+
 	latestCondition := jobConditions[len(jobConditions)-1]
 	currentType := latestCondition.Type
 	jobStage := latestCondition.Stage

diff --git a/pkg/globalmanager/controllers/lifelonglearning/downstream.go b/pkg/globalmanager/controllers/lifelonglearning/downstream.go
@@ -18,8 +18,6 @@ package lifelonglearning
 
 import (
 	"context"
-	"fmt"
-
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/apimachinery/pkg/watch"
@@ -40,18 +38,15 @@ func (c *Controller) syncToEdge(eventType watch.EventType, obj interface{}) erro
 	// more details at https://github.com/kubernetes/kubernetes/issues/3030
 	job.Kind = KindName
 
-	jobConditions := job.Status.Conditions
-	if len(jobConditions) == 0 {
-		return nil
-	}
-
 	dataName := job.Spec.Dataset.Name
+	// LC has dataset object on this node that may call dataset node
+	var dsNodeName string
 	ds, err := c.client.Datasets(job.Namespace).Get(context.TODO(), dataName, metav1.GetOptions{})
 	if err != nil {
-		return fmt.Errorf("dataset(%s/%s) not found", job.Namespace, dataName)
+		klog.Errorf("not found job(name=%s/%s)'s dataset, error: %v", job.Kind, job.Name, err)
+	} else {
+		dsNodeName = ds.Spec.NodeName
 	}
-	// LC has dataset object on this node that may call dataset node
-	dsNodeName := ds.Spec.NodeName
 
 	var trainNodeName string
 	var evalNodeName string
@@ -78,13 +73,22 @@ func (c *Controller) syncToEdge(eventType watch.EventType, obj interface{}) erro
 		return nil
 	}
 
+	if dsNodeName == "" {
+		return nil
+	}
+
+	jobConditions := job.Status.Conditions
+	if len(jobConditions) == 0 {
+		return nil
+	}
+
 	latestCondition := jobConditions[len(jobConditions)-1]
 	currentType := latestCondition.Type
 	jobStage := latestCondition.Stage
 
 	syncJobWithNodeName := func(nodeName string) {
 		if err := c.sendToEdgeFunc(nodeName, eventType, job); err != nil {
-			klog.Warningf("Error to sync incremental learning job %s to node %s in stage %s: %v",
+			klog.Warningf("Error to sync lifelong learning job %s to node %s in stage %s: %v",
 				job.Name, nodeName, jobStage, err)
 		}
 	}