cache: add support for xDS TTLs

pkg/cache/types/types.go

@@ -1,6 +1,8 @@
 package types
 
 import (
+	"time"
+
 	"github.com/golang/protobuf/proto"
 )
 
@@ -9,6 +11,13 @@ type Resource interface {
 	proto.Message
 }
 
+// ResourceWithTtl is a Resource with an optional TTL.
+type ResourceWithTtl struct {
+	Resource Resource
+
+	Ttl *time.Duration
+}
+
 // MarshaledResource is an alias for the serialized binary array.
 type MarshaledResource = []byte
 

pkg/cache/v2/cache.go

@@ -22,6 +22,7 @@ import (
 
 	discovery "github.com/envoyproxy/go-control-plane/envoy/api/v2"
 	"github.com/envoyproxy/go-control-plane/pkg/cache/types"
+	ttl "github.com/envoyproxy/go-control-plane/pkg/ttl/v2"
 	"github.com/golang/protobuf/ptypes/any"
 )
 
@@ -82,7 +83,9 @@ type RawResponse struct {
 	Version string
 
 	// Resources to be included in the response.
-	Resources []types.Resource
+	Resources []types.ResourceWithTtl
+
+	Heartbeat bool
 
 	// marshaledResponse holds an atomic reference to the serialized discovery response.
 	marshaledResponse atomic.Value
@@ -113,7 +116,11 @@ func (r *RawResponse) GetDiscoveryResponse() (*discovery.DiscoveryResponse, erro
 		marshaledResources := make([]*any.Any, len(r.Resources))
 
 		for i, resource := range r.Resources {
-			marshaledResource, err := MarshalResource(resource)
+			maybeTtldResource, err := ttl.MaybeCreateTtlResource(resource, GetResourceName(resource.Resource), r.Heartbeat)
+			if err != nil {
+				return nil, err
+			}
+			marshaledResource, err := MarshalResource(maybeTtldResource)
 			if err != nil {
 				return nil, err
 			}

pkg/cache/v2/cache_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/envoyproxy/go-control-plane/pkg/cache/types"
 	"github.com/envoyproxy/go-control-plane/pkg/cache/v2"
 	"github.com/envoyproxy/go-control-plane/pkg/resource/v2"
+	ttl_helper "github.com/envoyproxy/go-control-plane/pkg/ttl/v2"
 	"github.com/golang/protobuf/ptypes"
 	"github.com/golang/protobuf/ptypes/any"
 	"github.com/stretchr/testify/assert"
@@ -18,7 +19,7 @@ const (
 )
 
 func TestResponseGetDiscoveryResponse(t *testing.T) {
-	routes := []types.Resource{&route.RouteConfiguration{Name: resourceName}}
+	routes := []types.ResourceWithTtl{{Resource: &route.RouteConfiguration{Name: resourceName}}}
 	resp := cache.RawResponse{
 		Request:   &discovery.DiscoveryRequest{TypeUrl: resource.RouteType},
 		Version:   "v",
@@ -65,3 +66,28 @@ func TestPassthroughResponseGetDiscoveryResponse(t *testing.T) {
 	assert.Equal(t, r.Name, resourceName)
 	assert.Equal(t, discoveryResponse, dr)
 }
+
+func TestHeartbeatResponseGetDiscoveryResponse(t *testing.T) {
+	routes := []types.ResourceWithTtl{{Resource: &route.RouteConfiguration{Name: resourceName}}}
+	resp := cache.RawResponse{
+		Request:   &discovery.DiscoveryRequest{TypeUrl: resource.RouteType},
+		Version:   "v",
+		Resources: routes,
+		Heartbeat: true,
+	}
+
+	discoveryResponse, err := resp.GetDiscoveryResponse()
+	assert.Nil(t, err)
+	assert.Equal(t, discoveryResponse.VersionInfo, resp.Version)
+	assert.Equal(t, len(discoveryResponse.Resources), 1)
+	assert.True(t, ttl_helper.IsTTLResource(discoveryResponse.Resources[0]))
+
+	cachedResponse, err := resp.GetDiscoveryResponse()
+	assert.Nil(t, err)
+	assert.Same(t, discoveryResponse, cachedResponse)
+
+	r := &route.RouteConfiguration{}
+	err = ptypes.UnmarshalAny(discoveryResponse.Resources[0], r)
+	assert.Nil(t, err)
+	assert.Equal(t, r.Name, resourceName)
+}

pkg/cache/v2/linear.go

@@ -91,19 +91,19 @@ func NewLinearCache(typeURL string, opts ...LinearCacheOption) *LinearCache {
 }
 
 func (cache *LinearCache) respond(value chan<- Response, staleResources []string) {
-	var resources []types.Resource
+	var resources []types.ResourceWithTtl
 	// TODO: optimize the resources slice creations across different clients
 	if len(staleResources) == 0 {
-		resources = make([]types.Resource, 0, len(cache.resources))
+		resources = make([]types.ResourceWithTtl, 0, len(cache.resources))
 		for _, resource := range cache.resources {
-			resources = append(resources, resource)
+			resources = append(resources, types.ResourceWithTtl{Resource: resource})
 		}
 	} else {
-		resources = make([]types.Resource, 0, len(staleResources))
+		resources = make([]types.ResourceWithTtl, 0, len(staleResources))
 		for _, name := range staleResources {
 			resource := cache.resources[name]
 			if resource != nil {
-				resources = append(resources, resource)
+				resources = append(resources, types.ResourceWithTtl{Resource: resource})
 			}
 		}
 	}

pkg/cache/v2/resource.go

@@ -82,13 +82,13 @@ func MarshalResource(resource types.Resource) (types.MarshaledResource, error) {
 
 // GetResourceReferences returns the names for dependent resources (EDS cluster
 // names for CDS, RDS routes names for LDS).
-func GetResourceReferences(resources map[string]types.Resource) map[string]bool {
+func GetResourceReferences(resources map[string]types.ResourceWithTtl) map[string]bool {
 	out := make(map[string]bool)
 	for _, res := range resources {
-		if res == nil {
+		if res.Resource == nil {
 			continue
 		}
-		switch v := res.(type) {
+		switch v := res.Resource.(type) {
 		case *endpoint.ClusterLoadAssignment:
 			// no dependencies
 		case *cluster.Cluster:

pkg/cache/v2/resource_test.go

@@ -138,7 +138,7 @@ func TestGetResourceReferences(t *testing.T) {
 		},
 	}
 	for _, cs := range cases {
-		names := cache.GetResourceReferences(cache.IndexResourcesByName([]types.Resource{cs.in}))
+		names := cache.GetResourceReferences(cache.IndexResourcesByName([]types.ResourceWithTtl{{Resource: cs.in}}))
 		if !reflect.DeepEqual(names, cs.out) {
 			t.Errorf("GetResourceReferences(%v) => got %v, want %v", cs.in, names, cs.out)
 		}

pkg/cache/v2/simple.go

@@ -60,6 +60,10 @@ type SnapshotCache interface {
 	GetStatusKeys() []string
 }
 
+type heartbeatHandle struct {
+	cancel func()
+}
+
 type snapshotCache struct {
 	log log.Logger
 
@@ -78,6 +82,8 @@ type snapshotCache struct {
 	// watchCount is an atomic counter incremented for each watch
 	watchCount int64
 
+	heartbeatRoutines map[string]heartbeatHandle
+
 	mu sync.RWMutex
 }
 
@@ -94,11 +100,56 @@ type snapshotCache struct {
 // Logger is optional.
 func NewSnapshotCache(ads bool, hash NodeHash, logger log.Logger) SnapshotCache {
 	return &snapshotCache{
-		log:       logger,
-		ads:       ads,
-		snapshots: make(map[string]Snapshot),
-		status:    make(map[string]*statusInfo),
-		hash:      hash,
+		log:               logger,
+		ads:               ads,
+		snapshots:         make(map[string]Snapshot),
+		status:            make(map[string]*statusInfo),
+		hash:              hash,
+		heartbeatRoutines: make(map[string]heartbeatHandle),
+	}
+}
+
+func (cache *snapshotCache) sendHeartbeats(ctx context.Context, wg *sync.WaitGroup, node string) {
+	snapshot := cache.snapshots[node]
+	if info, ok := cache.status[node]; ok {
+		info.mu.Lock()
+		for id, watch := range info.watches {
+			// Respond with the current version regardless of whether the version has changed.
+			version := snapshot.GetVersion(watch.Request.TypeUrl)
+			resources := snapshot.GetResourcesAndTtl(watch.Request.TypeUrl)
+
+			// TODO(snowp): Construct this once per type instead of once per watch.
+			resourcesWithTtl := map[string]types.ResourceWithTtl{}
+			for k, v := range resources {
+				if v.Ttl != nil {
+					resourcesWithTtl[k] = v
+				}
+			}
+
+			if len(resourcesWithTtl) == 0 {
+				continue
+			}
+			if cache.log != nil {
+				cache.log.Debugf("respond open watch %d%v with heartbeat for version %q", id, watch.Request.ResourceNames, version)
+			}
+
+			wg.Add(1)
+			// Spawn a goroutine here so that we can give up the lock before issuing responses. This prevents lock contention if the
+			// response channels aren't ready to receive the response.
+			go func(request *Request, response chan<- Response) {
+				defer wg.Done()
+				select {
+				case <-ctx.Done():
+					return
+				default:
+				}
+				cache.respond(request, response, resourcesWithTtl, version, true)
+			}(watch.Request, watch.Response)
+
+			// The watch must be deleted and we must rely on the client to ack this response to create a new watch.
+			delete(info.watches, id)
+		}
+		info.mu.Unlock()
 	}
 }
 
@@ -110,6 +161,43 @@ func (cache *snapshotCache) SetSnapshot(node string, snapshot Snapshot) error {
 	// update the existing entry
 	cache.snapshots[node] = snapshot
 
+	handle, ok := cache.heartbeatRoutines[node]
+	// No heartbeat configured and no existing heartbeat, nothing to do.
+	if ok {
+		handle.cancel()
+	}
+
+	if snapshot.HeartbeatInterval != nil {
+		heartBeatCtx, cancel := context.WithCancel(context.Background())
+		cache.heartbeatRoutines[node] = heartbeatHandle{cancel: cancel}
+
+		go func(heartbeatInterval time.Duration) {
+			ticker := time.NewTicker(heartbeatInterval)
+			for {
+				select {
+				case <-heartBeatCtx.Done():
+					return
+				case <-ticker.C:
+					wg := &sync.WaitGroup{}
+					cache.mu.Lock()
+
+					// It's possible for this goroutine to be canceled while we're waiting to grab the lock. In this case
+					// we want to avoid sending a heartbeat here and just end. If heartbeats should continue the another
+					// goroutine will have been created.
+					select {
+					case <-heartBeatCtx.Done():
+					default:
+						cache.sendHeartbeats(heartBeatCtx, wg, node)
+					}
+					cache.mu.Unlock()
+
+					// Block until we've finished sending all the heartbeats.
+					wg.Wait()
+				}
+			}
+		}(*snapshot.HeartbeatInterval)
+	}
+
 	// trigger existing watches for which version changed
 	if info, ok := cache.status[node]; ok {
 		info.mu.Lock()
@@ -119,7 +207,8 @@ func (cache *snapshotCache) SetSnapshot(node string, snapshot Snapshot) error {
 				if cache.log != nil {
 					cache.log.Debugf("respond open watch %d%v with new version %q", id, watch.Request.ResourceNames, version)
 				}
-				cache.respond(watch.Request, watch.Response, snapshot.GetResources(watch.Request.TypeUrl), version)
+				resources := snapshot.GetResourcesAndTtl(watch.Request.TypeUrl)
+				cache.respond(watch.Request, watch.Response, resources, version, false)
 
 				// discard the watch
 				delete(info.watches, id)
@@ -162,7 +251,7 @@ func nameSet(names []string) map[string]bool {
 }
 
 // superset checks that all resources are listed in the names set.
-func superset(names map[string]bool, resources map[string]types.Resource) error {
+func superset(names map[string]bool, resources map[string]types.ResourceWithTtl) error {
 	for resourceName := range resources {
 		if _, exists := names[resourceName]; !exists {
 			return fmt.Errorf("%q not listed", resourceName)
@@ -206,7 +295,8 @@ func (cache *snapshotCache) CreateWatch(request *Request, value chan<- Response)
 	}
 
 	// otherwise, the watch may be responded immediately
-	cache.respond(request, value, snapshot.GetResources(request.TypeUrl), version)
+	resources := snapshot.GetResourcesAndTtl(request.TypeUrl)
+	cache.respond(request, value, resources, version, false)
 
 	return nil
 }
@@ -231,7 +321,7 @@ func (cache *snapshotCache) cancelWatch(nodeID string, watchID int64) func() {
 
 // Respond to a watch with the snapshot value. The value channel should have capacity not to block.
 // TODO(kuat) do not respond always, see issue https://github.com/envoyproxy/go-control-plane/issues/46
-func (cache *snapshotCache) respond(request *Request, value chan<- Response, resources map[string]types.Resource, version string) {
+func (cache *snapshotCache) respond(request *Request, value chan<- Response, resources map[string]types.ResourceWithTtl, version string, heartbeat bool) {
 	// for ADS, the request names must match the snapshot names
 	// if they do not, then the watch is never responded, and it is expected that envoy makes another request
 	if len(request.ResourceNames) != 0 && cache.ads {
@@ -247,11 +337,11 @@ func (cache *snapshotCache) respond(request *Request, value chan<- Response, res
 			request.TypeUrl, request.ResourceNames, request.VersionInfo, version)
 	}
 
-	value <- createResponse(request, resources, version)
+	value <- createResponse(request, resources, version, heartbeat)
 }
 
-func createResponse(request *Request, resources map[string]types.Resource, version string) Response {
-	filtered := make([]types.Resource, 0, len(resources))
+func createResponse(request *Request, resources map[string]types.ResourceWithTtl, version string, heartbeat bool) Response {
+	filtered := make([]types.ResourceWithTtl, 0, len(resources))
 
 	// Reply only with the requested resources. Envoy may ask each resource
 	// individually in a separate stream. It is ok to reply with the same version
@@ -273,6 +363,7 @@ func createResponse(request *Request, resources map[string]types.Resource, versi
 		Request:   request,
 		Version:   version,
 		Resources: filtered,
+		Heartbeat: heartbeat,
 	}
 }
 
@@ -295,8 +386,8 @@ func (cache *snapshotCache) Fetch(ctx context.Context, request *Request) (Respon
 			return nil, &types.SkipFetchError{}
 		}
 
-		resources := snapshot.GetResources(request.TypeUrl)
-		out := createResponse(request, resources, version)
+		resources := snapshot.GetResourcesAndTtl(request.TypeUrl)
+		out := createResponse(request, resources, version, false)
 		return out, nil
 	}