change apply to continue on error

Author: Liujingfang1

pkg/apply/applier_test.go

@@ -708,43 +708,35 @@ type fakeInfoHelper struct {
 // TODO(mortent): This has too much code in common with the
 // infoHelper implementation. We need to find a better way to structure
 // this.
-func (f *fakeInfoHelper) UpdateInfos(infos []*resource.Info) error {
+func (f *fakeInfoHelper) UpdateInfo(info *resource.Info) error {
 	mapper, err := f.factory.ToRESTMapper()
 	if err != nil {
 		return err
 	}
-	for _, info := range infos {
-		gvk := info.Object.GetObjectKind().GroupVersionKind()
-		mapping, err := mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
-		if err != nil {
-			return err
-		}
-		info.Mapping = mapping
+	gvk := info.Object.GetObjectKind().GroupVersionKind()
+	mapping, err := mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
+	if err != nil {
+		return err
+	}
+	info.Mapping = mapping
 
-		c, err := f.getClient(gvk.GroupVersion())
-		if err != nil {
-			return err
-		}
-		info.Client = c
+	c, err := f.getClient(gvk.GroupVersion())
+	if err != nil {
+		return err
 	}
+	info.Client = c
 	return nil
 }
 
-func (f *fakeInfoHelper) BuildInfos(objs []*unstructured.Unstructured) ([]*resource.Info, error) {
-	var infos []*resource.Info
-	for _, obj := range objs {
-		infos = append(infos, &resource.Info{
-			Name:      obj.GetName(),
-			Namespace: obj.GetNamespace(),
-			Source:    "unstructured",
-			Object:    obj,
-		})
-	}
-	err := f.UpdateInfos(infos)
-	if err != nil {
-		return nil, err
+func (f *fakeInfoHelper) BuildInfo(obj *unstructured.Unstructured) (*resource.Info, error) {
+	info := &resource.Info{
+		Name:      obj.GetName(),
+		Namespace: obj.GetNamespace(),
+		Source:    "unstructured",
+		Object:    obj,
 	}
-	return infos, nil
+	err := f.UpdateInfo(info)
+	return info, err
 }
 
 func (f *fakeInfoHelper) getClient(gv schema.GroupVersion) (resource.RESTClient, error) {

pkg/apply/error/error.go

@@ -0,0 +1,39 @@
+// Copyright 2019 The Kubernetes Authors.
+// SPDX-License-Identifier: Apache-2.0
+package error
+
+type UnknwonTypeError struct {
+	err error
+}
+
+func (e *UnknwonTypeError) Error() string {
+	return e.err.Error()
+}
+
+func NewUnknwonTypeError(err error) *UnknwonTypeError {
+	return &UnknwonTypeError{err: err}
+}
+
+type ApplyRunError struct {
+	err error
+}
+
+func (e *ApplyRunError) Error() string {
+	return e.err.Error()
+}
+
+func NewApplyRunError(err error) *ApplyRunError {
+	return &ApplyRunError{err: err}
+}
+
+type InitializeApplyOptionError struct {
+	err error
+}
+
+func (e *InitializeApplyOptionError) Error() string {
+	return e.err.Error()
+}
+
+func NewInitializeApplyOptionError(err error) *InitializeApplyOptionError {
+	return &InitializeApplyOptionError{err: err}
+}

pkg/apply/info/info_helper.go

@@ -15,12 +15,12 @@ import (
 
 // InfoHelper provides functions for interacting with Info objects.
 type InfoHelper interface {
-	// UpdateInfos sets the mapping and client for the provided Info
+	// UpdateInfo sets the mapping and client for the provided Info
 	// objects. This must be called at a time when all needed resource
 	// types are available in the RESTMapper.
-	UpdateInfos(infos []*resource.Info) error
+	UpdateInfo(infos *resource.Info) error
 
-	BuildInfos(objs []*unstructured.Unstructured) ([]*resource.Info, error)
+	BuildInfo(obj *unstructured.Unstructured) (*resource.Info, error)
 }
 
 func NewInfoHelper(factory util.Factory) *infoHelper {
@@ -55,6 +55,26 @@ func (ih *infoHelper) UpdateInfos(infos []*resource.Info) error {
 	return nil
 }
 
+func (ih *infoHelper) UpdateInfo(info *resource.Info) error {
+	mapper, err := ih.factory.ToRESTMapper()
+	if err != nil {
+		return err
+	}
+	gvk := info.Object.GetObjectKind().GroupVersionKind()
+	mapping, err := mapper.RESTMapping(gvk.GroupKind(), gvk.Version)
+	if err != nil {
+		return err
+	}
+	info.Mapping = mapping
+
+	c, err := ih.getClient(gvk.GroupVersion())
+	if err != nil {
+		return err
+	}
+	info.Client = c
+	return nil
+}
+
 func (ih *infoHelper) BuildInfos(objs []*unstructured.Unstructured) ([]*resource.Info, error) {
 	infos, err := object.UnstructuredsToInfos(objs)
 	if err != nil {
@@ -67,6 +87,15 @@ func (ih *infoHelper) BuildInfos(objs []*unstructured.Unstructured) ([]*resource
 	return infos, nil
 }
 
+func (ih *infoHelper) BuildInfo(obj *unstructured.Unstructured) (*resource.Info, error) {
+	info, err := object.UnstructuredToInfo(obj)
+	if err != nil {
+		return nil, err
+	}
+	err = ih.UpdateInfo(info)
+	return info, err
+}
+
 func (ih *infoHelper) ToRESTMapper() (meta.RESTMapper, error) {
 	return ih.factory.ToRESTMapper()
 }

pkg/apply/task/apply_task.go

@@ -15,6 +15,7 @@ import (
 	"k8s.io/kubectl/pkg/cmd/delete"
 	"k8s.io/kubectl/pkg/cmd/util"
 	"k8s.io/kubectl/pkg/util/slice"
+	applyerror "sigs.k8s.io/cli-utils/pkg/apply/error"
 	"sigs.k8s.io/cli-utils/pkg/apply/event"
 	"sigs.k8s.io/cli-utils/pkg/apply/info"
 	"sigs.k8s.io/cli-utils/pkg/apply/taskrunner"
@@ -72,7 +73,8 @@ func (a *ApplyTask) Start(taskContext *taskrunner.TaskContext) {
 			// the resource set.
 			objs, objsWithCRD, err := a.filterCRsWithCRDInSet(objects)
 			if err != nil {
-				a.sendTaskResult(taskContext, err)
+				taskContext.EventChannel() <- createApplyFailedEvent(object.ObjMetadata{}, err)
+				a.sendTaskResult(taskContext)
 				return
 			}
 
@@ -97,15 +99,7 @@ func (a *ApplyTask) Start(taskContext *taskrunner.TaskContext) {
 		// for that here. It could happen if this is dry-run and we removed
 		// all resources in the previous step.
 		if len(objects) == 0 {
-			a.sendTaskResult(taskContext, nil)
-			return
-		}
-
-		// Set the client and mapping fields on the provided
-		// infos so they can be applied to the cluster.
-		infos, err := a.InfoHelper.BuildInfos(objects)
-		if err != nil {
-			a.sendTaskResult(taskContext, err)
+			a.sendTaskResult(taskContext)
 			return
 		}
 
@@ -114,15 +108,31 @@ func (a *ApplyTask) Start(taskContext *taskrunner.TaskContext) {
 		ao, err := applyOptionsFactoryFunc(taskContext.EventChannel(),
 			a.ServerSideOptions, a.DryRunStrategy, a.Factory)
 		if err != nil {
-			a.sendTaskResult(taskContext, err)
+			taskContext.EventChannel() <- createApplyFailedEvent(object.ObjMetadata{},
+				applyerror.NewInitializeApplyOptionError(err))
+			a.sendTaskResult(taskContext)
 			return
 		}
-		ao.SetObjects(infos)
-		err = ao.Run()
-		if err != nil {
-			a.sendTaskResult(taskContext, err)
-			return
+
+		var infos []*resource.Info
+		for _, obj := range objects {
+			// Set the client and mapping fields on the provided
+			// info so they can be applied to the cluster.
+			info, err := a.InfoHelper.BuildInfo(obj)
+			if err != nil {
+				taskContext.EventChannel() <- createApplyFailedEvent(object.UnstructuredToObjMeta(obj),
+					applyerror.NewUnknwonTypeError(err))
+				continue
+			}
+			infos = append(infos, info)
+			ao.SetObjects([]*resource.Info{info})
+			err = ao.Run()
+			if err != nil {
+				taskContext.EventChannel() <- createApplyFailedEvent(object.UnstructuredToObjMeta(obj),
+					applyerror.NewApplyRunError(err))
+			}
 		}
+
 		// Fetch the Generation from all Infos after they have been
 		// applied.
 		for _, inf := range infos {
@@ -140,7 +150,7 @@ func (a *ApplyTask) Start(taskContext *taskrunner.TaskContext) {
 				taskContext.ResourceApplied(id, uid, gen)
 			}
 		}
-		a.sendTaskResult(taskContext, nil)
+		a.sendTaskResult(taskContext)
 	}()
 }
 
@@ -187,9 +197,9 @@ func newApplyOptions(eventChannel chan event.Event, serverSideOptions common.Ser
 	}, nil
 }
 
-func (a *ApplyTask) sendTaskResult(taskContext *taskrunner.TaskContext, err error) {
+func (a *ApplyTask) sendTaskResult(taskContext *taskrunner.TaskContext) {
 	taskContext.TaskChannel() <- taskrunner.TaskResult{
-		Err: err,
+		Err: nil,
 	}
 }
 
@@ -277,3 +287,15 @@ func buildCRDsInfo(crds []*unstructured.Unstructured) *crdsInfo {
 
 // ClearTimeout is not supported by the ApplyTask.
 func (a *ApplyTask) ClearTimeout() {}
+
+// createApplyFailedEvent is a helper function to package an apply failure event.
+func createApplyFailedEvent(id object.ObjMetadata, err error) event.Event {
+	return event.Event{
+		Type: event.ApplyType,
+		ApplyEvent: event.ApplyEvent{
+			Type:       event.ApplyEventFailed,
+			Identifier: id,
+			Error:      err,
+		},
+	}
+}