poc

ShareChat · Nov 6, 2024 · a16595b · a16595b
1 parent 302db05
commit a16595b
Showing 1 changed file with 23 additions and 14 deletions.
diff --git a/pkg/cache/v3/simple.go b/pkg/cache/v3/simple.go
@@ -17,13 +17,14 @@ package cache
 import (
 	"context"
 	"fmt"
-	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
-	"github.com/envoyproxy/go-control-plane/pkg/resource/v3"
 	"strconv"
 	"sync"
 	"sync/atomic"
 	"time"
 
+	core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
+	"github.com/envoyproxy/go-control-plane/pkg/resource/v3"
+
 	"github.com/envoyproxy/go-control-plane/pkg/cache/types"
 	"github.com/envoyproxy/go-control-plane/pkg/log"
 	"github.com/envoyproxy/go-control-plane/pkg/server/stream/v3"
@@ -92,15 +93,15 @@ type SnapshotCache interface {
 	// GetStatusKeys retrieves node IDs for all statuses.
 	GetStatusKeys() []string
 
-	UpsertResources(ctx context.Context, node string, typ string, resourcesUpserted map[string]*types.ResourceWithTTL) error
+	UpsertResources(ctx context.Context, node string, typ string, resourcesUpserted map[string]*types.ResourceWithTTL, isCanary bool) error
 
-	BatchUpsertResources(ctx context.Context, typ string, resourcesUpserted map[string]map[string]*types.ResourceWithTTL) error
+	BatchUpsertResources(ctx context.Context, typ string, resourcesUpserted map[string]map[string]*types.ResourceWithTTL, isCanary bool) error
 
-	DeleteResources(ctx context.Context, node string, typ string, resourcesToDeleted []string) error
+	DeleteResources(ctx context.Context, node string, typ string, resourcesToDeleted []string, isCanary bool) error
 
-	DrainResources(ctx context.Context, node string, typ string, resourcesToDeleted []string) error
+	DrainResources(ctx context.Context, node string, typ string, resourcesToDeleted []string, isCanary bool) error
 
-	UpdateVirtualHosts(ctx context.Context, node string, typ string, resources map[string]map[string]*types.ResourceWithTTL) error
+	UpdateVirtualHosts(ctx context.Context, node string, typ string, resources map[string]map[string]*types.ResourceWithTTL, isCanary bool) error
 }
 
 type snapshotCache struct {
@@ -242,7 +243,8 @@ func (cache *snapshotCache) ParseSystemVersionInfo(version string) int64 {
 	return parsed
 }
 
-func (cache *snapshotCache) BatchUpsertResources(ctx context.Context, typ string, batchResourcesUpserted map[string]map[string]*types.ResourceWithTTL) error {
+// TODO:
+func (cache *snapshotCache) BatchUpsertResources(ctx context.Context, typ string, batchResourcesUpserted map[string]map[string]*types.ResourceWithTTL, isCanary bool) error {
 	cache.mu.Lock()
 	defer cache.mu.Unlock()
 	for node, resourcesUpserted := range batchResourcesUpserted {
@@ -328,7 +330,8 @@ func (cache *snapshotCache) BatchUpsertResources(ctx context.Context, typ string
 	return nil
 }
 
-func (cache *snapshotCache) UpsertResources(ctx context.Context, node string, typ string, resourcesUpserted map[string]*types.ResourceWithTTL) error {
+// TODO:
+func (cache *snapshotCache) UpsertResources(ctx context.Context, node string, typ string, resourcesUpserted map[string]*types.ResourceWithTTL, isCanary bool) error {
 	cache.mu.Lock()
 	if snapshot, ok := cache.snapshots[node]; ok {
 		defer cache.mu.Unlock()
@@ -398,7 +401,8 @@ func (cache *snapshotCache) UpsertResources(ctx context.Context, node string, ty
 	return nil
 }
 
-func (cache *snapshotCache) UpdateVirtualHosts(ctx context.Context, _ string, typ string, resources map[string]map[string]*types.ResourceWithTTL) error {
+// TODO:
+func (cache *snapshotCache) UpdateVirtualHosts(ctx context.Context, _ string, typ string, resources map[string]map[string]*types.ResourceWithTTL, isCanary bool) error {
 	index := GetResponseType(typ)
 
 	var wg sync.WaitGroup
@@ -471,7 +475,8 @@ func (cache *snapshotCache) UpdateVirtualHosts(ctx context.Context, _ string, ty
 	return nil
 }
 
-func (cache *snapshotCache) DeleteResources(ctx context.Context, node string, typ string, resourcesToDeleted []string) error {
+// TODO:
+func (cache *snapshotCache) DeleteResources(ctx context.Context, node string, typ string, resourcesToDeleted []string, isCanary bool) error {
 	cache.mu.Lock()
 	defer cache.mu.Unlock()
 
@@ -497,7 +502,7 @@ func (cache *snapshotCache) DeleteResources(ctx context.Context, node string, ty
 			defer info.mu.Unlock()
 
 			// Respond to delta watches for the node.
-			return cache.respondDeltaWatches(ctx, info, snapshot)
+			return cache.respondDeltaWatches(ctx, info, snapshot, isCanary)
 		}
 
 	} else if typ == resource.EndpointType {
@@ -551,7 +556,8 @@ func (cache *snapshotCache) DeleteResources(ctx context.Context, node string, ty
 	return nil
 }
 
-func (cache *snapshotCache) DrainResources(ctx context.Context, _ string, typ string, resourcesToDrain []string) error {
+// TODO:
+func (cache *snapshotCache) DrainResources(ctx context.Context, _ string, typ string, resourcesToDrain []string, isCanary bool) error {
 	//cache.mu.Lock()
 	//defer cache.mu.Unlock()
 	//
@@ -681,7 +687,7 @@ func (cache *snapshotCache) respondSOTWWatches(ctx context.Context, info *status
 	return nil
 }
 
-func (cache *snapshotCache) respondDeltaWatches(ctx context.Context, info *statusInfo, snapshot ResourceSnapshot) error {
+func (cache *snapshotCache) respondDeltaWatches(ctx context.Context, info *statusInfo, snapshot ResourceSnapshot, isCanary bool) error {
 	// We only calculate version hashes when using delta. We don't
 	// want to do this when using SOTW so we can avoid unnecessary
 	// computational cost if not using delta.
@@ -701,6 +707,9 @@ func (cache *snapshotCache) respondDeltaWatches(ctx context.Context, info *statu
 		info.orderResponseDeltaWatches()
 		for _, key := range info.orderedDeltaWatches {
 			watch := info.deltaWatches[key.ID]
+			if isCanary {
+				// TODO: If node is not in canary mode, skip responding to canary watches
+			}
 			res, err := cache.respondDelta(
 				ctx,
 				snapshot,