Enable custom inventory storage logic

This commit introduces 2 new Storage interface methods to enable clients
to implement their own logic for applying inventory objects to the live
cluster.

Author: rquitales

pkg/inventory/inventory-client.go

@@ -103,6 +103,8 @@ func (cic *ClusterClient) Merge(localInv Info, objs object.ObjMetadataSet, dryRu
 	if err != nil {
 		return pruneIds, err
 	}
+
+	// Inventory does not exist on the cluster.
 	if clusterInv == nil {
 		// Wrap inventory object and store the inventory in it.
 		var status []actuation.ObjectStatus
@@ -113,60 +115,41 @@ func (cic *ClusterClient) Merge(localInv Info, objs object.ObjMetadataSet, dryRu
 		if err := inv.Store(objs, status); err != nil {
 			return nil, err
 		}
-		invInfo, err := inv.GetObject()
-		if err != nil {
-			return nil, err
-		}
 		klog.V(4).Infof("creating initial inventory object with %d objects", len(objs))
-		createdObj, err := cic.createInventoryObj(invInfo, dryRun)
-		if err != nil {
-			return nil, err
-		}
-		// Status update requires the latest ResourceVersion
-		invInfo.SetResourceVersion(createdObj.GetResourceVersion())
-		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
-		}
-		pruneIds = clusterObjs.Diff(objs)
-		unionObjs := clusterObjs.Union(objs)
-		var status []actuation.ObjectStatus
-		if cic.statusPolicy == StatusPolicyAll {
-			status = getObjStatus(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, status); err != nil {
-			return pruneIds, err
-		}
-		clusterInv, err = wrappedInv.GetObject()
-		if err != nil {
-			return pruneIds, err
-		}
+
 		if dryRun.ClientOrServerDryRun() {
-			return pruneIds, nil
-		}
-		if !objs.Equal(clusterObjs) {
-			klog.V(4).Infof("update cluster inventory: %s/%s", clusterInv.GetNamespace(), clusterInv.GetName())
-			appliedObj, err := cic.applyInventoryObj(clusterInv, dryRun)
-			if err != nil {
-				return pruneIds, err
-			}
-			// Status update requires the latest ResourceVersion
-			clusterInv.SetResourceVersion(appliedObj.GetResourceVersion())
-		}
-		if err := cic.updateStatus(clusterInv, dryRun); err != nil {
-			return pruneIds, err
+			klog.V(4).Infof("dry-run create inventory object: not created")
+			return nil, nil
 		}
+
+		err = inv.Apply(cic.dc, cic.mapper, cic.statusPolicy)
+		return nil, err
+	}
+
+	// Update existing cluster inventory with merged union of objects
+	clusterObjs, err := cic.GetClusterObjs(localInv)
+	if err != nil {
+		return pruneIds, err
+	}
+	pruneIds = clusterObjs.Diff(objs)
+	unionObjs := clusterObjs.Union(objs)
+	var status []actuation.ObjectStatus
+	if cic.statusPolicy == StatusPolicyAll {
+		status = getObjStatus(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, status); err != nil {
+		return pruneIds, err
 	}
 
-	return pruneIds, nil
+	if dryRun.ClientOrServerDryRun() {
+		klog.V(4).Infof("dry-run create inventory object: not created")
+		return pruneIds, nil
+	}
+	err = wrappedInv.Apply(cic.dc, cic.mapper, cic.statusPolicy)
+	return pruneIds, err
 }
 
 // Replace stores the passed objects in the cluster inventory object, or
@@ -178,49 +161,49 @@ func (cic *ClusterClient) Replace(localInv Info, objs object.ObjMetadataSet, sta
 		klog.V(4).Infoln("dry-run replace inventory object: not applied")
 		return nil
 	}
-	clusterObjs, err := cic.GetClusterObjs(localInv)
-	if err != nil {
-		return fmt.Errorf("failed to read inventory objects from cluster: %w", err)
-	}
 	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, status)
+
+	clusterObjs, err := cic.GetClusterObjs(localInv)
+	if err != nil {
+		return fmt.Errorf("failed to read inventory objects from cluster: %w", err)
+	}
+
+	clusterInv, wrappedInv, err := cic.replaceInventory(clusterInv, objs, status)
 	if err != nil {
 		return 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))
-		appliedObj, err := cic.applyInventoryObj(clusterInv, dryRun)
-		if err != nil {
+
+		if err := wrappedInv.ApplyWithPrune(cic.dc, cic.mapper, cic.statusPolicy, objs); err != nil {
 			return fmt.Errorf("failed to write updated inventory to cluster: %w", err)
 		}
-		// Status update requires the latest ResourceVersion
-		clusterInv.SetResourceVersion(appliedObj.GetResourceVersion())
-	}
-	if err := cic.updateStatus(clusterInv, dryRun); err != nil {
-		return err
 	}
+
 	return nil
 }
 
 // replaceInventory stores the passed objects into the passed inventory object.
 func (cic *ClusterClient) replaceInventory(inv *unstructured.Unstructured, objs object.ObjMetadataSet,
-	status []actuation.ObjectStatus) (*unstructured.Unstructured, error) {
+	status []actuation.ObjectStatus) (*unstructured.Unstructured, Storage, error) {
 	if cic.statusPolicy == StatusPolicyNone {
 		status = nil
 	}
 	wrappedInv := cic.InventoryFactoryFunc(inv)
 	if err := wrappedInv.Store(objs, status); err != nil {
-		return nil, err
+		return nil, nil, err
 	}
 	clusterInv, err := wrappedInv.GetObject()
 	if err != nil {
-		return nil, err
+		return nil, nil, err
 	}
-	return clusterInv, nil
+
+	return clusterInv, wrappedInv, nil
 }
 
 // DeleteInventoryObj deletes the inventory object from the cluster.
@@ -364,26 +347,6 @@ func (cic *ClusterClient) GetClusterInventoryObjs(inv Info) (object.Unstructured
 	return clusterInvObjects, err
 }
 
-// applyInventoryObj applies the passed inventory object to the APIServer.
-func (cic *ClusterClient) applyInventoryObj(obj *unstructured.Unstructured, dryRun common.DryRunStrategy) (*unstructured.Unstructured, error) {
-	if dryRun.ClientOrServerDryRun() {
-		klog.V(4).Infof("dry-run apply inventory object: not applied")
-		return obj.DeepCopy(), nil
-	}
-	if obj == nil {
-		return nil, fmt.Errorf("attempting apply a nil inventory object")
-	}
-
-	mapping, err := cic.getMapping(obj)
-	if err != nil {
-		return nil, err
-	}
-
-	klog.V(4).Infof("replacing inventory object: %s/%s", obj.GetNamespace(), obj.GetName())
-	return cic.dc.Resource(mapping.Resource).Namespace(obj.GetNamespace()).
-		Update(context.TODO(), obj, metav1.UpdateOptions{})
-}
-
 // createInventoryObj creates the passed inventory object on the APIServer.
 func (cic *ClusterClient) createInventoryObj(obj *unstructured.Unstructured, dryRun common.DryRunStrategy) (*unstructured.Unstructured, error) {
 	if dryRun.ClientOrServerDryRun() {
@@ -463,62 +426,6 @@ func (cic *ClusterClient) getMapping(obj *unstructured.Unstructured) (*meta.REST
 	return cic.mapper.RESTMapping(obj.GroupVersionKind().GroupKind(), obj.GroupVersionKind().Version)
 }
 
-func (cic *ClusterClient) updateStatus(obj *unstructured.Unstructured, dryRun common.DryRunStrategy) error {
-	if cic.statusPolicy != StatusPolicyAll {
-		klog.V(4).Infof("inventory status update skipped (StatusPolicy: %s)", cic.statusPolicy)
-		return nil
-	}
-	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
-	}
-	mapping, err := cic.mapper.RESTMapping(obj.GroupVersionKind().GroupKind())
-	if err != nil {
-		return nil
-	}
-	hasStatus, err := cic.hasSubResource(obj.GetAPIVersion(), mapping.Resource.Resource, "status")
-	if err != nil {
-		return err
-	}
-	if !hasStatus {
-		klog.V(4).Infof("skip updating inventory status")
-		return nil
-	}
-
-	klog.V(4).Infof("update inventory status")
-	resource := cic.dc.Resource(mapping.Resource).Namespace(obj.GetNamespace())
-	meta := metav1.TypeMeta{
-		Kind:       obj.GetKind(),
-		APIVersion: obj.GetAPIVersion(),
-	}
-	if err = unstructured.SetNestedMap(obj.Object, status, "status"); err != nil {
-		return err
-	}
-	if _, err = resource.UpdateStatus(context.TODO(), obj, metav1.UpdateOptions{TypeMeta: meta}); err != nil {
-		return fmt.Errorf("failed to write updated inventory status to cluster: %w", err)
-	}
-	return nil
-}
-
-// hasSubResource checks if a resource has the given subresource using the discovery client.
-func (cic *ClusterClient) hasSubResource(groupVersion, resource, subresource string) (bool, error) {
-	resources, err := cic.discoveryClient.ServerResourcesForGroupVersion(groupVersion)
-	if err != nil {
-		return false, err
-	}
-
-	for _, r := range resources.APIResources {
-		if r.Name == fmt.Sprintf("%s/%s", resource, subresource) {
-			return true, nil
-		}
-	}
-	return false, nil
-}
-
 // getObjStatus returns the list of object status
 // at the beginning of an apply process.
 func getObjStatus(pruneIds, unionIds []object.ObjMetadata) []actuation.ObjectStatus {

pkg/inventory/inventory-client_test.go

@@ -393,7 +393,7 @@ func TestReplace(t *testing.T) {
 				t.Fatalf("unexpected error storing inventory objects: %s", err)
 			}
 			// Call replaceInventory with the new set of "localObjs"
-			inv, err = invClient.replaceInventory(inv, tc.localObjs, tc.objStatus)
+			inv, _, err = invClient.replaceInventory(inv, tc.localObjs, tc.objStatus)
 			if err != nil {
 				t.Fatalf("unexpected error received: %s", err)
 			}

pkg/inventory/inventorycm.go

@@ -9,10 +9,16 @@
 package inventory
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/client-go/dynamic"
+	"k8s.io/klog/v2"
 	"sigs.k8s.io/cli-utils/pkg/apis/actuation"
 	"sigs.k8s.io/cli-utils/pkg/common"
 	"sigs.k8s.io/cli-utils/pkg/object"
@@ -119,6 +125,69 @@ func (icm *ConfigMap) GetObject() (*unstructured.Unstructured, error) {
 	return invCopy, nil
 }
 
+// Apply is an Storage interface function implemented to apply the inventory
+// object. StatusPolicy is not needed since ConfigMaps do not have a status subresource.
+func (icm *ConfigMap) Apply(dc dynamic.Interface, mapper meta.RESTMapper, _ StatusPolicy) error {
+	invInfo, namespacedClient, err := icm.getNamespacedClient(dc, mapper)
+	if err != nil {
+		return err
+	}
+
+	// Get cluster object, if exsists.
+	clusterObj, err := namespacedClient.Get(context.TODO(), invInfo.GetName(), metav1.GetOptions{})
+	if err != nil && !apierrors.IsNotFound(err) {
+		return err
+	}
+
+	// Create cluster inventory object, if it does not exist on cluster.
+	if clusterObj == nil {
+		klog.V(4).Infof("creating inventory object: %s/%s", invInfo.GetNamespace(), invInfo.GetName())
+		_, err = namespacedClient.Create(context.TODO(), invInfo, metav1.CreateOptions{})
+		return err
+	}
+
+	// Update the cluster inventory object instead.
+	klog.V(4).Infof("updating inventory object: %s/%s", invInfo.GetNamespace(), invInfo.GetName())
+	_, err = namespacedClient.Update(context.TODO(), invInfo, metav1.UpdateOptions{})
+	return err
+}
+
+// ApplyWithPrune is a Storage interface function implemented to apply the inventory object with a list of objects
+// to be pruned. StatusPolicy is not needed since ConfigMaps do not have a status subresource.
+func (icm *ConfigMap) ApplyWithPrune(dc dynamic.Interface, mapper meta.RESTMapper, _ StatusPolicy, _ object.ObjMetadataSet) error {
+	invInfo, namespacedClient, err := icm.getNamespacedClient(dc, mapper)
+	if err != nil {
+		return err
+	}
+
+	// Update the cluster inventory object.
+	klog.V(4).Infof("updating inventory object: %s/%s", invInfo.GetNamespace(), invInfo.GetName())
+	_, err = namespacedClient.Update(context.TODO(), invInfo, metav1.UpdateOptions{})
+	return err
+}
+
+// getNamespacedClient is a helper function for Apply and ApplyWithPrune that creates a namespaced client for interacting with the live
+// cluster, as well as returning the ConfigMap object as a wrapped resource.Info object.
+func (icm *ConfigMap) getNamespacedClient(dc dynamic.Interface, mapper meta.RESTMapper) (*unstructured.Unstructured, dynamic.ResourceInterface, error) {
+	invInfo, err := icm.GetObject()
+	if err != nil {
+		return nil, nil, err
+	}
+	if invInfo == nil {
+		return nil, nil, fmt.Errorf("attempting to create a nil inventory object")
+	}
+
+	mapping, err := mapper.RESTMapping(invInfo.GroupVersionKind().GroupKind(), invInfo.GroupVersionKind().Version)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	// Create client to interact with cluster.
+	namespacedClient := dc.Resource(mapping.Resource).Namespace(invInfo.GetNamespace())
+
+	return invInfo, namespacedClient, nil
+}
+
 func buildObjMap(objMetas object.ObjMetadataSet, objStatus []actuation.ObjectStatus) map[string]string {
 	objMap := map[string]string{}
 	objStatusMap := map[object.ObjMetadata]actuation.ObjectStatus{}

pkg/inventory/inventory.go renamed to pkg/inventory/storage.go

@@ -15,7 +15,9 @@ import (
 	"fmt"
 	"strings"
 
+	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
+	"k8s.io/client-go/dynamic"
 	"k8s.io/klog/v2"
 	"sigs.k8s.io/cli-utils/pkg/apis/actuation"
 	"sigs.k8s.io/cli-utils/pkg/common"
@@ -31,10 +33,19 @@ const legacyInvName = "inventory"
 type Storage 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 the set of object metadata in the inventory object. This will
+	// replace the metadata, spec and status.
 	Store(objs object.ObjMetadataSet, status []actuation.ObjectStatus) error
 	// GetObject returns the object that stores the inventory
 	GetObject() (*unstructured.Unstructured, error)
+	// Apply applies the inventory object. This utility function is used
+	// in InventoryClient.Merge and merges the metadata, spec and status.
+	Apply(dynamic.Interface, meta.RESTMapper, StatusPolicy) error
+	// ApplyWithPrune applies the inventory object with a set of pruneIDs of
+	// objects to be pruned (object.ObjMetadataSet). This function is used in
+	// InventoryClient.Replace. pruneIDs are required for enabling custom logic
+	// handling of multiple ResourceGroup inventories.
+	ApplyWithPrune(dynamic.Interface, meta.RESTMapper, StatusPolicy, object.ObjMetadataSet) error
 }
 
 // StorageFactoryFunc creates the object which implements the Inventory