Incorrect handling of errors in the worker thread

pkg/controller/queuejob/quequejob_controller_ex_test.go

@@ -0,0 +1,49 @@
+/*
+Copyright 2022, 2203 The Multi-Cluster App Dispatcher Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package queuejob_test
+
+import (
+"encoding/json"
+"errors"
+"testing"
+
+"github.com/project-codeflare/multi-cluster-app-dispatcher/pkg/controller/queuejob"
+"github.com/stretchr/testify/assert"
+)
+
+// TestNewQuotaManagerConsumerAllocationRelease function emulates multiple threads adding quota consumers and removing them
+func TestIsJsonSyntaxError(t *testing.T) {
+// Define the test table
+var tests = []struct {
+name string
+inputErr error
+expectedValue bool
+}{
+{"Nill error", nil, false},
+{"Job resource template item not define as a PodTemplate", errors.New("Job resource template item not define as a PodTemplate"), true},
+{"unexpected end of JSON input", errors.New("unexpected end of JSON input"), true},
+{"json.SyntaxError", new(json.SyntaxError), true},
+}
+// Execute tests in parallel
+for _, tc := range tests {
+tc := tc // capture range variable
+t.Run(tc.name, func(t *testing.T) {
+t.Parallel()
+assert.Equal(t, tc.expectedValue, queuejob.IsJsonSyntaxError(tc.inputErr))
+})
+}
+}

pkg/controller/queuejob/queuejob_controller_ex.go

@@ -31,6 +31,7 @@ limitations under the License.
 package queuejob
 
 import (
+"errors"
 "fmt"
 "math"
 "math/rand"
@@ -69,7 +70,7 @@ import (
 
 "k8s.io/apimachinery/pkg/runtime"
 "k8s.io/apimachinery/pkg/runtime/schema"
-"k8s.io/apimachinery/pkg/runtime/serializer/json"
+runtimeJson "k8s.io/apimachinery/pkg/runtime/serializer/json"
 
 "github.com/project-codeflare/multi-cluster-app-dispatcher/pkg/controller/queuejobresources"
 resconfigmap "github.com/project-codeflare/multi-cluster-app-dispatcher/pkg/controller/queuejobresources/configmap" // ConfigMap
@@ -675,7 +676,7 @@ func GetPodTemplate(qjobRes *arbv1.AppWrapperResource) (*v1.PodTemplateSpec, err
 rtScheme := runtime.NewScheme()
 v1.AddToScheme(rtScheme)
 
-jsonSerializer := json.NewYAMLSerializer(json.DefaultMetaFactory, rtScheme, rtScheme)
+jsonSerializer := runtimeJson.NewYAMLSerializer(runtimeJson.DefaultMetaFactory, rtScheme, rtScheme)
 
 podGVK := schema.GroupVersion{Group: v1.GroupName, Version: "v1"}.WithKind("PodTemplate")
 
@@ -1903,20 +1904,19 @@ func (cc *XController) worker() {
 
 // sync AppWrapper
 if err := cc.syncQueueJob(queuejob); err != nil {
-klog.Errorf("[worker] Failed to sync AppWrapper '%s/%s', err %#v", queuejob.Namespace, queuejob.Name, err)
 // If any error, requeue it.
 return err
 }
 
 klog.V(10).Infof("[worker] Ending %s Delay=%.6f seconds &newQJ=%p Version=%s Status=%+v", queuejob.Name, time.Now().Sub(queuejob.Status.ControllerFirstTimestamp.Time).Seconds(), queuejob, queuejob.ResourceVersion, queuejob.Status)
 return nil
 })
-if err != nil {
+if err != nil && !apierrors.IsNotFound(err) && !apierrors.IsInvalid(err) && !IsJsonSyntaxError(err) {
 klog.Warningf("[worker] Fail to process item from eventQueue, err %v. Attempting to re-enqueque...", err)
 if err00 := cc.enqueueIfNotPresent(item); err00 != nil {
-klog.Errorf("[worker] Fatal error railed to re-enqueue item, err %v", err00)
+klog.Errorf("[worker] Fatal error trying to re-enqueue item, err =%v", err00)
 } else {
-klog.Warning("[worker] Item re-enqueued")
+klog.Warning("[worker] Item re-enqueued.")
 }
 return
 }
@@ -2140,7 +2140,6 @@ func (cc *XController) manageQueueJob(qj *arbv1.AppWrapper, podPhaseChanges bool
 klog.Errorf("[manageQueueJob] Error dispatching generic item for app wrapper='%s/%s' type=%v err=%v", qj.Namespace, qj.Name, err00)
 }
 dispatchFailureMessage = fmt.Sprintf("%s/%s creation failure: %+v", qj.Namespace, qj.Name, err00)
-klog.Errorf("[manageQueueJob] Error dispatching job=%s Status=%+v err=%+v", qj.Name, qj.Status, err00)
 dispatched = false
 }
 }
@@ -2337,7 +2336,7 @@ func (cc *XController) Cleanup(appwrapper *arbv1.AppWrapper) error {
 // we call clean-up for each controller
 for _, ar := range appwrapper.Spec.AggrResources.Items {
 err00 := cc.qjobResControls[ar.Type].Cleanup(appwrapper, &ar)
-if err00 != nil && !apierrors.IsNotFound(err00) {
+if err00 != nil && !apierrors.IsNotFound(err00) && !apierrors.IsInvalid(err00) {
 klog.Errorf("[Cleanup] Error deleting item %s from app wrapper='%s/%s' err=%v.",
 ar.Type, appwrapper.Namespace, appwrapper.Name, err00)
 err = multierror.Append(err, err00)
@@ -2350,7 +2349,7 @@ func (cc *XController) Cleanup(appwrapper *arbv1.AppWrapper) error {
 if appwrapper.Spec.AggrResources.GenericItems != nil {
 for _, ar := range appwrapper.Spec.AggrResources.GenericItems {
 genericResourceName, gvk, err00 := cc.genericresources.Cleanup(appwrapper, &ar)
-if err00 != nil && !apierrors.IsNotFound(err00) {
+if err00 != nil && !apierrors.IsNotFound(err00) && !apierrors.IsInvalid(err00) && !IsJsonSyntaxError(err00) {
 klog.Errorf("[Cleanup] Error deleting generic item %s, from app wrapper='%s/%s' err=%v.",
 genericResourceName, appwrapper.Namespace, appwrapper.Name, err00)
 err = multierror.Append(err, err00)
@@ -2443,3 +2442,18 @@ func (qjm *XController) schedulingAWAtomicSet(qj *arbv1.AppWrapper) {
 qjm.schedulingAW = qj
 qjm.schedulingMutex.Unlock()
 }
+
+func IsJsonSyntaxError(err error) bool {
+var tt *jsons.SyntaxError
+if err == nil {
+return false
+} else if err.Error() == "unexpected end of JSON input" {
+return true
+} else if err.Error() == "Job resource template item not define as a PodTemplate" {
+return true
+} else if errors.As(err, &tt) {
+return true
+} else {
+return false
+}
+}