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,12 @@ type RawResponse struct {
 	Version string
 
 	// Resources to be included in the response.
-	Resources []types.Resource
+	Resources []types.ResourceWithTtl
+
+	// Whether this is a heartbeat response. For xDS versions that support TTL, this
+	// will be converted into a response that doesn't contain the actual resource protobuf.
+	// This allows for more lightweight updates that server only to update the TTL timer.
+	Heartbeat bool
 
 	// marshaledResponse holds an atomic reference to the serialized discovery response.
 	marshaledResponse atomic.Value
@@ -113,12 +119,16 @@ 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, resourceType, err := ttl.MaybeCreateTtlResourceIfSupported(resource, GetResourceName(resource.Resource), r.Request.TypeUrl, r.Heartbeat)
+			if err != nil {
+				return nil, err
+			}
+			marshaledResource, err := MarshalResource(maybeTtldResource)
 			if err != nil {
 				return nil, err
 			}
 			marshaledResources[i] = &any.Any{
-				TypeUrl: r.Request.TypeUrl,
+				TypeUrl: resourceType,
 				Value:   marshaledResource,
 			}
 		}

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

@@ -86,13 +86,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 {
 	// watchCount is an atomic counter incremented for each watch. This needs to
 	// be the first field in the struct to guarantee that it is 64-bit aligned,
@@ -81,6 +85,9 @@ type snapshotCache struct {
 	// hash is the hashing function for Envoy nodes
 	hash NodeHash
 
+	// heartbeatRoutines keeps track of the active heartbeat goroutines
+	heartbeatRoutines map[string]heartbeatHandle
+
 	mu sync.RWMutex
 }
 
@@ -96,12 +103,89 @@ 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,
+	return newSnapshotCache(ads, hash, logger)
+}
+
+func newSnapshotCache(ads bool, hash NodeHash, logger log.Logger) *snapshotCache {
+	cache := &snapshotCache{
+		log:               logger,
+		ads:               ads,
+		snapshots:         make(map[string]Snapshot),
+		status:            make(map[string]*statusInfo),
+		hash:              hash,
+		heartbeatRoutines: make(map[string]heartbeatHandle),
+	}
+
+	return cache
+}
+
+// NewSnapshotCacheWithHeartbeating initializes a simple cache that sends periodic heartbeat
+// responses for resources with a TTL.
+//
+// ADS flag forces a delay in responding to streaming requests until all
+// resources are explicitly named in the request. This avoids the problem of a
+// partial request over a single stream for a subset of resources which would
+// require generating a fresh version for acknowledgement. ADS flag requires
+// snapshot consistency. For non-ADS case (and fetch), multiple partial
+// requests are sent across multiple streams and re-using the snapshot version
+// is OK.
+//
+// Logger is optional.
+//
+// The context provides a way to cancel the heartbeating routine, while the heartbeatInterval
+// parameter controls how often heartbeating occurs.
+func NewSnapshotCacheWithHeartbeating(ctx context.Context, ads bool, hash NodeHash, logger log.Logger, heartbeatInterval time.Duration) SnapshotCache {
+	cache := newSnapshotCache(ads, hash, logger)
+	go func() {
+		t := time.NewTicker(heartbeatInterval)
+
+		for {
+			select {
+			case <-t.C:
+				cache.mu.Lock()
+				for node := range cache.status {
+					// TODO(snowp): Omit heartbeats if a real response has been sent recently.
+					cache.sendHeartbeats(ctx, node)
+				}
+				cache.mu.Unlock()
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+	return cache
+}
+
+func (cache *snapshotCache) sendHeartbeats(ctx context.Context, 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)
+			}
+
+			cache.respond(watch.Request, watch.Response, resourcesWithTtl, version, true)
+
+			// 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()
 	}
 }
 
@@ -113,6 +197,12 @@ 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()
+	}
+
 	// trigger existing watches for which version changed
 	if info, ok := cache.status[node]; ok {
 		info.mu.Lock()
@@ -122,7 +212,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)
@@ -165,7 +256,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)
@@ -212,7 +303,8 @@ func (cache *snapshotCache) CreateWatch(request *Request) (chan Response, func()
 	}
 
 	// 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 value, nil
 }
@@ -237,7 +329,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 {
@@ -253,11 +345,11 @@ func (cache *snapshotCache) respond(request *Request, value chan Response, resou
 			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
@@ -279,6 +371,7 @@ func createResponse(request *Request, resources map[string]types.Resource, versi
 		Request:   request,
 		Version:   version,
 		Resources: filtered,
+		Heartbeat: heartbeat,
 	}
 }
 
@@ -301,8 +394,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
 	}