Merge pull request #286 from Liujingfang1/apply

change apply to continue on error

Author: k8s-ci-robot

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 UnknownTypeError struct {
+	err error
+}
+
+func (e *UnknownTypeError) Error() string {
+	return e.err.Error()
+}
+
+func NewUnknownTypeError(err error) *UnknownTypeError {
+	return &UnknownTypeError{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/event/applyeventtype_string.go

@@ -81,7 +81,6 @@ type ApplyEventType int
 const (
 	ApplyEventResourceUpdate ApplyEventType = iota
 	ApplyEventCompleted
-	ApplyEventFailed
 )
 
 //go:generate stringer -type=ApplyEventOperation

pkg/apply/event/event.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
-	// objects. This must be called at a time when all needed resource
+	// UpdateInfo sets the mapping and client for the provided Info
+	// object. This must be called at a time when all needed resource
 	// types are available in the RESTMapper.
-	UpdateInfos(infos []*resource.Info) error
+	UpdateInfo(info *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/info/info_helper.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,22 +73,16 @@ func (a *ApplyTask) Start(taskContext *taskrunner.TaskContext) {
 			// the resource set.
 			objs, objsWithCRD, err := a.filterCRsWithCRDInSet(objects)
 			if err != nil {
-				a.sendTaskResult(taskContext, err)
+				sendBatchApplyEvents(taskContext, objs, err)
+				a.sendTaskResult(taskContext)
 				return
 			}
 
 			// Just send the apply event here. We know it must be a
 			// Created event since the type didn't already exist in the
 			// cluster.
 			for _, obj := range objsWithCRD {
-				taskContext.EventChannel() <- event.Event{
-					Type: event.ApplyType,
-					ApplyEvent: event.ApplyEvent{
-						Type:      event.ApplyEventResourceUpdate,
-						Operation: event.Created,
-						Object:    obj,
-					},
-				}
+				taskContext.EventChannel() <- createApplyEvent(object.UnstructuredToObjMeta(obj), event.Created, nil)
 			}
 			// Update the resource set to no longer include the CRs.
 			objects = objs
@@ -97,15 +92,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 +101,30 @@ 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)
+			sendBatchApplyEvents(taskContext, objects, 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() <- createApplyEvent(
+					object.UnstructuredToObjMeta(obj), event.Unchanged, applyerror.NewUnknownTypeError(err))
+				continue
+			}
+			infos = append(infos, info)
+			ao.SetObjects([]*resource.Info{info})
+			err = ao.Run()
+			if err != nil {
+				taskContext.EventChannel() <- createApplyEvent(
+					object.UnstructuredToObjMeta(obj), event.Unchanged, applyerror.NewApplyRunError(err))
+			}
 		}
+
 		// Fetch the Generation from all Infos after they have been
 		// applied.
 		for _, inf := range infos {
@@ -140,7 +142,7 @@ func (a *ApplyTask) Start(taskContext *taskrunner.TaskContext) {
 				taskContext.ResourceApplied(id, uid, gen)
 			}
 		}
-		a.sendTaskResult(taskContext, nil)
+		a.sendTaskResult(taskContext)
 	}()
 }
 
@@ -187,10 +189,8 @@ func newApplyOptions(eventChannel chan event.Event, serverSideOptions common.Ser
 	}, nil
 }
 
-func (a *ApplyTask) sendTaskResult(taskContext *taskrunner.TaskContext, err error) {
-	taskContext.TaskChannel() <- taskrunner.TaskResult{
-		Err: err,
-	}
+func (a *ApplyTask) sendTaskResult(taskContext *taskrunner.TaskContext) {
+	taskContext.TaskChannel() <- taskrunner.TaskResult{}
 }
 
 // filterCRsWithCRDInSet loops through all the resources and filters out the
@@ -277,3 +277,25 @@ func buildCRDsInfo(crds []*unstructured.Unstructured) *crdsInfo {
 
 // ClearTimeout is not supported by the ApplyTask.
 func (a *ApplyTask) ClearTimeout() {}
+
+// createApplyEvent is a helper function to package an apply event for a single resource.
+func createApplyEvent(id object.ObjMetadata, operation event.ApplyEventOperation, err error) event.Event {
+	return event.Event{
+		Type: event.ApplyType,
+		ApplyEvent: event.ApplyEvent{
+			Type:       event.ApplyEventResourceUpdate,
+			Operation:  operation,
+			Identifier: id,
+			Error:      err,
+		},
+	}
+}
+
+// sendBatchApplyEvents is a helper function to send out multiple apply events for
+// a list of resources when failed to initialize the apply process.
+func sendBatchApplyEvents(taskContext *taskrunner.TaskContext, objects []*unstructured.Unstructured, err error) {
+	for _, obj := range objects {
+		taskContext.EventChannel() <- createApplyEvent(
+			object.UnstructuredToObjMeta(obj), event.Unchanged, applyerror.NewInitializeApplyOptionError(err))
+	}
+}