feat(k8s): watch V1Endpoints in KubeDiscoveryService

[shruthi713]

What

Adds an opt-in Kubernetes Endpoints watcher to KubeDiscoveryService, alongside the existing Service watcher. When enabled, discovery yields a Discoverable per ready pod IP instead of a single ClusterIP.

Why

Service-level discovery resolves to one virtual IP and leaves kube-proxy to pick a backend. That's correct for load balancing, and wrong for a caller that needs to address one specific pod and keep addressing that same pod — there is no way to express "this one" through a ClusterIP.

Watching Endpoints gives the caller the individual backing addresses, updated as pods become ready or terminate, with no DNS polling interval to wait out.

Design

• Off by default. The watcher only starts when a caller invokes enableEndpointsWatcher(). Existing discovery behaviour is byte-for-byte unchanged for everyone who doesn't.
• Both watchers coexist. Enabling the endpoints watcher doesn't replace the service watcher — callers wanting the ClusterIP keep getting it.
• Late enablement is supported. enableEndpointsWatcher() starts a thread immediately if services are already registered; otherwise the next discover() call starts it. This matters because callers typically enable before their first discover().
• toDiscoverables(String, V1Endpoints) is package-private and @VisibleForTesting — it's the pure mapping function from an Endpoints object to the discoverable set, and the natural unit-test seam.

Note for reviewers: the discover() restructuring

The control flow in discover() changed more than the diff makes obvious, so calling it out explicitly.

The original returned early from inside the service-watcher creation branch:

if (watcherThread == null) {
  synchronized (this) {
    watcherThread = this.watcherThread;      // re-read under lock
    if (watcherThread == null) {
      ... create, addService(name), start ...
      return serviceDiscovered;              // early return
    }
  }
}
watcherThread.addService(name);              // everything else falls through
return serviceDiscovered;

That early return can't survive, because the endpoints block now has to run afterwards. It's replaced by explicit else branches on both the outer check and the double-checked inner one.

The inner else is the load-bearing part. A caller that loses the creation race sees a non-null thread when it re-reads under the lock, and still has to register its own service name — behaviour the original got for free by falling through to the unconditional addService. Without it, a service can be silently omitted from the watcher's label selector and its ServiceDiscovered never populates.

RBAC prerequisite

The watcher needs get/list/watch on endpoints in the CDAP role. Without the grant the watcher logs a permission failure and discovery continues returning Service IPs — so this is a prerequisite for the feature, not for the deployment, and this PR is safe to merge ahead of the role change.

Testing

cdap-kubernetes: 65 tests, 0 failures, 0 checkstyle violations.

Exercised end to end on a live GKE cluster. The proxy that consumes this API logs:

Successfully enabled Kubernetes Endpoints watcher on DiscoveryServiceClient
Start watching for changes in kubernetes resource type class ...V1Endpoints

and resolves individual task worker pod addresses rather than the service ClusterIP.

Note

toDiscoverables has no dedicated unit test yet. It's pure and trivially testable, and I'll follow up with coverage — flagging it rather than letting a reviewer find it.

[gemini-code-assist]

Code Review

This pull request introduces a real-time Kubernetes endpoints watching mechanism in KubeDiscoveryService via a new EndpointsWatcherThread and an enableEndpointsWatcher method. Feedback on these changes highlights a potential bug where checking !discoverables.isEmpty() before updating discoverables can lead to stale data when all pods become unready. Additionally, it is suggested to overload createDiscoverable to accept an integer port directly, avoiding unnecessary object allocation in a loop.

[gemini-code-assist]

The if (!discoverables.isEmpty()) check can lead to stale data. If a service's endpoints are updated such that there are no longer any ready pods, toDiscoverables will correctly return an empty set. However, this check prevents setDiscoverables from being called with the empty set, so consumers will continue to see the old, now-stale list of pod IPs.

To ensure consumers always have the most up-to-date information, you should always update the discoverables, even if the new set is empty.

            s.setDiscoverables(discoverables);

[gemini-code-assist]

The creation of a new V1ServicePort instance here (new V1ServicePort().port(port.getPort())) just to pass the port number to createDiscoverable is a bit indirect and adds a small object creation overhead in a loop.

To make the code cleaner and more direct, consider overloading the createDiscoverable method to accept an Integer port directly. The existing createDiscoverable that takes a V1ServicePort can then call this new overload.

          Discoverable d = createDiscoverable(name, address.getIp(), port.getPort(), payload);