feat: rewrite inventory client with basic CRUD

This change rewrites the inventory client interface using simple CRUD
methods. Prior to this change the inventory client consisted of several
tightly coupled interfaces which made the client more complicated and
less flexible.

The old client also performed a Get of the inventory many times
throughout the applier/destroyer processes, including before every
update. This approach discarded optimistic concurrency protection since
the resourceVersion was never persisted. The new approach preserves an
internal interface representation of the inventory so that the
resourceVersion can be preserved across updates.

The new client interface is simpler to implement and is intended to
support implementations for sharded inventories, which may consist of
multiple underlying KRM objects. The prior interfaces made assumptions
of a single KRM object in several places, which made it incompatible
with a sharded inventory.

Author: sdowell

cmd/main.go

@@ -63,7 +63,7 @@ func main() {
 	}
 
 	loader := manifestreader.NewManifestLoader(f)
-	invFactory := inventory.ClusterClientFactory{StatusPolicy: inventory.StatusPolicyNone}
+	invFactory := inventory.ClusterClientFactory{}
 
 	names := []string{"init", "apply", "destroy", "diff", "preview", "status"}
 	subCmds := []*cobra.Command{

cmd/status/cmdstatus.go

@@ -164,11 +164,13 @@ func (r *Runner) loadInvFromDisk(cmd *cobra.Command, args []string) (*printer.Pr
 
 	// Based on the inventory template manifest we look up the inventory
 	// from the live state using the inventory client.
-	identifiers, err := invClient.GetClusterObjs(cmd.Context(), inv)
+	clusterInventory, err := invClient.Get(cmd.Context(), inv, inventory.GetOptions{})
 	if err != nil {
 		return nil, err
 	}
 
+	identifiers := clusterInventory.Objects()
+
 	printData := printer.PrintData{
 		Identifiers: object.ObjMetadataSet{},
 		InvNameMap:  make(map[object.ObjMetadata]string),
@@ -201,11 +203,16 @@ func (r *Runner) listInvFromCluster() (*printer.PrintData, error) {
 		StatusSet:   r.statusSet,
 	}
 
-	identifiersMap, err := invClient.ListClusterInventoryObjs(r.ctx)
+	inventories, err := invClient.List(r.ctx, inventory.ListOptions{})
 	if err != nil {
 		return nil, err
 	}
 
+	identifiersMap := make(map[string]object.ObjMetadataSet)
+	for _, inv := range inventories {
+		identifiersMap[inv.ID()] = inv.Objects()
+	}
+
 	for invName, identifiers := range identifiersMap {
 		// Check if there are targeted inventory names and include the current inventory name
 		if _, ok := r.inventoryNameSet[invName]; !ok && len(r.inventoryNameSet) != 0 {

pkg/apis/actuation/types.go

@@ -4,31 +4,9 @@
 package actuation
 
 import (
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/types"
 )
 
-// Inventory represents the inventory object in memory.
-// Inventory is currently only used for in-memory storage and not serialized to
-// disk or to the API server.
-type Inventory struct {
-	metav1.TypeMeta   `json:",inline"`
-	metav1.ObjectMeta `json:"metadata,omitempty"`
-
-	Spec   InventorySpec   `json:"spec,omitempty"`
-	Status InventoryStatus `json:"status,omitempty"`
-}
-
-// InventorySpec is the specification of the desired/expected inventory state.
-type InventorySpec struct {
-	Objects []ObjectReference `json:"objects,omitempty"`
-}
-
-// InventoryStatus is the status of the current/last-known inventory state.
-type InventoryStatus struct {
-	Objects []ObjectStatus `json:"objects,omitempty"`
-}
-
 // ObjectReference is a reference to a KRM resource by name and kind.
 //
 // Kubernetes only stores one API Version for each Kind at any given time,

pkg/apis/actuation/zz_generated.deepcopy.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"time"
 
+	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/util/sets"
@@ -48,23 +49,24 @@ type Applier struct {
 	openAPIGetter discovery.OpenAPISchemaInterface
 	mapper        meta.RESTMapper
 	infoHelper    info.Helper
+	statusPolicy  inventory.StatusPolicy
 }
 
 // prepareObjects returns the set of objects to apply and to prune or
 // an error if one occurred.
-func (a *Applier) prepareObjects(ctx context.Context, localInv inventory.Info, localObjs object.UnstructuredSet,
+func (a *Applier) prepareObjects(ctx context.Context, clusterInv inventory.Inventory, localObjs object.UnstructuredSet,
 	o ApplierOptions) (object.UnstructuredSet, object.UnstructuredSet, error) {
-	if localInv == nil {
+	if clusterInv == nil {
 		return nil, nil, fmt.Errorf("the local inventory can't be nil")
 	}
 	if err := inventory.ValidateNoInventory(localObjs); err != nil {
 		return nil, nil, err
 	}
 	// Add the inventory annotation to the resources being applied.
 	for _, localObj := range localObjs {
-		inventory.AddInventoryIDAnnotation(localObj, localInv)
+		inventory.AddInventoryIDAnnotation(localObj, clusterInv.ID())
 	}
-	pruneObjs, err := a.pruner.GetPruneObjs(ctx, localInv, localObjs, prune.Options{
+	pruneObjs, err := a.pruner.GetPruneObjs(ctx, clusterInv, localObjs, prune.Options{
 		DryRunStrategy: o.DryRunStrategy,
 	})
 	if err != nil {
@@ -96,8 +98,20 @@ func (a *Applier) Run(ctx context.Context, invInfo inventory.Info, objects objec
 		}
 		validator.Validate(objects)
 
+		clusterInventory, err := a.invClient.Get(ctx, invInfo, inventory.GetOptions{})
+		if apierrors.IsNotFound(err) {
+			clusterInventory = invInfo.InitialInventory()
+		} else if err != nil {
+			handleError(eventChannel, err)
+			return
+		}
+		if clusterInventory.ID() != invInfo.ID() {
+			handleError(eventChannel, fmt.Errorf("inventory-id of inventory object %s/%s in cluster doesn't match provided id %q",
+				invInfo.Namespace(), invInfo.Name(), invInfo.ID()))
+		}
+
 		// Decide which objects to apply and which to prune
-		applyObjs, pruneObjs, err := a.prepareObjects(ctx, invInfo, objects, options)
+		applyObjs, pruneObjs, err := a.prepareObjects(ctx, clusterInventory, objects, options)
 		if err != nil {
 			handleError(eventChannel, err)
 			return
@@ -154,11 +168,13 @@ func (a *Applier) Run(ctx context.Context, invInfo inventory.Info, objects objec
 			OpenAPIGetter: a.openAPIGetter,
 			InfoHelper:    a.infoHelper,
 			Mapper:        a.mapper,
+			Inventory:     clusterInventory,
 			InvClient:     a.invClient,
 			Collector:     vCollector,
 			ApplyFilters:  applyFilters,
 			ApplyMutators: applyMutators,
 			PruneFilters:  pruneFilters,
+			StatusPolicy:  a.statusPolicy,
 		}
 		opts := solver.Options{
 			ServerSideOptions:      options.ServerSideOptions,

pkg/apply/applier.go

@@ -44,6 +44,7 @@ func (b *ApplierBuilder) Build() (*Applier, error) {
 		openAPIGetter: bx.discoClient,
 		mapper:        bx.mapper,
 		infoHelper:    info.NewHelper(bx.mapper, bx.unstructuredClientForMapping),
+		statusPolicy:  bx.statusPolicy,
 	}, nil
 }
 
@@ -91,3 +92,8 @@ func (b *ApplierBuilder) WithStatusWatcherFilters(filters *watcher.Filters) *App
 	b.statusWatcherFilters = filters
 	return b
 }
+
+func (b *ApplierBuilder) WithStatusPolicy(statusPolicy inventory.StatusPolicy) *ApplierBuilder {
+	b.statusPolicy = statusPolicy
+	return b
+}

pkg/apply/applier_builder.go

@@ -1431,7 +1431,7 @@ func TestApplier(t *testing.T) {
 			testCtx, testCancel := context.WithTimeout(context.Background(), testTimeout)
 			defer testCancel() // cleanup
 
-			eventChannel := applier.Run(runCtx, tc.invInfo.toWrapped(), tc.resources, tc.options)
+			eventChannel := applier.Run(runCtx, tc.invInfo.toInfo(), tc.resources, tc.options)
 
 			// only start sending events once
 			var once sync.Once
@@ -1872,7 +1872,7 @@ func TestApplierCancel(t *testing.T) {
 			testCtx, testCancel := context.WithTimeout(context.Background(), tc.testTimeout)
 			defer testCancel() // cleanup
 
-			eventChannel := applier.Run(runCtx, tc.invInfo.toWrapped(), tc.resources, tc.options)
+			eventChannel := applier.Run(runCtx, tc.invInfo.toInfo(), tc.resources, tc.options)
 
 			// only start sending events once
 			var once sync.Once

pkg/apply/applier_test.go

@@ -28,6 +28,7 @@ type commonBuilder struct {
 	unstructuredClientForMapping func(*meta.RESTMapping) (resource.RESTClient, error)
 	statusWatcher                watcher.StatusWatcher
 	statusWatcherFilters         *watcher.Filters
+	statusPolicy                 inventory.StatusPolicy
 }
 
 func (cb *commonBuilder) finalize() (*commonBuilder, error) {

pkg/apply/builder.go

@@ -64,10 +64,14 @@ func (i inventoryInfo) toUnstructured() *unstructured.Unstructured {
 	}
 }
 
-func (i inventoryInfo) toWrapped() inventory.Info {
+func (i inventoryInfo) toInfo() inventory.Info {
 	return inventory.WrapInventoryInfoObj(i.toUnstructured())
 }
 
+func (i inventoryInfo) toWrapped() inventory.Inventory {
+	return inventory.WrapInventoryObj(i.toUnstructured())
+}
+
 func newTestApplier(
 	t *testing.T,
 	invInfo inventoryInfo,
@@ -125,7 +129,7 @@ func newTestInventory(
 ) inventory.Client {
 	// Use an Client with a fakeInfoHelper to allow generating Info
 	// objects that use the FakeRESTClient as the UnstructuredClient.
-	invClient, err := inventory.ClusterClientFactory{StatusPolicy: inventory.StatusPolicyAll}.NewClient(tf)
+	invClient, err := inventory.ClusterClientFactory{}.NewClient(tf)
 	require.NoError(t, err)
 	return invClient
 }

pkg/apply/common_test.go

@@ -8,6 +8,7 @@ import (
 	"fmt"
 	"time"
 
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/discovery"
@@ -38,6 +39,7 @@ type Destroyer struct {
 	client        dynamic.Interface
 	openAPIGetter discovery.OpenAPISchemaInterface
 	infoHelper    info.Helper
+	statusPolicy  inventory.StatusPolicy
 }
 
 type DestroyerOptions struct {
@@ -79,10 +81,22 @@ func (d *Destroyer) Run(ctx context.Context, invInfo inventory.Info, options Des
 	setDestroyerDefaults(&options)
 	go func() {
 		defer close(eventChannel)
+		clusterInventory, err := d.invClient.Get(ctx, invInfo, inventory.GetOptions{})
+		if apierrors.IsNotFound(err) {
+			clusterInventory = invInfo.InitialInventory()
+		} else if err != nil {
+			handleError(eventChannel, err)
+			return
+		}
+		if clusterInventory.ID() != invInfo.ID() {
+			handleError(eventChannel, fmt.Errorf("inventory-id of inventory object %s/%s in cluster doesn't match provided id %q",
+				invInfo.Namespace(), invInfo.Name(), invInfo.ID()))
+		}
+
 		// Retrieve the objects to be deleted from the cluster. Second parameter is empty
 		// because no local objects returns all inventory objects for deletion.
 		emptyLocalObjs := object.UnstructuredSet{}
-		deleteObjs, err := d.pruner.GetPruneObjs(ctx, invInfo, emptyLocalObjs, prune.Options{
+		deleteObjs, err := d.pruner.GetPruneObjs(ctx, clusterInventory, emptyLocalObjs, prune.Options{
 			DryRunStrategy: options.DryRunStrategy,
 		})
 		if err != nil {
@@ -123,8 +137,10 @@ func (d *Destroyer) Run(ctx context.Context, invInfo inventory.Info, options Des
 			InfoHelper:    d.infoHelper,
 			Mapper:        d.mapper,
 			InvClient:     d.invClient,
+			Inventory:     clusterInventory,
 			Collector:     vCollector,
 			PruneFilters:  deleteFilters,
+			StatusPolicy:  d.statusPolicy,
 		}
 		opts := solver.Options{
 			Destroy:                true,

pkg/apply/destroyer.go

@@ -44,6 +44,7 @@ func (b *DestroyerBuilder) Build() (*Destroyer, error) {
 		client:        bx.client,
 		openAPIGetter: bx.discoClient,
 		infoHelper:    info.NewHelper(bx.mapper, bx.unstructuredClientForMapping),
+		statusPolicy:  bx.statusPolicy,
 	}, nil
 }
 
@@ -91,3 +92,8 @@ func (b *DestroyerBuilder) WithStatusWatcherFilters(filters *watcher.Filters) *D
 	b.statusWatcherFilters = filters
 	return b
 }
+
+func (b *DestroyerBuilder) WithStatusPolicy(statusPolicy inventory.StatusPolicy) *DestroyerBuilder {
+	b.statusPolicy = statusPolicy
+	return b
+}

pkg/apply/destroyer_builder.go

@@ -11,7 +11,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 	"sigs.k8s.io/cli-utils/pkg/apply/event"
-	"sigs.k8s.io/cli-utils/pkg/inventory"
 	pollevent "sigs.k8s.io/cli-utils/pkg/kstatus/polling/event"
 	"sigs.k8s.io/cli-utils/pkg/kstatus/status"
 	"sigs.k8s.io/cli-utils/pkg/object"
@@ -313,12 +312,12 @@ func TestDestroyerCancel(t *testing.T) {
 		t.Run(tn, func(t *testing.T) {
 			statusWatcher := newFakeWatcher(tc.statusEvents)
 
-			invInfo := tc.invInfo.toWrapped()
+			invInfo := tc.invInfo.toInfo()
 
 			destroyer := newTestDestroyer(t,
 				tc.invInfo,
 				// Add the inventory to the cluster (to allow deletion)
-				append(tc.clusterObjs, inventory.InvInfoToConfigMap(invInfo)),
+				append(tc.clusterObjs, tc.invInfo.toUnstructured()),
 				statusWatcher,
 			)