@@ -26,6 +26,35 @@ import (
2626 "github.com/envoyproxy/go-control-plane/pkg/server/stream/v3"
2727)
2828
29+ // ResourceSnapshot is an abstract snapshot of a collection of resources that
30+ // can be stored in a SnapshotCache. This enables applications to use the
31+ // SnapshotCache watch machinery with their own resource types. Most
32+ // applications will use Snapshot.
33+ type ResourceSnapshot interface {
34+ // GetVersion should return the current version of the resource indicated
35+ // by typeURL. The version string that is returned is opaque and should
36+ // only be compared for equality.
37+ GetVersion (typeURL string ) string
38+
39+ // GetResourcesAndTTL returns all resources of the type indicted by
40+ // typeURL, together with their TTL.
41+ GetResourcesAndTTL (typeURL string ) map [string ]types.ResourceWithTTL
42+
43+ // GetResources returns all resources of the type indicted by
44+ // typeURL. This is identical to GetResourcesAndTTL, except that
45+ // the TTL is omitted.
46+ GetResources (typeURL string ) map [string ]types.Resource
47+
48+ // ConstructVersionMap is a hint that a delta watch will soon make a
49+ // call to GetVersionMap. The snapshot should construct an internal
50+ // opaque version string for each collection of resource types.
51+ ConstructVersionMap () error
52+
53+ // GetVersionMap returns a map of resource name to resource version for
54+ // all the resources of type indicated by typeURL.
55+ GetVersionMap (typeURL string ) map [string ]string
56+ }
57+
2958// SnapshotCache is a snapshot-based cache that maintains a single versioned
3059// snapshot of responses per node. SnapshotCache consistently replies with the
3160// latest snapshot. For the protocol to work correctly in ADS mode, EDS/RDS
@@ -46,10 +75,10 @@ type SnapshotCache interface {
4675 //
4776 // This method will cause the server to respond to all open watches, for which
4877 // the version differs from the snapshot version.
49- SetSnapshot (ctx context.Context , node string , snapshot Snapshot ) error
78+ SetSnapshot (ctx context.Context , node string , snapshot ResourceSnapshot ) error
5079
5180 // GetSnapshots gets the snapshot for a node.
52- GetSnapshot (node string ) (Snapshot , error )
81+ GetSnapshot (node string ) (ResourceSnapshot , error )
5382
5483 // ClearSnapshot removes all status and snapshot information associated with a node.
5584 ClearSnapshot (node string )
@@ -75,7 +104,7 @@ type snapshotCache struct {
75104 ads bool
76105
77106 // snapshots are cached resources indexed by node IDs
78- snapshots map [string ]Snapshot
107+ snapshots map [string ]ResourceSnapshot
79108
80109 // status information for all nodes indexed by node IDs
81110 status map [string ]* statusInfo
@@ -109,7 +138,7 @@ func newSnapshotCache(ads bool, hash NodeHash, logger log.Logger) *snapshotCache
109138 cache := & snapshotCache {
110139 log : logger ,
111140 ads : ads ,
112- snapshots : make (map [string ]Snapshot ),
141+ snapshots : make (map [string ]ResourceSnapshot ),
113142 status : make (map [string ]* statusInfo ),
114143 hash : hash ,
115144 }
@@ -188,7 +217,7 @@ func (cache *snapshotCache) sendHeartbeats(ctx context.Context, node string) {
188217}
189218
190219// SetSnapshotCacheContext updates a snapshot for a node.
191- func (cache * snapshotCache ) SetSnapshot (ctx context.Context , node string , snapshot Snapshot ) error {
220+ func (cache * snapshotCache ) SetSnapshot (ctx context.Context , node string , snapshot ResourceSnapshot ) error {
192221 cache .mu .Lock ()
193222 defer cache .mu .Unlock ()
194223
@@ -229,7 +258,7 @@ func (cache *snapshotCache) SetSnapshot(ctx context.Context, node string, snapsh
229258 for id , watch := range info .deltaWatches {
230259 res , err := cache .respondDelta (
231260 ctx ,
232- & snapshot ,
261+ snapshot ,
233262 watch .Request ,
234263 watch .Response ,
235264 watch .StreamState ,
@@ -249,13 +278,13 @@ func (cache *snapshotCache) SetSnapshot(ctx context.Context, node string, snapsh
249278}
250279
251280// GetSnapshots gets the snapshot for a node, and returns an error if not found.
252- func (cache * snapshotCache ) GetSnapshot (node string ) (Snapshot , error ) {
281+ func (cache * snapshotCache ) GetSnapshot (node string ) (ResourceSnapshot , error ) {
253282 cache .mu .RLock ()
254283 defer cache .mu .RUnlock ()
255284
256285 snap , ok := cache .snapshots [node ]
257286 if ! ok {
258- return Snapshot {} , fmt .Errorf ("no snapshot found for node %s" , node )
287+ return nil , fmt .Errorf ("no snapshot found for node %s" , node )
259288 }
260289 return snap , nil
261290}
@@ -306,8 +335,12 @@ func (cache *snapshotCache) CreateWatch(request *Request, streamState stream.Str
306335 info .lastWatchRequestTime = time .Now ()
307336 info .mu .Unlock ()
308337
338+ var version string
339+
309340 snapshot , exists := cache .snapshots [nodeID ]
310- version := snapshot .GetVersion (request .TypeUrl )
341+ if exists {
342+ version = snapshot .GetVersion (request .TypeUrl )
343+ }
311344
312345 if exists {
313346 knownResourceNames := streamState .GetKnownResourceNames (request .TypeUrl )
@@ -339,7 +372,6 @@ func (cache *snapshotCache) CreateWatch(request *Request, streamState stream.Str
339372 if ! exists || request .VersionInfo == version {
340373 watchID := cache .nextWatchID ()
341374 cache .log .Debugf ("open watch %d for %s%v from nodeID %q, version %q" , watchID , request .TypeUrl , request .ResourceNames , nodeID , request .VersionInfo )
342-
343375 info .mu .Lock ()
344376 info .watches [watchID ] = ResponseWatch {Request : request , Response : value }
345377 info .mu .Unlock ()
@@ -452,29 +484,34 @@ func (cache *snapshotCache) CreateDeltaWatch(request *DeltaRequest, state stream
452484 if exists {
453485 err := snapshot .ConstructVersionMap ()
454486 if err != nil {
455- cache .log .Errorf ("failed to compute version for snapshot resources inline, waiting for next snapshot update" )
487+ cache .log .Errorf ("failed to compute version for snapshot resources inline: %s" , err )
456488 }
457- response , err := cache .respondDelta (context .Background (), & snapshot , request , value , state )
489+ response , err := cache .respondDelta (context .Background (), snapshot , request , value , state )
458490 if err != nil {
459- cache .log .Errorf ("failed to respond with delta response, waiting for next snapshot update : %s" , err )
491+ cache .log .Errorf ("failed to respond with delta response: %s" , err )
460492 }
461493
462494 delayedResponse = response == nil
463495 }
464496
465497 if delayedResponse {
466498 watchID := cache .nextDeltaWatchID ()
467- cache .log .Infof ("open delta watch ID:%d for %s Resources:%v from nodeID: %q, system version %q" , watchID , t , state .GetResourceVersions (), nodeID , snapshot .GetVersion (t ))
468- info .setDeltaResponseWatch (watchID , DeltaResponseWatch {Request : request , Response : value , StreamState : state })
469499
500+ if exists {
501+ cache .log .Infof ("open delta watch ID:%d for %s Resources:%v from nodeID: %q, version %q" , watchID , t , state .GetResourceVersions (), nodeID , snapshot .GetVersion (t ))
502+ } else {
503+ cache .log .Infof ("open delta watch ID:%d for %s Resources:%v from nodeID: %q" , watchID , t , state .GetResourceVersions (), nodeID )
504+ }
505+
506+ info .setDeltaResponseWatch (watchID , DeltaResponseWatch {Request : request , Response : value , StreamState : state })
470507 return cache .cancelDeltaWatch (nodeID , watchID )
471508 }
472509
473510 return nil
474511}
475512
476513// Respond to a delta watch with the provided snapshot value. If the response is nil, there has been no state change.
477- func (cache * snapshotCache ) respondDelta (ctx context.Context , snapshot * Snapshot , request * DeltaRequest , value chan DeltaResponse , state stream.StreamState ) (* RawDeltaResponse , error ) {
514+ func (cache * snapshotCache ) respondDelta (ctx context.Context , snapshot ResourceSnapshot , request * DeltaRequest , value chan DeltaResponse , state stream.StreamState ) (* RawDeltaResponse , error ) {
478515 resp := createDeltaResponse (ctx , request , state , resourceContainer {
479516 resourceMap : snapshot .GetResources (request .TypeUrl ),
480517 versionMap : snapshot .GetVersionMap (request .TypeUrl ),
0 commit comments