capture the apply/reconcile status in the inventory object

Capture the apply status for individual resources after each apply. If the inventory object supports status field, it is updated after setting the inventory list at then of one apply process. BREAKING CHANGE: Update the inventory client and inventory interfaces to pass the apply/reconcile status.
kubernetes-sigs · Feb 2, 2022 · edee25b · edee25b
1 parent 5fb19a1
commit edee25b
Show file tree

Hide file tree

Showing 24 changed files with 285 additions and 31 deletions.
diff --git a/pkg/apply/event/actiongroupeventtype_string.go b/pkg/apply/event/actiongroupeventtype_string.go
diff --git a/pkg/apply/event/applyeventoperation_string.go b/pkg/apply/event/applyeventoperation_string.go
diff --git a/pkg/apply/event/deleteeventoperation_string.go b/pkg/apply/event/deleteeventoperation_string.go
diff --git a/pkg/apply/event/pruneeventoperation_string.go b/pkg/apply/event/pruneeventoperation_string.go
diff --git a/pkg/apply/event/resourceaction_string.go b/pkg/apply/event/resourceaction_string.go
diff --git a/pkg/apply/event/type_string.go b/pkg/apply/event/type_string.go
diff --git a/pkg/apply/event/waiteventoperation_string.go b/pkg/apply/event/waiteventoperation_string.go
diff --git a/pkg/apply/prune/prune_test.go b/pkg/apply/prune/prune_test.go
@@ -126,7 +126,7 @@ func createInventoryInfo(children ...*unstructured.Unstructured) inventory.Inven
 	inventoryObjCopy := inventoryObj.DeepCopy()
 	wrappedInv := inventory.WrapInventoryObj(inventoryObjCopy)
 	objs := object.UnstructuredSetToObjMetadataSet(children)
-	if err := wrappedInv.Store(objs); err != nil {
+	if err := wrappedInv.Store(objs, nil); err != nil {
 		return nil
 	}
 	obj, err := wrappedInv.GetObject()

diff --git a/pkg/apply/task/inv_set_task.go b/pkg/apply/task/inv_set_task.go
@@ -9,6 +9,7 @@ import (
 	"sigs.k8s.io/cli-utils/pkg/apply/taskrunner"
 	"sigs.k8s.io/cli-utils/pkg/common"
 	"sigs.k8s.io/cli-utils/pkg/inventory"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/status"
 	"sigs.k8s.io/cli-utils/pkg/object"
 )
 
@@ -111,8 +112,11 @@ func (i *InvSetTask) Start(taskContext *taskrunner.TaskContext) {
 		klog.V(4).Infof("keep in inventory %d invalid objects", len(invalidObjects))
 		invObjs = invObjs.Union(invalidObjects)
 
+		klog.V(4).Infof("get the apply status for %d objects", len(invObjs))
+		applyStatus := i.getApplyReoncileStatus(taskContext)
+
 		klog.V(4).Infof("set inventory %d total objects", len(invObjs))
-		err := i.InvClient.Replace(i.InvInfo, invObjs, i.DryRun)
+		err := i.InvClient.Replace(i.InvInfo, invObjs, i.DryRun, applyStatus)
 
 		klog.V(2).Infof("inventory set task completing (name: %q)", i.Name())
 		taskContext.TaskChannel() <- taskrunner.TaskResult{Err: err}
@@ -124,3 +128,41 @@ func (i *InvSetTask) Cancel(_ *taskrunner.TaskContext) {}
 
 // StatusUpdate is not supported by the InvSetTask.
 func (i *InvSetTask) StatusUpdate(_ *taskrunner.TaskContext, _ object.ObjMetadata) {}
+
+// getApplyReoncileStatus captures the apply status and reconcile status
+// for each individual resource in the inventory list.
+// The apply/reconcile status is then passed
+// to the inventory client and stored in the inventory object
+// through the Inventory interface.
+func (i InvSetTask) getApplyReoncileStatus(taskContext *taskrunner.TaskContext) map[object.ObjMetadata]status.ApplyReconcileStatus {
+	applyStatus := map[object.ObjMetadata]status.ApplyReconcileStatus{}
+	// capture the apply status
+	for _, obj := range taskContext.SuccessfulApplies() {
+		applyStatus[obj] = status.ApplyReconcileStatus{
+			ApplyStatus: status.ApplySucceeded,
+		}
+	}
+	for _, obj := range i.PrevInventory.Intersection(taskContext.FailedApplies()) {
+		applyStatus[obj] = status.ApplyReconcileStatus{
+			ApplyStatus: status.ApplyFailed,
+		}
+	}
+	for _, obj := range i.PrevInventory.Intersection(taskContext.SkippedApplies()) {
+		applyStatus[obj] = status.ApplyReconcileStatus{
+			ApplyStatus: status.ApplySkipped,
+		}
+	}
+	for _, obj := range i.PrevInventory.Intersection(taskContext.FailedDeletes()) {
+		applyStatus[obj] = status.ApplyReconcileStatus{
+			ApplyStatus: status.PruneFailed,
+		}
+	}
+	for _, obj := range i.PrevInventory.Intersection(taskContext.SkippedDeletes()) {
+		applyStatus[obj] = status.ApplyReconcileStatus{
+			ApplyStatus: status.PruneSkipped,
+		}
+	}
+
+	// TODO(Liujingfang1): capture the reconcile status
+	return applyStatus
+}
diff --git a/pkg/common/dryrunstrategy_string.go b/pkg/common/dryrunstrategy_string.go
diff --git a/pkg/inventory/fake-inventory-client.go b/pkg/inventory/fake-inventory-client.go
@@ -7,6 +7,7 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
 	"sigs.k8s.io/cli-utils/pkg/common"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/status"
 	"sigs.k8s.io/cli-utils/pkg/object"
 )
 
@@ -58,7 +59,8 @@ func (fic *FakeInventoryClient) Merge(_ InventoryInfo, objs object.ObjMetadataSe
 // Replace the stored cluster inventory objs with the passed obj, or an
 // error if one is set up.
 
-func (fic *FakeInventoryClient) Replace(_ InventoryInfo, objs object.ObjMetadataSet, _ common.DryRunStrategy) error {
+func (fic *FakeInventoryClient) Replace(_ InventoryInfo, objs object.ObjMetadataSet, _ common.DryRunStrategy,
+	_ map[object.ObjMetadata]status.ApplyReconcileStatus) error {
 	if fic.Err != nil {
 		return fic.Err
 	}

diff --git a/pkg/inventory/inventory-client.go b/pkg/inventory/inventory-client.go
@@ -16,6 +16,7 @@ import (
 	cmdutil "k8s.io/kubectl/pkg/cmd/util"
 	"k8s.io/kubectl/pkg/util"
 	"sigs.k8s.io/cli-utils/pkg/common"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/status"
 	"sigs.k8s.io/cli-utils/pkg/object"
 )
 
@@ -33,7 +34,7 @@ type InventoryClient interface {
 	Merge(inv InventoryInfo, objs object.ObjMetadataSet, dryRun common.DryRunStrategy) (object.ObjMetadataSet, error)
 	// Replace replaces the set of objects stored in the inventory
 	// object with the passed set of objects, or an error if one occurs.
-	Replace(inv InventoryInfo, objs object.ObjMetadataSet, dryRun common.DryRunStrategy) error
+	Replace(inv InventoryInfo, objs object.ObjMetadataSet, dryRun common.DryRunStrategy, applyStatus map[object.ObjMetadata]status.ApplyReconcileStatus) error
 	// DeleteInventoryObj deletes the passed inventory object from the APIServer.
 	DeleteInventoryObj(inv InventoryInfo, dryRun common.DryRunStrategy) error
 	// ApplyInventoryNamespace applies the Namespace that the inventory object should be in.
@@ -93,8 +94,9 @@ func (cic *ClusterInventoryClient) Merge(localInv InventoryInfo, objs object.Obj
 	}
 	if clusterInv == nil {
 		// Wrap inventory object and store the inventory in it.
+		applyStatus := getApplyStatus(nil, objs)
 		inv := cic.InventoryFactoryFunc(invObj)
-		if err := inv.Store(objs); err != nil {
+		if err := inv.Store(objs, applyStatus); err != nil {
 			return nil, err
 		}
 		invInfo, err := inv.GetObject()
@@ -105,25 +107,32 @@ func (cic *ClusterInventoryClient) Merge(localInv InventoryInfo, objs object.Obj
 		if err := cic.createInventoryObj(invInfo, dryRun); err != nil {
 			return nil, err
 		}
+		if err := cic.updateStatus(invInfo, dryRun); err != nil {
+			return nil, err
+		}
 	} else {
 		// Update existing cluster inventory with merged union of objects
 		clusterObjs, err := cic.GetClusterObjs(localInv)
 		if err != nil {
 			return pruneIds, err
 		}
-		if objs.Equal(clusterObjs) {
-			klog.V(4).Infof("applied objects same as cluster inventory: do nothing")
-			return pruneIds, nil
-		}
 		pruneIds = clusterObjs.Diff(objs)
 		unionObjs := clusterObjs.Union(objs)
+		applyStatus := getApplyStatus(pruneIds, unionObjs)
 		klog.V(4).Infof("num objects to prune: %d", len(pruneIds))
 		klog.V(4).Infof("num merged objects to store in inventory: %d", len(unionObjs))
 		wrappedInv := cic.InventoryFactoryFunc(clusterInv)
-		if err = wrappedInv.Store(unionObjs); err != nil {
+		if err = wrappedInv.Store(unionObjs, applyStatus); err != nil {
 			return pruneIds, err
 		}
-		if !dryRun.ClientOrServerDryRun() {
+		clusterInv, err = wrappedInv.GetObject()
+		if err != nil {
+			return pruneIds, err
+		}
+		if dryRun.ClientOrServerDryRun() {
+			return pruneIds, nil
+		}
+		if !objs.Equal(clusterObjs) {
 			clusterInv, err = wrappedInv.GetObject()
 			if err != nil {
 				return pruneIds, err
@@ -133,14 +142,18 @@ func (cic *ClusterInventoryClient) Merge(localInv InventoryInfo, objs object.Obj
 				return pruneIds, err
 			}
 		}
+		if err := cic.updateStatus(clusterInv, dryRun); err != nil {
+			return pruneIds, err
+		}
 	}
 
 	return pruneIds, nil
 }
 
 // Replace stores the passed objects in the cluster inventory object, or
 // an error if one occurred.
-func (cic *ClusterInventoryClient) Replace(localInv InventoryInfo, objs object.ObjMetadataSet, dryRun common.DryRunStrategy) error {
+func (cic *ClusterInventoryClient) Replace(localInv InventoryInfo, objs object.ObjMetadataSet, dryRun common.DryRunStrategy,
+	applyStatus map[object.ObjMetadata]status.ApplyReconcileStatus) error {
 	// Skip entire function for dry-run.
 	if dryRun.ClientOrServerDryRun() {
 		klog.V(4).Infoln("dry-run replace inventory object: not applied")
@@ -150,30 +163,32 @@ func (cic *ClusterInventoryClient) Replace(localInv InventoryInfo, objs object.O
 	if err != nil {
 		return fmt.Errorf("failed to read inventory objects from cluster: %w", err)
 	}
-	if objs.Equal(clusterObjs) {
-		klog.V(4).Infof("applied objects same as cluster inventory: do nothing")
-		return nil
-	}
 	clusterInv, err := cic.GetClusterInventoryInfo(localInv)
 	if err != nil {
 		return fmt.Errorf("failed to read inventory from cluster: %w", err)
 	}
-	clusterInv, err = cic.replaceInventory(clusterInv, objs)
+	clusterInv, err = cic.replaceInventory(clusterInv, objs, applyStatus)
 	if err != nil {
 		return err
 	}
-	klog.V(4).Infof("replace cluster inventory: %s/%s", clusterInv.GetNamespace(), clusterInv.GetName())
-	klog.V(4).Infof("replace cluster inventory %d objects", len(objs))
-	if err := cic.applyInventoryObj(clusterInv, dryRun); err != nil {
-		return fmt.Errorf("failed to write updated inventory to cluster: %w", err)
+	if !objs.Equal(clusterObjs) {
+		klog.V(4).Infof("replace cluster inventory: %s/%s", clusterInv.GetNamespace(), clusterInv.GetName())
+		klog.V(4).Infof("replace cluster inventory %d objects", len(objs))
+		if err := cic.applyInventoryObj(clusterInv, dryRun); err != nil {
+			return fmt.Errorf("failed to write updated inventory to cluster: %w", err)
+		}
+	}
+	if err := cic.updateStatus(clusterInv, dryRun); err != nil {
+		return err
 	}
 	return nil
 }
 
 // replaceInventory stores the passed objects into the passed inventory object.
-func (cic *ClusterInventoryClient) replaceInventory(inv *unstructured.Unstructured, objs object.ObjMetadataSet) (*unstructured.Unstructured, error) {
+func (cic *ClusterInventoryClient) replaceInventory(inv *unstructured.Unstructured, objs object.ObjMetadataSet,
+	applyStatus map[object.ObjMetadata]status.ApplyReconcileStatus) (*unstructured.Unstructured, error) {
 	wrappedInv := cic.InventoryFactoryFunc(inv)
-	if err := wrappedInv.Store(objs); err != nil {
+	if err := wrappedInv.Store(objs, applyStatus); err != nil {
 		return nil, err
 	}
 	clusterInv, err := wrappedInv.GetObject()
@@ -421,3 +436,58 @@ func (cic *ClusterInventoryClient) ApplyInventoryNamespace(obj *unstructured.Uns
 func (cic *ClusterInventoryClient) getMapping(obj *unstructured.Unstructured) (*meta.RESTMapping, error) {
 	return cic.mapper.RESTMapping(obj.GroupVersionKind().GroupKind(), obj.GroupVersionKind().Version)
 }
+
+func (cic *ClusterInventoryClient) updateStatus(obj *unstructured.Unstructured, dryRun common.DryRunStrategy) error {
+	if dryRun.ClientOrServerDryRun() {
+		klog.V(4).Infof("dry-run update inventory status: not updated")
+		return nil
+	}
+	status, found, _ := unstructured.NestedMap(obj.UnstructuredContent(), "status")
+	if !found {
+		return nil
+	}
+
+	klog.V(4).Infof("update inventory status")
+	mapping, err := cic.mapper.RESTMapping(obj.GroupVersionKind().GroupKind())
+	if err != nil {
+		return err
+	}
+	resource := cic.dc.Resource(mapping.Resource).Namespace(obj.GetNamespace())
+
+	liveObj, err := resource.Get(context.TODO(), obj.GetName(), metav1.GetOptions{TypeMeta: metav1.TypeMeta{
+		Kind:       obj.GetKind(),
+		APIVersion: obj.GetAPIVersion(),
+	}})
+	if err != nil {
+		return err
+	}
+
+	err = unstructured.SetNestedMap(liveObj.UnstructuredContent(), status, "status")
+	if err != nil {
+		return err
+	}
+	_, err = resource.UpdateStatus(context.TODO(), liveObj, metav1.UpdateOptions{TypeMeta: metav1.TypeMeta{
+		Kind:       obj.GetKind(),
+		APIVersion: obj.GetAPIVersion(),
+	}})
+	if err != nil {
+		klog.V(4).Infof("failed to update inventory status: %v", err)
+	}
+	// Don't exit the apply process if failed to update the inventory object status.
+	return nil
+}
+
+func getApplyStatus(pruneIds, unionIds []object.ObjMetadata) map[object.ObjMetadata]status.ApplyReconcileStatus {
+	applyStatus := map[object.ObjMetadata]status.ApplyReconcileStatus{}
+	for _, obj := range unionIds {
+		applyStatus[obj] = status.ApplyReconcileStatus{
+			ApplyStatus: status.ApplyPending,
+		}
+	}
+	for _, obj := range pruneIds {
+		applyStatus[obj] = status.ApplyReconcileStatus{
+			ApplyStatus: status.PrunePending,
+		}
+	}
+	return applyStatus
+}
diff --git a/pkg/inventory/inventory-client_test.go b/pkg/inventory/inventory-client_test.go
@@ -291,11 +291,11 @@ func TestReplace(t *testing.T) {
 	// Client and server dry-run do not throw errors.
 	invClient, err := NewInventoryClient(tf, WrapInventoryObj, InvInfoToConfigMap)
 	require.NoError(t, err)
-	err = invClient.Replace(copyInventory(), object.ObjMetadataSet{}, common.DryRunClient)
+	err = invClient.Replace(copyInventory(), object.ObjMetadataSet{}, common.DryRunClient, nil)
 	if err != nil {
 		t.Fatalf("unexpected error received: %s", err)
 	}
-	err = invClient.Replace(copyInventory(), object.ObjMetadataSet{}, common.DryRunServer)
+	err = invClient.Replace(copyInventory(), object.ObjMetadataSet{}, common.DryRunServer, nil)
 	if err != nil {
 		t.Fatalf("unexpected error received: %s", err)
 	}
@@ -307,15 +307,15 @@ func TestReplace(t *testing.T) {
 				WrapInventoryObj, InvInfoToConfigMap)
 			require.NoError(t, err)
 			wrappedInv := invClient.InventoryFactoryFunc(inventoryObj)
-			if err := wrappedInv.Store(tc.clusterObjs); err != nil {
+			if err := wrappedInv.Store(tc.clusterObjs, nil); err != nil {
 				t.Fatalf("unexpected error storing inventory objects: %s", err)
 			}
 			inv, err := wrappedInv.GetObject()
 			if err != nil {
 				t.Fatalf("unexpected error storing inventory objects: %s", err)
 			}
 			// Call replaceInventory with the new set of "localObjs"
-			inv, err = invClient.replaceInventory(inv, tc.localObjs)
+			inv, err = invClient.replaceInventory(inv, tc.localObjs, nil)
 			if err != nil {
 				t.Fatalf("unexpected error received: %s", err)
 			}

diff --git a/pkg/inventory/inventory.go b/pkg/inventory/inventory.go
@@ -18,6 +18,7 @@ import (
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/cli-utils/pkg/common"
+	"sigs.k8s.io/cli-utils/pkg/kstatus/status"
 	"sigs.k8s.io/cli-utils/pkg/object"
 )
 
@@ -31,7 +32,7 @@ type Inventory interface {
 	// Load retrieves the set of object metadata from the inventory object
 	Load() (object.ObjMetadataSet, error)
 	// Store the set of object metadata in the inventory object
-	Store(objs object.ObjMetadataSet) error
+	Store(objs object.ObjMetadataSet, applyStatus map[object.ObjMetadata]status.ApplyReconcileStatus) error
 	// GetObject returns the object that stores the inventory
 	GetObject() (*unstructured.Unstructured, error)
 }

diff --git a/pkg/inventory/inventory_test.go b/pkg/inventory/inventory_test.go
@@ -423,7 +423,7 @@ func copyInventory() InventoryInfo {
 
 func storeObjsInInventory(info InventoryInfo, objs object.ObjMetadataSet) *unstructured.Unstructured {
 	wrapped := WrapInventoryObj(InvInfoToConfigMap(info))
-	_ = wrapped.Store(objs)
+	_ = wrapped.Store(objs, nil)
 	inv, _ := wrapped.GetObject()
 	return inv
 }