fix: Repo-server has silent unmarshalling errors leading to empty applications (#4423)

reposerver/cache/cache.go

@@ -1,8 +1,10 @@
 package cache
 
 import (
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
+	"hash/fnv"
 	"time"
 
 	"github.com/go-redis/redis/v8"
@@ -12,6 +14,8 @@ import (
 	"github.com/argoproj/argo-cd/reposerver/apiclient"
 	cacheutil "github.com/argoproj/argo-cd/util/cache"
 	"github.com/argoproj/argo-cd/util/hash"
+
+	log "github.com/sirupsen/logrus"
 )
 
 var ErrCacheMiss = cacheutil.ErrCacheMiss
@@ -70,10 +74,48 @@ func manifestCacheKey(revision string, appSrc *appv1.ApplicationSource, namespac
 }
 
 func (c *Cache) GetManifests(revision string, appSrc *appv1.ApplicationSource, namespace string, appLabelKey string, appLabelValue string, res *CachedManifestResponse) error {
-	return c.cache.GetItem(manifestCacheKey(revision, appSrc, namespace, appLabelKey, appLabelValue), res)
+	err := c.cache.GetItem(manifestCacheKey(revision, appSrc, namespace, appLabelKey, appLabelValue), res)
+
+	if err != nil {
+		return err
+	}
+
+	hash, err := res.generateCacheEntryHash()
+	if err != nil {
+		return fmt.Errorf("Unable to generate hash value: %s", err)
+	}
+
+	// If the expected hash of the cache entry does not match the actual hash value...
+	if hash != res.CacheEntryHash {
+		log.Warnf("Manifest hash did not match expected value, treating as a cache miss: %s", appLabelValue)
+
+		err = c.DeleteManifests(revision, appSrc, namespace, appLabelKey, appLabelValue)
+		if err != nil {
+			return fmt.Errorf("Unable to delete manifest after hash mismatch, %v", err)
+		}
+
+		// Treat hash mismatches as cache misses, so that the underlying resource is reacquired
+		return ErrCacheMiss
+	}
+
+	// The expected hash matches the actual hash, so remove the hash from the returned value
+	res.CacheEntryHash = ""
+
+	return nil
 }
 
 func (c *Cache) SetManifests(revision string, appSrc *appv1.ApplicationSource, namespace string, appLabelKey string, appLabelValue string, res *CachedManifestResponse) error {
+
+	// Generate and apply the cache entry hash, before writing
+	if res != nil {
+		res = res.shallowCopy()
+		hash, err := res.generateCacheEntryHash()
+		if err != nil {
+			return fmt.Errorf("Unable to generate hash value: %s", err)
+		}
+		res.CacheEntryHash = hash
+	}
+
 	return c.cache.SetItem(manifestCacheKey(revision, appSrc, namespace, appLabelKey, appLabelValue), res, c.repoCacheExpiration, res == nil)
 }
 
@@ -106,12 +148,52 @@ func (c *Cache) SetRevisionMetadata(repoURL, revision string, item *appv1.Revisi
 	return c.cache.SetItem(revisionMetadataKey(repoURL, revision), item, c.repoCacheExpiration, false)
 }
 
+func (cmr *CachedManifestResponse) shallowCopy() *CachedManifestResponse {
+	if cmr == nil {
+		return nil
+	}
+
+	return &CachedManifestResponse{
+		CacheEntryHash:                  cmr.CacheEntryHash,
+		FirstFailureTimestamp:           cmr.FirstFailureTimestamp,
+		ManifestResponse:                cmr.ManifestResponse,
+		MostRecentError:                 cmr.MostRecentError,
+		NumberOfCachedResponsesReturned: cmr.NumberOfCachedResponsesReturned,
+		NumberOfConsecutiveFailures:     cmr.NumberOfConsecutiveFailures,
+	}
+}
+
+func (cmr *CachedManifestResponse) generateCacheEntryHash() (string, error) {
+
+	// Copy, then remove the old hash
+	copy := cmr.shallowCopy()
+	copy.CacheEntryHash = ""
+
+	// Hash the JSON representation into a base-64-encoded FNV 64a (we don't need a cryptographic hash algorithm, since this is only for detecting data corruption)
+	bytes, err := json.Marshal(copy)
+	if err != nil {
+		return "", err
+	}
+	h := fnv.New64a()
+	_, err = h.Write(bytes)
+	if err != nil {
+		return "", err
+	}
+	fnvHash := h.Sum(nil)
+	return base64.URLEncoding.EncodeToString(fnvHash), nil
+
+}
+
 // CachedManifestResponse represents a cached result of a previous manifest generation operation, including the caching
 // of a manifest generation error, plus additional information on previous failures
 type CachedManifestResponse struct {
-	ManifestResponse                *apiclient.ManifestResponse
-	MostRecentError                 string
-	FirstFailureTimestamp           int64
-	NumberOfConsecutiveFailures     int
-	NumberOfCachedResponsesReturned int
+
+	// NOTE: When adding fields to this struct, you MUST also update shallowCopy()
+
+	CacheEntryHash                  string                      `json:"cacheEntryHash"`
+	ManifestResponse                *apiclient.ManifestResponse `json:"manifestResponse"`
+	MostRecentError                 string                      `json:"mostRecentError"`
+	FirstFailureTimestamp           int64                       `json:"firstFailureTimestamp"`
+	NumberOfConsecutiveFailures     int                         `json:"numberOfConsecutiveFailures"`
+	NumberOfCachedResponsesReturned int                         `json:"numberOfCachedResponsesReturned"`
 }

reposerver/cache/cache_test.go

@@ -1,13 +1,17 @@
 package cache
 
 import (
+	"encoding/json"
+	"errors"
+	"strings"
 	"testing"
 	"time"
 
 	"github.com/spf13/cobra"
 	"github.com/stretchr/testify/assert"
 
 	. "github.com/argoproj/argo-cd/pkg/apis/application/v1alpha1"
+	appv1 "github.com/argoproj/argo-cd/pkg/apis/application/v1alpha1"
 	"github.com/argoproj/argo-cd/reposerver/apiclient"
 	cacheutil "github.com/argoproj/argo-cd/util/cache"
 )
@@ -120,3 +124,174 @@ func TestAddCacheFlagsToCmd(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, 24*time.Hour, cache.repoCacheExpiration)
 }
+
+func TestCachedManifestResponse_HashBehavior(t *testing.T) {
+
+	inMemCache := cacheutil.NewInMemoryCache(1 * time.Hour)
+
+	repoCache := NewCache(
+		cacheutil.NewCache(inMemCache),
+		1*time.Minute,
+	)
+
+	response := apiclient.ManifestResponse{
+		Namespace: "default",
+		Revision:  "revision",
+		Manifests: []string{"sample-text"},
+	}
+	appSrc := &appv1.ApplicationSource{}
+	appKey := "key"
+	appValue := "value"
+
+	// Set the value in the cache
+	store := &CachedManifestResponse{
+		FirstFailureTimestamp:           0,
+		ManifestResponse:                &response,
+		MostRecentError:                 "",
+		NumberOfCachedResponsesReturned: 0,
+		NumberOfConsecutiveFailures:     0,
+	}
+	err := repoCache.SetManifests(response.Revision, appSrc, response.Namespace, appKey, appValue, store)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	// Get the cache entry of the set value directly from the in memory cache, and check the values
+	var cacheKey string
+	var cmr *CachedManifestResponse
+	{
+
+		items := getInMemoryCacheContents(t, inMemCache)
+
+		assert.Equal(t, len(items), 1)
+
+		for key, val := range items {
+			cmr = val
+			cacheKey = key
+		}
+		assert.NotEmpty(t, cmr.CacheEntryHash)
+		assert.NotNil(t, cmr.ManifestResponse)
+		assert.Equal(t, cmr.ManifestResponse, store.ManifestResponse)
+
+		regeneratedHash, err := cmr.generateCacheEntryHash()
+		if err != nil {
+			t.Fatal(err)
+		}
+		assert.Equal(t, cmr.CacheEntryHash, regeneratedHash)
+	}
+
+	// Retrieve the value using 'GetManifests' and confirm it works
+	retrievedVal := &CachedManifestResponse{}
+	err = repoCache.GetManifests(response.Revision, appSrc, response.Namespace, appKey, appValue, retrievedVal)
+	if err != nil {
+		t.Fatal(err)
+	}
+	assert.Equal(t, retrievedVal, store)
+
+	// Corrupt the hash so that it doesn't match
+	{
+		newCmr := cmr.shallowCopy()
+		newCmr.CacheEntryHash = "!bad-hash!"
+
+		err := inMemCache.Set(&cacheutil.Item{
+			Key:    cacheKey,
+			Object: &newCmr,
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+	}
+
+	// Retrieve the value using GetManifests and confirm it returns a cache miss
+	retrievedVal = &CachedManifestResponse{}
+	err = repoCache.GetManifests(response.Revision, appSrc, response.Namespace, appKey, appValue, retrievedVal)
+
+	assert.True(t, err == cacheutil.ErrCacheMiss)
+
+	// Verify that the hash mismatch item has been deleted
+	items := getInMemoryCacheContents(t, inMemCache)
+	assert.Equal(t, len(items), 0)
+
+}
+
+func getInMemoryCacheContents(t *testing.T, inMemCache *cacheutil.InMemoryCache) map[string]*CachedManifestResponse {
+	items, err := inMemCache.Items(func() interface{} { return &CachedManifestResponse{} })
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	result := map[string]*CachedManifestResponse{}
+	for key, val := range items {
+		obj, ok := val.(*CachedManifestResponse)
+		if !ok {
+			t.Fatal(errors.New("Unexpected type in cache"))
+		}
+
+		result[key] = obj
+	}
+
+	return result
+}
+
+func TestCachedManifestResponse_ShallowCopy(t *testing.T) {
+
+	pre := &CachedManifestResponse{
+		CacheEntryHash:        "value",
+		FirstFailureTimestamp: 1,
+		ManifestResponse: &apiclient.ManifestResponse{
+			Manifests: []string{"one", "two"},
+		},
+		MostRecentError:                 "error",
+		NumberOfCachedResponsesReturned: 2,
+		NumberOfConsecutiveFailures:     3,
+	}
+
+	post := pre.shallowCopy()
+	assert.Equal(t, pre, post)
+
+	unequal := &CachedManifestResponse{
+		CacheEntryHash:        "diff-value",
+		FirstFailureTimestamp: 1,
+		ManifestResponse: &apiclient.ManifestResponse{
+			Manifests: []string{"one", "two"},
+		},
+		MostRecentError:                 "error",
+		NumberOfCachedResponsesReturned: 2,
+		NumberOfConsecutiveFailures:     3,
+	}
+	assert.NotEqual(t, pre, unequal)
+}
+
+func TestCachedManifestResponse_ShallowCopyExpectedFields(t *testing.T) {
+
+	// Attempt to ensure that the developer updated CachedManifestResponse.shallowCopy(), by doing a sanity test of the structure here
+
+	val := &CachedManifestResponse{}
+
+	str, err := json.Marshal(val)
+	if err != nil {
+		assert.FailNow(t, "Unable to marshal", err)
+		return
+	}
+
+	jsonMap := map[string]interface{}{}
+	err = json.Unmarshal(str, &jsonMap)
+	if err != nil {
+		assert.FailNow(t, "Unable to unmarshal", err)
+		return
+	}
+
+	expectedFields := []string{"cacheEntryHash", "manifestResponse", "mostRecentError", "firstFailureTimestamp",
+		"numberOfConsecutiveFailures", "numberOfCachedResponsesReturned"}
+
+	assert.Equal(t, len(jsonMap), len(expectedFields))
+
+	// If this test failed, you probably also forgot to update CachedManifestResponse.shallowCopy(), so
+	// go do that first :)
+
+	for _, expectedField := range expectedFields {
+		assert.Truef(t, strings.Contains(string(str), "\""+expectedField+"\""), "Missing field: %s", expectedField)
+	}
+
+}

util/cache/inmemory.go

@@ -54,3 +54,25 @@ func (i *InMemoryCache) OnUpdated(ctx context.Context, key string, callback func
 func (i *InMemoryCache) NotifyUpdated(key string) error {
 	return nil
 }
+
+// Items return a list of items in the cache; requires passing a constructor function
+// so that the items can be decoded from gob format.
+func (i *InMemoryCache) Items(createNewObject func() interface{}) (map[string]interface{}, error) {
+
+	result := map[string]interface{}{}
+
+	for key, value := range i.memCache.Items() {
+
+		buf := value.Object.(bytes.Buffer)
+		obj := createNewObject()
+		err := gob.NewDecoder(&buf).Decode(obj)
+		if err != nil {
+			return nil, err
+		}
+
+		result[key] = obj
+
+	}
+
+	return result, nil
+}