Implement caching for download of virtual machine images (vmware-tanzu#47)

This patch introduces an in-memory cache to store and manage downloaded OVF for virtual machine images. It also cleans up stale cached items after a specified time (currently set to 30 minutes). This avoids unnecessary requests to vCenter when the same content library is associated with multiple namespaces within a cluster and speeds up vm image reconciliation time.

Testing Done:
- Internal kind-based e2e tests (both fast and full) passed
- Manually deployed this patch in a WCP testbed (with Image-Registry FSS enabled) and verified the expected workflow by associating a content library to multiple namespaces simultaneously. Logs from VM-Operator (in verbose mode):
```
I0207 09:15:21.266601       1 vmprovider.go:248] vsphere "msg"="Cache item miss, downloading OVF from vCenter" "contentVersion"="2" "itemID"="732f3e46-4699-4b66-aa6c-2740b0bfe6ed"
...
I0207 09:15:29.134953       1 vmprovider.go:264] vsphere "msg"="Cache item put" "contentVersion"="2" "itemID"="732f3e46-4699-4b66-aa6c-2740b0bfe6ed" "putResult"=1
I0207 09:15:29.135280       1 vmprovider.go:246] vsphere "msg"="Cache item hit, using cached OVF" "contentVersion"="2" "itemID"="732f3e46-4699-4b66-aa6c-2740b0bfe6ed"
I0207 09:15:29.136055       1 vmprovider.go:246] vsphere "msg"="Cache item hit, using cached OVF" "contentVersion"="2" "itemID"="732f3e46-4699-4b66-aa6c-2740b0bfe6ed"
```

Author: dilyar85

controllers/contentlibrary/clustercontentlibraryitem/clustercontentlibraryitem_controller.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2022 VMware, Inc. All Rights Reserved.
+// Copyright (c) 2022-2023 VMware, Inc. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
 package clustercontentlibraryitem
@@ -260,7 +260,7 @@ func (r *Reconciler) syncImageContent(ctx goctx.Context,
 		return nil
 	}
 
-	err := r.VMProvider.SyncVirtualMachineImage(ctx, cclItem.Spec.UUID, cvmi)
+	err := r.VMProvider.SyncVirtualMachineImage(ctx, cclItem, cvmi)
 	if err != nil {
 		conditions.MarkFalse(cvmi,
 			vmopv1a1.VirtualMachineImageSyncedCondition,

controllers/contentlibrary/clustercontentlibraryitem/clustercontentlibraryitem_controller_intg_test.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2022 VMware, Inc. All Rights Reserved.
+// Copyright (c) 2022-2023 VMware, Inc. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
 package clustercontentlibraryitem_test
@@ -65,8 +65,8 @@ func cclItemReconcile() {
 		ctx = suite.NewIntegrationTestContext()
 		intgFakeVMProvider.Lock()
 		defer intgFakeVMProvider.Unlock()
-		intgFakeVMProvider.SyncVirtualMachineImageFn = func(ctx context.Context,
-			itemID string, cvmiObj client.Object) error {
+		intgFakeVMProvider.SyncVirtualMachineImageFn = func(
+			_ context.Context, _, cvmiObj client.Object) error {
 			// Change a random spec and status field to verify the provider function is called.
 			cvmi := cvmiObj.(*vmopv1a1.ClusterVirtualMachineImage)
 			cvmi.Spec.HardwareVersion = 123

controllers/contentlibrary/clustercontentlibraryitem/clustercontentlibraryitem_controller_unit_test.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2022 VMware, Inc. All Rights Reserved.
+// Copyright (c) 2022-2023 VMware, Inc. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
 package clustercontentlibraryitem_test
@@ -56,8 +56,8 @@ func unitTestsReconcile() {
 			ctx.VMProvider,
 		)
 		fakeVMProvider = ctx.VMProvider.(*providerfake.VMProvider)
-		fakeVMProvider.SyncVirtualMachineImageFn = func(_ context.Context, _ string,
-			cvmiObj client.Object) error {
+		fakeVMProvider.SyncVirtualMachineImageFn = func(
+			_ context.Context, _, cvmiObj client.Object) error {
 			cvmi := cvmiObj.(*vmopv1a1.ClusterVirtualMachineImage)
 			// Change a random spec and status field to verify the provider function is called.
 			cvmi.Spec.HardwareVersion = 123

controllers/contentlibrary/contentlibraryitem/contentlibraryitem_controller.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2022 VMware, Inc. All Rights Reserved.
+// Copyright (c) 2022-2023 VMware, Inc. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
 package contentlibraryitem
@@ -255,10 +255,7 @@ func (r *Reconciler) syncImageContent(ctx goctx.Context,
 		return nil
 	}
 
-	// TODO: .
-	// We should use cache here to avoid downloading from VC for every single VM image under every namespaces and
-	// cluster VMI with the same item ID.
-	err := r.VMProvider.SyncVirtualMachineImage(ctx, clItem.Spec.UUID, vmi)
+	err := r.VMProvider.SyncVirtualMachineImage(ctx, clItem, vmi)
 	if err != nil {
 		conditions.MarkFalse(vmi,
 			vmopv1alpha1.VirtualMachineImageSyncedCondition,

controllers/contentlibrary/contentlibraryitem/contentlibraryitem_controller_intg_test.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2022 VMware, Inc. All Rights Reserved.
+// Copyright (c) 2022-2023 VMware, Inc. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
 package contentlibraryitem_test
@@ -60,8 +60,8 @@ func clItemReconcile() {
 
 		intgFakeVMProvider.Lock()
 		defer intgFakeVMProvider.Unlock()
-		intgFakeVMProvider.SyncVirtualMachineImageFn = func(ctx context.Context,
-			itemID string, vmiObj client.Object) error {
+		intgFakeVMProvider.SyncVirtualMachineImageFn = func(
+			_ context.Context, _, vmiObj client.Object) error {
 			vmi := vmiObj.(*vmopv1a1.VirtualMachineImage)
 			// Change a random spec and status field to verify the provider function is called.
 			vmi.Spec.HardwareVersion = 123

controllers/contentlibrary/contentlibraryitem/contentlibraryitem_controller_unit_test.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2022 VMware, Inc. All Rights Reserved.
+// Copyright (c) 2022-2023 VMware, Inc. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
 package contentlibraryitem_test
@@ -55,8 +55,7 @@ func unitTestsReconcile() {
 			ctx.VMProvider,
 		)
 		fakeVMProvider = ctx.VMProvider.(*providerfake.VMProvider)
-		fakeVMProvider.SyncVirtualMachineImageFn = func(_ context.Context, _ string,
-			vmi client.Object) error {
+		fakeVMProvider.SyncVirtualMachineImageFn = func(_ context.Context, _, vmi client.Object) error {
 			vmiObj := vmi.(*vmopv1a1.VirtualMachineImage)
 			// Change a random spec and status field to verify the provider function is called.
 			vmiObj.Spec.HardwareVersion = 123

controllers/contentlibrary/utils/test_utils.go

@@ -1,4 +1,4 @@
-// Copyright (c) 2022 VMware, Inc. All Rights Reserved.
+// Copyright (c) 2022-2023 VMware, Inc. All Rights Reserved.
 // SPDX-License-Identifier: Apache-2.0
 
 package utils
@@ -105,7 +105,7 @@ func GetServiceTypeLabels(labels map[string]string) map[string]string {
 }
 
 func GetExpectedCVMIFrom(cclItem imgregv1a1.ClusterContentLibraryItem,
-	providerFunc func(context.Context, string, crtlclient.Object) error) *vmopv1a1.ClusterVirtualMachineImage {
+	providerFunc func(context.Context, crtlclient.Object, crtlclient.Object) error) *vmopv1a1.ClusterVirtualMachineImage {
 
 	cvmi := &vmopv1a1.ClusterVirtualMachineImage{
 		ObjectMeta: metav1.ObjectMeta{
@@ -152,14 +152,14 @@ func GetExpectedCVMIFrom(cclItem imgregv1a1.ClusterContentLibraryItem,
 	}
 
 	if providerFunc != nil {
-		_ = providerFunc(nil, "", cvmi)
+		_ = providerFunc(nil, nil, cvmi)
 	}
 
 	return cvmi
 }
 
 func GetExpectedVMIFrom(clItem imgregv1a1.ContentLibraryItem,
-	providerFunc func(context.Context, string, crtlclient.Object) error) *vmopv1a1.VirtualMachineImage {
+	providerFunc func(context.Context, crtlclient.Object, crtlclient.Object) error) *vmopv1a1.VirtualMachineImage {
 
 	vmi := &vmopv1a1.VirtualMachineImage{
 		ObjectMeta: metav1.ObjectMeta{
@@ -206,7 +206,7 @@ func GetExpectedVMIFrom(clItem imgregv1a1.ContentLibraryItem,
 	}
 
 	if providerFunc != nil {
-		_ = providerFunc(nil, "", vmi)
+		_ = providerFunc(nil, nil, vmi)
 	}
 
 	return vmi

pkg/util/cache.go

@@ -0,0 +1,183 @@
+// Copyright (c) 2023 VMware, Inc. All Rights Reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package util
+
+import (
+	"sync"
+	"time"
+)
+
+// CacheItem wraps T so it has a LastUpdated field and can be used with Cache.
+type CacheItem[T any] struct {
+	// Item is the cached item.
+	Item T
+
+	// LastUpdated is the last time the item was updated.
+	LastUpdated time.Time
+}
+
+// Cache is a generic implementation of a cache that can be configured with a
+// maximum number of items and can evict items after a certain amount of time.
+type Cache[T any] struct {
+	mu       sync.Mutex
+	items    map[string]CacheItem[T]
+	maxItems int
+	done     chan struct{}
+	donce    sync.Once
+
+	// expiredChan returns a channel on which the IDs of items that are expired
+	// are sent. This channel is closed when the cache is closed, but only after
+	// all, pending eviction notices are received.
+	expiredChan chan string
+	expiringNum sync.WaitGroup
+}
+
+// NewCache initializes a new cache with the provided expiration options.
+func NewCache[T any](
+	expireAfter, checkExpireInterval time.Duration,
+	maxItems int) *Cache[T] {
+
+	c := &Cache[T]{
+		expiredChan: make(chan string),
+		done:        make(chan struct{}),
+		items:       map[string]CacheItem[T]{},
+		maxItems:    maxItems,
+	}
+
+	// Start a goroutine to expire items from the cache.
+	go func() {
+		ticker := time.NewTicker(checkExpireInterval)
+
+		// evictFn iterates over the cached items and evicts expired items
+		evictFn := func() {
+			c.mu.Lock()
+			defer c.mu.Unlock()
+			for k, v := range c.items {
+				if time.Since(v.LastUpdated) > expireAfter {
+					// Delete the item from the cache.
+					delete(c.items, k)
+
+					// Notify subscribers that the item with the ID of k has
+					// been evicted if the cache is not closed. Otherwise we
+					// add one to the number of items being expired. This
+					// ensures the channel cannot be closed until all, pending
+					// notifications are sent.
+					select {
+					case <-c.done:
+					default:
+						c.expiringNum.Add(1)
+						go func(k string) {
+							c.expiredChan <- k
+							c.expiringNum.Done()
+						}(k)
+					}
+				}
+			}
+		}
+		for {
+			select {
+			case <-c.done:
+				ticker.Stop()
+				return
+			case <-ticker.C:
+				evictFn()
+			}
+		}
+	}()
+
+	return c
+}
+
+// ExpiredChan returns a channel on which the IDs of items that are expired are
+// sent. This channel is closed when the cache is closed, but only after all,
+// pending eviction notices are received.
+func (c *Cache[T]) ExpiredChan() <-chan string {
+	return c.expiredChan
+}
+
+// Close shuts down the cache. It is safe to call this function more than once.
+func (c *Cache[T]) Close() {
+	c.donce.Do(func() {
+		close(c.done)
+
+		// Wait until all pending, expiry notifications are received until the
+		// expired channel is closed.
+		go func() {
+			c.expiringNum.Wait()
+			close(c.expiredChan)
+		}()
+	})
+}
+
+// Delete removes the specified item from the cache.
+func (c *Cache[T]) Delete(id string) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	delete(c.items, id)
+}
+
+// Get returns the cached item with the provided ID. The function isHit may be
+// optionally provided to further determine whether a cached item should hit
+// or miss.
+func (c *Cache[T]) Get(id string, isHit func(t T) bool) (T, bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	var emptyT T
+
+	v, ok := c.items[id]
+	if !ok {
+		return emptyT, false
+	}
+
+	if isHit != nil && !isHit(v.Item) {
+		return emptyT, false
+	}
+
+	return v.Item, true
+}
+
+// CachePutResult describes the result of putting an item into the cache.
+type CachePutResult uint
+
+const (
+	// CachePutResultCreate indicates the item was created in the cache.
+	CachePutResultCreate CachePutResult = iota + 1
+
+	// CachePutResultUpdate indicates an existing, cached item was updated.
+	CachePutResultUpdate
+
+	// CachePutResultMaxItemsExceeded indicates the item was not stored because
+	// the cache's maximum number of items would have been exceeded.
+	CachePutResultMaxItemsExceeded
+)
+
+// Put stores the provided item in the cache, updating existing items.
+func (c *Cache[T]) Put(id string, t T) CachePutResult {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	v, ok := c.items[id]
+
+	if ok {
+		// Update existing item.
+		v.Item = t
+		v.LastUpdated = time.Now()
+		c.items[id] = v
+		return CachePutResultUpdate
+	}
+
+	// Store the new item in the cache as long as the maximum number of items
+	// was note exceeded.
+	if len(c.items) < c.maxItems {
+		// Store new item in the cache.
+		c.items[id] = CacheItem[T]{
+			Item:        t,
+			LastUpdated: time.Now(),
+		}
+		return CachePutResultCreate
+	}
+
+	return CachePutResultMaxItemsExceeded
+}