Orchestrator implementation

internal/app/cache/cache.go

@@ -113,7 +113,7 @@ func (c *cache) Fetch(key string) (*Resource, 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)
+	marshaledResources, err := MarshalResources(resp.Resources)
 	if err != nil {
 		return nil, fmt.Errorf("failed to marshal resources for key: %s, err %v", key, err)
 	}
@@ -126,6 +126,7 @@ func (c *cache) SetResponse(key string, resp v2.DiscoveryResponse) (map[*v2.Disc
 		resource := Resource{
 			Resp:           response,
 			ExpirationTime: c.getExpirationTime(time.Now()),
+			Requests:       make(map[*v2.DiscoveryRequest]bool),
 		}
 		c.cache.Add(key, resource)
 		return nil, nil
@@ -193,7 +194,9 @@ func (c *cache) getExpirationTime(currentTime time.Time) time.Time {
 	return time.Time{}
 }
 
-func marshalResources(resources []*any.Any) ([]gcp_types.MarshaledResource, error) {
+// MarshalResource converts the raw xDS discovery resources into a serialized
+// form accepted by go-control-plane.
+func MarshalResources(resources []*any.Any) ([]gcp_types.MarshaledResource, error) {
 	var marshaledResources []gcp_types.MarshaledResource
 	for _, resource := range resources {
 		marshaledResource, err := gcp.MarshalResource(resource)

internal/app/cache/cache_test.go

@@ -74,7 +74,8 @@ var testResponse = Response{
 }
 
 var testResource = Resource{
-	Resp: &testResponse,
+	Resp:     &testResponse,
+	Requests: make(map[*v2.DiscoveryRequest]bool),
 }
 
 func TestAddRequestAndFetch(t *testing.T) {

internal/app/orchestrator/downstream.go

@@ -0,0 +1,80 @@
+// Package orchestrator is responsible for instrumenting inbound xDS client
+// requests to the correct aggregated key, forwarding a representative request
+// to the upstream origin server, and managing the lifecycle of downstream and
+// upstream connections and associates streams. It implements
+// go-control-plane's Cache interface in order to receive xDS-based requests,
+// send responses, and handle gRPC streams.
+//
+// This file manages the bookkeeping of downstream clients by tracking inbound
+// requests to their corresponding response channels. The contents of this file
+// are intended to only be used within the orchestrator module and should not
+// be exported.
+package orchestrator
+
+import (
+	"sync"
+
+	gcp "github.com/envoyproxy/go-control-plane/pkg/cache/v2"
+)
+
+// downstreamResponseMap is a map of downstream xDS client requests to response
+// channels.
+type downstreamResponseMap struct {
+	mu               sync.RWMutex
+	responseChannels map[*gcp.Request]chan gcp.Response
+}
+
+func newDownstreamResponseMap() downstreamResponseMap {
+	return downstreamResponseMap{
+		responseChannels: make(map[*gcp.Request]chan gcp.Response),
+	}
+}
+
+// createChannel initializes a new channel for a request if it doesn't already
+// exist.
+func (d *downstreamResponseMap) createChannel(req *gcp.Request) chan gcp.Response {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if _, ok := d.responseChannels[req]; !ok {
+		d.responseChannels[req] = make(chan gcp.Response, 1)
+	}
+	return d.responseChannels[req]
+}
+
+// get retrieves the channel where responses are set for the specified request.
+func (d *downstreamResponseMap) get(req *gcp.Request) (chan gcp.Response, bool) {
+	d.mu.RLock()
+	defer d.mu.RUnlock()
+	channel, ok := d.responseChannels[req]
+	return channel, ok
+}
+
+// delete removes the response channel and request entry from the map.
+// Note: We don't close the response channel prior to deletion because there
+// can be separate go routines that are still attempting to write to the
+// channel. We rely on garbage collection to clean up and close outstanding
+// response channels once the go routines finish writing to them.
+func (d *downstreamResponseMap) delete(req *gcp.Request) chan gcp.Response {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if channel, ok := d.responseChannels[req]; ok {
+		delete(d.responseChannels, req)
+		return channel
+	}
+	return nil
+}
+
+// deleteAll removes all response channels and request entries from the map.
+// Note: We don't close the response channel prior to deletion because there
+// can be separate go routines that are still attempting to write to the
+// channel. We rely on garbage collection to clean up and close outstanding
+// response channels once the go routines finish writing to them.
+func (d *downstreamResponseMap) deleteAll(watchers map[*gcp.Request]bool) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	for watch := range watchers {
+		if d.responseChannels[watch] != nil {
+			delete(d.responseChannels, watch)
+		}
+	}
+}