Repository: kubeflow/training-operator (language: python)

Issue:
I have some questions about the function `createNewPod` in `pkg/controller.v1/tensorflow/pod.go`
In this function, why we do `ExpectCreations` first then `CreatePodsWithControllerRef`?

In my test, If `CreatePodsWithControllerRef` failed, the controller will create a new pod when the expectation expires.


Discussion / hints:
Do you mean that you think there is a consistency problem?
Not a consistency problem.

But it may cause that too slow(use `ExpectationsTimeout` seconds) to create the pod which is created failed at the first time。

I will test it. Do you have any suggestions to fix it?
I just put `ExpectCreations` after `CreatePodsWithControllerRef`(without careful consideration), and everything runs ok; 

```go
	err := tc.PodControl.CreatePodsWithControllerRef(tfjob.Namespace, podTemplate, tfjob, controllerRef)
	if err != nil && errors.IsTimeout(err) {
		// Pod is created but its initialization has timed out.
		// If the initialization is successful eventually, the
		// controller will observe the creation via the informer.
		// If the initialization fails, or if the pod keeps
		// uninitialized for a long time, the informer will not
		// receive any update, and the controller will create a new
		// pod when the expectation expires.
		// return nil
	} else if err != nil {
		return err
	}
	tfjobKey, err := KeyFunc(tfjob)
	if err != nil {
		utilruntime.HandleError(fmt.Errorf("couldn't get key for tfjob object %#v: %v", tfjob, err))
		return err
	}
	expectationPodsKey := expectation.GenExpectationPodsKey(tfjobKey, rt)
	err = tc.Expectations.ExpectCreations(expectationPodsKey, 1)
	if err != nil {
		return err
	}
```
I found this code block in  [https://github.com/kubernetes/kubernetes/blob/master/pkg/controller/job/job_controller.go](https://github.com/kubernetes/kubernetes/blob/master/pkg/controller/job/job_controller.go).

```go
......
jm.expectations.ExpectCreations(jobKey, int(diff))
errCh = make(chan error, diff)
klog.V(4).Infof("Too few pods running job %q, need %d, creating %d", jobKey, wantActive, diff)

active += diff
wait := sync.WaitGroup{}

// Batch the pod creates. Batch sizes start at SlowStartInitialBatchSize
// and double with each successful iteration in a kind of "slow start".
// This handles attempts to start large numbers of pods that would
// likely all fail with the same error. For example a project with a
// low quota that attempts to create a large number of pods will be
// prevented from spamming the API service with the pod create requests
// after one of its pods fails.  Conveniently, this also prevents the
// event spam that those failures would generate.
for batchSize := int32(integer.IntMin(int(diff), controller.SlowStartInitialBatchSize)); diff > 0; batchSize = integer.Int32Min(2*batchSize, diff) {
	errorCount := len(errCh)
	wait.Add(int(batchSize))
	for i := int32(0); i < batchSize; i++ {
		go func() {
			defer wait.Done()
			err := jm.podControl.CreatePodsWithControllerRef(job.Namespace, &job.Spec.Template, job, metav1.NewControllerRef(job, controllerKind))
			if err != nil {
				if errors.HasStatusCause(err, v1.NamespaceTerminatingCause) {
					// If the namespace is being torn down, we can safely ignore
					// this error since all subsequent creations will fail.
					return
				}
			}
			if err != nil {
				defer utilruntime.HandleError(err)
				// Decrement the expected number of creates because the informer won't observe this pod
				klog.V(2).Infof("Failed creation, decrementing expectations for job %q/%q", job.Namespace, job.Name)
				jm.expectations.CreationObserved(jobKey)
				activeLock.Lock()
				active--
				activeLock.Unlock()
				errCh <- err
			}
		}()
	}
	wait.Wait()
        ......
```

Fix patch:
diff --git a/pkg/controller.v1/tensorflow/pod.go b/pkg/controller.v1/tensorflow/pod.go
index 5a792c67bf..1e69ba398f 100644
--- a/pkg/controller.v1/tensorflow/pod.go
+++ b/pkg/controller.v1/tensorflow/pod.go
@@ -241,6 +241,11 @@ func (tc *TFController) createNewPod(tfjob *tfv1.TFJob, rt, index string, spec *
 		// pod when the expectation expires.
 		return nil
 	} else if err != nil {
+		// Decrement the expected number of creates because the informer won't observe this pod
+		logger.Infof(
+			"Failed creation, decrementing expectations for tfjob %s/%s, key %s",
+			tfjob.Namespace, tfjob.Name, expectationPodsKey)
+		tc.Expectations.CreationObserved(expectationPodsKey)
 		return err
 	}
 	return nil


Test patch:
diff --git a/pkg/controller.v1/tensorflow/pod_test.go b/pkg/controller.v1/tensorflow/pod_test.go
index e4e1300bc3..65746cdca5 100644
--- a/pkg/controller.v1/tensorflow/pod_test.go
+++ b/pkg/controller.v1/tensorflow/pod_test.go
@@ -16,6 +16,7 @@
 package tensorflow
 
 import (
+	"fmt"
 	"os"
 	"reflect"
 	"testing"
@@ -30,6 +31,7 @@ import (
 	commonv1 "github.com/kubeflow/common/pkg/apis/common/v1"
 	"github.com/kubeflow/common/pkg/controller.v1/common"
 	"github.com/kubeflow/common/pkg/controller.v1/control"
+	"github.com/kubeflow/common/pkg/controller.v1/expectation"
 	"github.com/kubeflow/tf-operator/cmd/tf-operator.v1/app/options"
 	tfv1 "github.com/kubeflow/tf-operator/pkg/apis/tensorflow/v1"
 	tfjobclientset "github.com/kubeflow/tf-operator/pkg/client/clientset/versioned"
@@ -104,6 +106,127 @@ func TestAddPod(t *testing.T) {
 	close(stopCh)
 }
 
+func TestExpectation(t *testing.T) {
+	// Prepare the clientset and controller for the test.
+	kubeClientSet := kubeclientset.NewForConfigOrDie(&rest.Config{
+		Host: "",
+		ContentConfig: rest.ContentConfig{
+			GroupVersion: &v1.SchemeGroupVersion,
+		},
+	},
+	)
+
+	// Prepare the volcano clientset and controller for the test.
+	volcanoClientSet := volcanoclient.NewForConfigOrDie(&rest.Config{
+		Host: "",
+		ContentConfig: rest.ContentConfig{
+			GroupVersion: &batchv1beta1.SchemeGroupVersion,
+		},
+	},
+	)
+
+	config := &rest.Config{
+		Host: "",
+		ContentConfig: rest.ContentConfig{
+			GroupVersion: &tfv1.SchemeGroupVersion,
+		},
+	}
+	tfJobClientSet := tfjobclientset.NewForConfigOrDie(config)
+	ctr, _, _ := newTFController(config, kubeClientSet,
+		volcanoClientSet, tfJobClientSet, 0, options.ServerOption{})
+	ctr.tfJobInformerSynced = testutil.AlwaysReady
+	ctr.PodInformerSynced = testutil.AlwaysReady
+	ctr.ServiceInformerSynced = testutil.AlwaysReady
+
+	ctr.PodControl = &control.FakePodControl{}
+	tfJob := testutil.NewTFJob(2, 1)
+
+	var err error
+	if err = ctr.createNewPod(tfJob, "worker", "0",
+		tfJob.Spec.TFReplicaSpecs[tfv1.TFReplicaTypeWorker],
+		false, tfJob.Spec.TFReplicaSpecs); err != nil {
+		t.Errorf("Expected get nil, got error %v", err)
+	}
+
+	tfjobKey, err := KeyFunc(tfJob)
+	if err != nil {
+		t.Errorf("Expected nil, got error %v", err)
+	}
+	expectationPodsKey := expectation.GenExpectationPodsKey(tfjobKey, "worker")
+	e, found, err := ctr.Expectations.GetExpectations(expectationPodsKey)
+	if err != nil {
+		t.Errorf("Expected nil, got error %v", err)
+	}
+	if !found {
+		t.Errorf("Expected to get the corresponding expectation")
+	}
+	if add, del := e.GetExpectations(); add != 1 || del != 0 {
+		t.Errorf("Expected get 1 add and 0 del, got %d add and %d del", add, del)
+	}
+}
+
+func TestExpectationWithError(t *testing.T) {
+	// Prepare the clientset and controller for the test.
+	kubeClientSet := kubeclientset.NewForConfigOrDie(&rest.Config{
+		Host: "",
+		ContentConfig: rest.ContentConfig{
+			GroupVersion: &v1.SchemeGroupVersion,
+		},
+	},
+	)
+
+	// Prepare the volcano clientset and controller for the test.
+	volcanoClientSet := volcanoclient.NewForConfigOrDie(&rest.Config{
+		Host: "",
+		ContentConfig: rest.ContentConfig{
+			GroupVersion: &batchv1beta1.SchemeGroupVersion,
+		},
+	},
+	)
+
+	config := &rest.Config{
+		Host: "",
+		ContentConfig: rest.ContentConfig{
+			GroupVersion: &tfv1.SchemeGroupVersion,
+		},
+	}
+	tfJobClientSet := tfjobclientset.NewForConfigOrDie(config)
+	ctr, _, _ := newTFController(config, kubeClientSet,
+		volcanoClientSet, tfJobClientSet, 0, options.ServerOption{})
+	ctr.tfJobInformerSynced = testutil.AlwaysReady
+	ctr.PodInformerSynced = testutil.AlwaysReady
+	ctr.ServiceInformerSynced = testutil.AlwaysReady
+
+	ctr.PodControl = &control.FakePodControl{}
+	tfJob := testutil.NewTFJob(2, 1)
+
+	// Fake an error.
+	ctr.PodControl.(*control.FakePodControl).Err = fmt.Errorf("Fake")
+
+	var err error
+	if err = ctr.createNewPod(tfJob, "worker", "0",
+		tfJob.Spec.TFReplicaSpecs[tfv1.TFReplicaTypeWorker],
+		false, tfJob.Spec.TFReplicaSpecs); err == nil {
+		t.Errorf("Expected error, got nil")
+	}
+
+	tfjobKey, err := KeyFunc(tfJob)
+	if err != nil {
+		t.Errorf("Expected nil, got error %v", err)
+	}
+	expectationPodsKey := expectation.GenExpectationPodsKey(tfjobKey, "worker")
+	e, found, err := ctr.Expectations.GetExpectations(expectationPodsKey)
+	if err != nil {
+		t.Errorf("Expected nil, got error %v", err)
+	}
+	if !found {
+		t.Errorf("Expected to get the corresponding expectation")
+	}
+	if add, del := e.GetExpectations(); add != 0 || del != 0 {
+		t.Errorf("Expected get 0 add and 0 del, got %d add and %d del", add, del)
+	}
+}
+
 func TestClusterSpec(t *testing.T) {
 	type tc struct {
 		tfJob               *tfv1.TFJob