Add federated service watcher

[adleong]

We add support for federated services to the destination controller by adding a new FederatedServiceWatcher. When the destination controller receives a Get request for a Service with the multicluster.linkerd.io/remote-discovery and/or the multicluster.linkerd.io/local-discovery annotations, it subscribes to the FederatedServiceWatcher instead of subscribing to the EndpointsWatcher directly. The FederatedServiceWatcher watches the federated service for any changes to these annotations, and maintains the appropriate watches on the local EndpointWatcher and/or remote EndpointWatchers fetched through the ClusterStore.

This means that we will often have multiple EndpointTranslators writing to the same Get response stream. In order for a NoEndpoints message sent to one EndpointTranslator to not clobber the whole stream, we make a change where NoEndpoints messages are no longer sent to the response stream, but are replaced by a Remove message containing all of the addresses from that EndpointTranslator. This allows multiple EndpointTranslators to coexist on the same stream.

[zaharidichev]

I think there are two problems with this:

1. You are modifying a value in one go-routing and reading it in another. This value is not an atomic and is not guarded by a mutex so this is potentially racey, no ?
2. Simply flipping the bool will not always work. Imagine you are blocked on waiting for update := <-s.ch and you want to stop the stream. You call stop but a value is never received through the update chan. You will hang forever, no ?

My advise is to implement the standard mechanism of supplying a closed chan and selecting over both the updates and the closed chan. This will ensure that things are thread-safe.

for {
 select {
   case <-stopChan:
	 // we are done
	 return
   case update := <-s.ch:
	 // send update
   }
}