DeleteRequest API (envoyproxy#59)

Fixes envoyproxy#57

This change adds an API to the cache to support deleting a request when given the key and the (pointer to the) request, which the orchestrator can use upon cancellation. If the key is not in the cache, nothing happens.

The input type to AddRequest was also changed to take a pointer to a request.

Signed-off-by: Lisa Lu <lisalu@lyft.com>

internal/app/cache/cache.go

@@ -19,10 +19,13 @@ type Cache interface {
 	Fetch(key string) (*Resource, error)
 
 	// SetResponse sets the cache response and returns the list of requests.
-	SetResponse(key string, resp v2.DiscoveryResponse) ([]*v2.DiscoveryRequest, error)
+	SetResponse(key string, resp v2.DiscoveryResponse) (map[*v2.DiscoveryRequest]bool, error)
 
 	// AddRequest adds the request to the cache.
 	AddRequest(key string, req *v2.DiscoveryRequest) error
+
+	// DeleteRequest removes the given request from any cache entries it's present in.
+	DeleteRequest(key string, req *v2.DiscoveryRequest) error
 }
 
 type cache struct {
@@ -39,7 +42,7 @@ type Response struct {
 
 type Resource struct {
 	Resp           *Response
-	Requests       []*v2.DiscoveryRequest
+	Requests       map[*v2.DiscoveryRequest]bool
 	ExpirationTime time.Time
 }
 
@@ -107,7 +110,7 @@ func (c *cache) Fetch(key string) (*Resource, error) {
 	return &resource, nil
 }
 
-func (c *cache) SetResponse(key string, resp v2.DiscoveryResponse) ([]*v2.DiscoveryRequest, error) {
+func (c *cache) SetResponse(key string, resp v2.DiscoveryResponse) (map[*v2.DiscoveryRequest]bool, error) {
 	c.cacheMu.Lock()
 	defer c.cacheMu.Unlock()
 	marshaledResources, err := MarshalResources(resp.Resources)
@@ -142,8 +145,10 @@ func (c *cache) AddRequest(key string, req *v2.DiscoveryRequest) error {
 	defer c.cacheMu.Unlock()
 	value, found := c.cache.Get(key)
 	if !found {
+		requests := make(map[*v2.DiscoveryRequest]bool)
+		requests[req] = true
 		resource := Resource{
-			Requests:       []*v2.DiscoveryRequest{req},
+			Requests:       requests,
 			ExpirationTime: c.getExpirationTime(time.Now()),
 		}
 		c.cache.Add(key, resource)
@@ -153,7 +158,23 @@ func (c *cache) AddRequest(key string, req *v2.DiscoveryRequest) error {
 	if !ok {
 		return fmt.Errorf("unable to cast cache value to type resource for key: %s", key)
 	}
-	resource.Requests = append(resource.Requests, req)
+	resource.Requests[req] = true
+	c.cache.Add(key, resource)
+	return nil
+}
+
+func (c *cache) DeleteRequest(key string, req *v2.DiscoveryRequest) error {
+	c.cacheMu.Lock()
+	defer c.cacheMu.Unlock()
+	value, found := c.cache.Get(key)
+	if !found {
+		return nil
+	}
+	resource, ok := value.(Resource)
+	if !ok {
+		return fmt.Errorf("unable to cast cache value to type resource for key: %s", key)
+	}
+	delete(resource.Requests, req)
 	c.cache.Add(key, resource)
 	return nil
 }

internal/app/cache/cache_test.go

@@ -29,7 +29,7 @@ func testOnEvict(key string, value Resource) {
 	})
 }
 
-var testRequest = v2.DiscoveryRequest{
+var testRequestA = v2.DiscoveryRequest{
 	VersionInfo: "version_A",
 	Node: &core.Node{
 		Id:      "id_A",
@@ -40,6 +40,17 @@ var testRequest = v2.DiscoveryRequest{
 	ResponseNonce: "nonce_A",
 }
 
+var testRequestB = v2.DiscoveryRequest{
+	VersionInfo: "version_B",
+	Node: &core.Node{
+		Id:      "id_B",
+		Cluster: "cluster_B",
+	},
+	ResourceNames: []string{"resource_B"},
+	TypeUrl:       "typeURL_B",
+	ResponseNonce: "nonce_B",
+}
+
 var testDiscoveryResponse = v2.DiscoveryResponse{
 	VersionInfo: "version_A",
 	Resources: []*any.Any{
@@ -74,7 +85,7 @@ func TestAddRequestAndFetch(t *testing.T) {
 	assert.EqualError(t, err, "no value found for key: key_A")
 	assert.Nil(t, resource)
 
-	err = cache.AddRequest(testKeyA, &testRequest)
+	err = cache.AddRequest(testKeyA, &testRequestA)
 	assert.NoError(t, err)
 
 	resource, err = cache.Fetch(testKeyA)
@@ -101,20 +112,20 @@ func TestSetResponseAndFetch(t *testing.T) {
 }
 
 func TestAddRequestAndSetResponse(t *testing.T) {
-	cache, err := NewCache(1, testOnEvict, time.Second*60)
+	cache, err := NewCache(2, testOnEvict, time.Second*60)
 	assert.NoError(t, err)
 
-	err = cache.AddRequest(testKeyA, &testRequest)
+	err = cache.AddRequest(testKeyA, &testRequestA)
 	assert.NoError(t, err)
 
-	err = cache.AddRequest(testKeyA, &testRequest)
+	err = cache.AddRequest(testKeyA, &testRequestB)
 	assert.NoError(t, err)
 
 	requests, err := cache.SetResponse(testKeyA, testDiscoveryResponse)
 	assert.NoError(t, err)
 	assert.Equal(t, 2, len(requests))
-	assert.Equal(t, testRequest, *requests[0])
-	assert.Equal(t, testRequest, *requests[1])
+	assert.Equal(t, true, requests[&testRequestA])
+	assert.Equal(t, true, requests[&testRequestB])
 
 	resource, err := cache.Fetch(testKeyA)
 	assert.NoError(t, err)
@@ -136,7 +147,7 @@ func TestMaxEntries(t *testing.T) {
 		key:    testKeyA,
 		reason: "testOnEvict called",
 	}, func() {
-		err = cache.AddRequest(testKeyB, &testRequest)
+		err = cache.AddRequest(testKeyB, &testRequestB)
 		assert.NoError(t, err)
 	})
 
@@ -223,3 +234,24 @@ func TestGetExpirationTime(t *testing.T) {
 	expirationTime := time.Date(0, 0, 0, 0, 0, 2, 0, time.UTC)
 	assert.Equal(t, expirationTime, c.getExpirationTime(currentTime))
 }
+
+func TestDeleteRequest(t *testing.T) {
+	cache, err := NewCache(1, testOnEvict, time.Second*60)
+	assert.NoError(t, err)
+
+	err = cache.AddRequest(testKeyA, &testRequestA)
+	assert.NoError(t, err)
+
+	err = cache.AddRequest(testKeyA, &testRequestA)
+	assert.NoError(t, err)
+
+	err = cache.DeleteRequest(testKeyA, &testRequestA)
+	assert.NoError(t, err)
+
+	requests, err := cache.SetResponse(testKeyA, testDiscoveryResponse)
+	assert.NoError(t, err)
+	assert.Equal(t, 0, len(requests))
+
+	err = cache.DeleteRequest(testKeyB, &testRequestB)
+	assert.NoError(t, err)
+}