Set up watchable to communicate between components

docs/design/SYSTEM_DESIGN.md

@@ -40,6 +40,9 @@ defined as Kubernetes resources that provide the following services:
 
 ### Components
 
+Envoy Gateway is made up of several components that communicate in-process; how this communication happens is described
+in [watching.md][].
+
 #### Provider
 A Provider is an infrastructure component that Envoy Gateway calls to establish its runtime configuration, resolve
 services, persist data, etc. Kubernetes and File are the only supported providers. However, other providers can be added
@@ -141,3 +144,4 @@ The draft for this document is [here][draft_design].
 [be]: https://gateway-api.sigs.k8s.io/v1alpha2/references/spec/#gateway.networking.k8s.io/v1alpha2.BackendObjectReference
 [svc]: https://kubernetes.io/docs/concepts/services-networking/service/
 [issue_95]: https://github.com/envoyproxy/gateway/pull/95
+[watching.md]: ./watching.md

docs/design/watching.md

@@ -0,0 +1,112 @@
+## Watching the results of other components
+
+Envoy Gateway is made up of several components that communicate in-process.  Some of them (namely providers) watch
+external resources, and "publish" what they see for other components to consume; others watch what another publishes and
+act on it (such as the resource translator watches what the providers publish, and then publishes its own results that
+are watched by another component).  Some of these internally published results are consumed by multiple components.
+
+To facilitate this communication use the [watchable][] library.  The `watchable.Map` type is very similar to the
+standard library's `sync.Map` type, but supports a `.Subscribe` (and `.SubscribeSubset`) method that promotes a pub/sub
+pattern.
+
+### pub
+
+Many of the things we communicate around are naturally named, either by a bare "name" string or by a "name"/"namespace"
+tuple.  And because `watchable.Map` is typed, it makes sense to have one map for each type of thing (very similar to if
+we were using native Go `map`s).  For example, a struct that might be written to by the Kubernetes provider, and read by
+the IR translator:
+
+   ```go
+   type ResourceTable struct {
+       // gateway classes are cluster-scoped; no namespace
+       GatewayClasses watchable.Map[string, *gwapiv1b1.GatewayClass]
+
+       // gateways are namespace-scoped, so use a k8s.io/apimachinery/pkg/types.NamespacedName as the map key.
+       Gateways watchable.Map[types.NamespacedName, *gwapiv1b1.Gateway]
+
+       HTTPRoutes watchable.Map[types.NamespacedName, *gwapiv1b1.HTTPRoute]
+   }
+   ```
+
+The Kubernetes provider updates the table by calling `table.Thing.Store(name, val)` and `table.Thing.Delete(name)`;
+updating a map key with a value that is deep-equal (usually `reflect.DeepEqual`, but you can implement your own `.Equal`
+method) the current value is a no-op; it won't trigger an event for subscribers.  This is handy so that the publisher
+doesn't have as much state to keep track of; it doesn't need to know "did I already publish this thing", it can just
+`.Store` its data and `watchable` will do the right thing.
+
+### sub
+
+Meanwhile, the translator and other interested components subscribe to it with `table.Thing.Subscribe` (or
+`table.Thing.SubscrubeSubset` if they only care about a few "Thing"s).  So the translator goroutine might look like:
+
+   ```go
+   func(ctx context.Context) error {
+       for snapshot := range k8sTable.HTTPRoutes.Subscribe(ctx) {
+           fullState := irInput{
+              GatewayClasses: k8sTable.GatewayClasses.LoadAll(),
+              Gateways:       k8sTable.Gateways.LoadAll(),
+              HTTPRoutes:     snapshot.State,
+           }
+           translate(irInput)
+       }
+   }
+   ```
+
+Or, to watch multiple maps in the same loop:
+
+   ```go
+   func worker(ctx context.Context) error {
+       classCh := k8sTable.GatewayClasses.Subscribe(ctx)
+       gwCh := k8sTable.Gateways.Subscribe(ctx)
+       routeCh := k8sTable.HTTPRoutes.Subscribe(ctx)
+       for ctx.Err() == nil {
+           var arg irInput
+           select {
+           case snapshot := <-classCh:
+               arg.GatewayClasses = snapshot.State
+           case snapshot := <-gwCh:
+               arg.Gateways = snapshot.State
+           case snapshot := <-routeCh:
+               arg.Routes = snapshot.State
+           }
+           if arg.GateWayClasses == nil {
+               arg.GatewayClasses = k8sTable.GateWayClasses.LoadAll()
+           }
+           if arg.GateWays == nil {
+               arg.Gateways = k8sTable.GateWays.LoadAll()
+           }
+           if arg.HTTPRoutes == nil {
+               arg.HTTPRoutes = k8sTable.HTTPRoutes.LoadAll()
+           }
+           translate(irInput)
+       }
+   }
+   ```
+
+From the updates it gets from `.Subscribe`, it can get a full view of the map being subscribed to via `snapshot.State`;
+but it must read the other maps explicitly.  Like `sync.Map`, `watchable.Map`s are thread-safe; while `.Subscribe` is a
+handy way to know when to run, `.Load` and friends can be used without subscribing.
+
+There can be any number of subscribers.  For that matter, there can be any number of publishers `.Store`ing things, but
+it's probably wise to just have one publisher for each map.
+
+The channel returned from `.Subscribe` is immediately readable with a snapshot of the map as it existed when
+`.Subscribe` was called; after that initial read it becomes readable again whenever `.Store` or `.Delete` mutates the
+map.  If multiple mutations happen between reads, they are coalesced in to one snapshot to be read; the `snapshot.State`
+is the most-recent full state, and `snapshot.Updates` is a listing of each of the mutations that cause this snapshot to
+be different than the last-read one.  This way subscribers don't need to worry about a backlog accumulating if they
+can't keep up with the rate of changes from the publisher.
+
+### other notes
+
+The common pattern will likely be that the entrypoint that launches the goroutines for each component instantiates the
+map, and passes them to the appropriate publishers and subscribers; same as if they were communicating via a dumb
+`chan`.
+
+A limitation of `watchable.Map` is that in order to ensure safety between goroutines, it does require that value types
+be deep-copiable; either by having a `DeepCopy` method, being a `proto.Message`, or by containing no reference types and
+so can be deep-copied by naive assignment.  Fortunately, we're using `controller-gen` anyway, and `controller-gen` can
+generate `DeepCopy` methods for us: just stick a `// +k8s:deepcopy-gen=true` on the types that you want it to generate
+methods for.
+
+[watchable]: https://pkg.go.dev/github.com/telepresenceio/watchable

go.mod

@@ -5,11 +5,12 @@ go 1.18
 require (
 	github.com/envoyproxy/go-control-plane v0.10.3
 	github.com/go-logr/zapr v1.2.0
-	github.com/google/go-cmp v0.5.7
+	github.com/google/go-cmp v0.5.8
 	github.com/spf13/cobra v1.4.0
-	github.com/stretchr/testify v1.7.1
+	github.com/stretchr/testify v1.8.0
+	github.com/telepresenceio/watchable v0.0.0-20220726211108-9bb86f92afa7
 	go.uber.org/zap v1.19.1
-	google.golang.org/grpc v1.45.0
+	google.golang.org/grpc v1.46.2
 	k8s.io/api v0.24.2
 	k8s.io/apimachinery v0.24.2
 	k8s.io/client-go v0.24.2
@@ -65,18 +66,18 @@ require (
 	github.com/spf13/pflag v1.0.5 // indirect
 	go.uber.org/atomic v1.7.0 // indirect
 	go.uber.org/multierr v1.6.0 // indirect
-	golang.org/x/net v0.0.0-20220127200216-cd36cc0744dd // indirect
+	golang.org/x/net v0.0.0-20220526153639-5463443f8c37 // indirect
 	golang.org/x/oauth2 v0.0.0-20211104180415-d3ed0bb246c8 // indirect
-	golang.org/x/sys v0.0.0-20220209214540-3681064d5158 // indirect
+	golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a // indirect
 	golang.org/x/term v0.0.0-20210927222741-03fcf44c2211 // indirect
-	golang.org/x/text v0.3.7 // indirect
+	golang.org/x/text v0.3.8-0.20211105212822-18b340fc7af2 // indirect
 	golang.org/x/time v0.0.0-20220210224613-90d013bbcef8 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.2.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
-	google.golang.org/genproto v0.0.0-20220329172620-7be39ac1afc7 // indirect
+	google.golang.org/genproto v0.0.0-20220505152158-f39f71e6c8f3 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
-	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/apiextensions-apiserver v0.24.2 // indirect
 	k8s.io/component-base v0.24.2 // indirect
 	k8s.io/klog/v2 v2.60.1 // indirect