Minor fixes

pkg/controller.v1beta1/tensorflow/controller.go

@@ -178,18 +178,11 @@ func (tc *TFController) Run(threadiness int, stopCh <-chan struct{}) error {
 
 	// Wait for the caches to be synced before starting workers.
 	log.Info("Waiting for informer caches to sync")
-	if ok := cache.WaitForCacheSync(stopCh, tc.tfJobInformerSynced); !ok {
-		return fmt.Errorf("failed to wait for tfjob caches to sync")
-	}
-
-	if ok := cache.WaitForCacheSync(stopCh, tc.PodInformerSynced); !ok {
-		return fmt.Errorf("failed to wait for pod caches to sync")
-	}
 
-	if ok := cache.WaitForCacheSync(stopCh, tc.ServiceInformerSynced); !ok {
-		return fmt.Errorf("failed to wait for service caches to sync")
+	if ok := cache.WaitForCacheSync(stopCh, tc.tfJobInformerSynced,
+		tc.PodInformerSynced, tc.ServiceInformerSynced); !ok {
+		return fmt.Errorf("failed to wait for caches to sync")
 	}
-
 	log.Infof("Starting %v workers", threadiness)
 	// Launch workers to process TFJob resources.
 	for i := 0; i < threadiness; i++ {
@@ -214,15 +207,25 @@ func (tc *TFController) runWorker() {
 // processNextWorkItem will read a single work item off the workqueue and
 // attempt to process it, by calling the syncHandler.
 func (tc *TFController) processNextWorkItem() bool {
-	key, quit := tc.WorkQueue.Get()
+	obj, quit := tc.WorkQueue.Get()
 	if quit {
 		return false
 	}
-	defer tc.WorkQueue.Done(key)
-
-	logger := tflogger.LoggerForKey(key.(string))
+	defer tc.WorkQueue.Done(obj)
+
+	var key string
+	var ok bool
+	if key, ok = obj.(string); !ok {
+		// As the item in the workqueue is actually invalid, we call
+		// Forget here else we'd go into a loop of attempting to
+		// process a work item that is invalid.
+		tc.WorkQueue.Forget(obj)
+		utilruntime.HandleError(fmt.Errorf("expected string in workqueue but got %#v", obj))
+		return true
+	}
+	logger := tflogger.LoggerForKey(key)
 
-	tfJob, err := tc.getTFJobFromKey(key.(string))
+	tfJob, err := tc.getTFJobFromKey(key)
 	if err != nil {
 		if err == errNotExists {
 			logger.Infof("TFJob has been deleted: %v", key)
@@ -241,7 +244,7 @@ func (tc *TFController) processNextWorkItem() bool {
 	}
 
 	// Sync TFJob to match the actual state to this desired state.
-	forget, err := tc.syncHandler(key.(string))
+	forget, err := tc.syncHandler(key)
 	if err == nil {
 		if forget {
 			tc.WorkQueue.Forget(key)

pkg/controller.v1beta1/tensorflow/informer.go

@@ -77,11 +77,7 @@ func (tc *TFController) getTFJobFromKey(key string) (*tfv1beta1.TFJob, error) {
 		return nil, errNotExists
 	}
 
-	tfjob, err := tfJobFromUnstructured(obj)
-	if err != nil {
-		return nil, err
-	}
-	return tfjob, nil
+	return tfJobFromUnstructured(obj)
 }
 
 func tfJobFromUnstructured(obj interface{}) (*tfv1beta1.TFJob, error) {

pkg/controller.v1beta1/tensorflow/job.go

@@ -18,7 +18,7 @@ import (
 )
 
 const (
-	failedMarshalTFJobReason = "FailedInvalidTFJobSpec"
+	failedMarshalTFJobReason = "InvalidTFJobSpec"
 )
 
 // When a pod is added, set the defaults and enqueue the current tfjob.