+ * This resolver watches for changes in Kubernetes EndpointSlices and updates + * the gRPC client with the resolved addresses. + *
+ *+ * The target URI for this resolver is parsed by {@link ResolverTarget}, which + * supports the following formats: + *
+ * If the namespace is not provided in the URI, the resolver will attempt to read + * the current pod's namespace from the mounted file at + * {@code /var/run/secrets/kubernetes.io/serviceaccount/namespace}. If this file + * is not found or cannot be read, the resolver will default to using the + * {@code default} namespace. + *
+ *+ * If the port is not provided in the URI, the resolver will use any of the ports + * found in the EndpointSlice. If a port name is provided (e.g., + * {@code kubernetes:///myservice:grpc}), the resolver will look for a port with + * that name in the EndpointSlice. If a numerical port is provided, that port + * will be used directly. + *
+ */ public final class KubernetesNameResolver extends NameResolver { private static final Logger LOGGER = Logger.getLogger(KubernetesNameResolver.class.getName()); @@ -41,13 +82,26 @@ public final class KubernetesNameResolver extends NameResolver { private boolean defaultExecutorUsed = false; private Listener listener; + /** + * Creates a new {@link KubernetesNameResolver} with a default single-threaded + * executor. + * + * @param params the target parameters for the resolver + * @throws IOException if an error occurs while initializing the watcher + */ public KubernetesNameResolver(ResolverTarget params) throws IOException { this(Executors.newSingleThreadExecutor(), params); this.defaultExecutorUsed = true; } - public KubernetesNameResolver(Executor executor, ResolverTarget params) - throws IOException { + /** + * Creates a new {@link KubernetesNameResolver} with a custom executor. + * + * @param executor the executor to use for background tasks + * @param params the target parameters for the resolver + * @throws IOException if an error occurs while initializing the watcher + */ + public KubernetesNameResolver(Executor executor, ResolverTarget params) throws IOException { this.executor = executor; this.params = params; if (params.namespace() != null) { @@ -57,12 +111,22 @@ public KubernetesNameResolver(Executor executor, ResolverTarget params) } } + /** + * Starts the name resolution process. + * + * @param listener the listener to notify when addresses are resolved or errors + * occur + */ @Override public void start(Listener listener) { this.listener = listener; resolve(); } + /** + * Refreshes the name resolution process. This method is called when the gRPC + * client requests a refresh. + */ @Override public void refresh() { if (semaphore.tryAcquire()) { @@ -70,10 +134,17 @@ public void refresh() { } } + /** + * Resolves the Kubernetes service by watching EndpointSlices. + */ private void resolve() { executor.execute(this::watch); } + /** + * Watches for changes in EndpointSlices and updates the listener with resolved + * addresses. + */ private void watch() { watcher.watch(params.service(), new EndpointSliceWatcher.Subscriber() { @Override @@ -81,13 +152,13 @@ public void onEvent(Event event) { // watch event occurred if (!SUPPORTED_KUBERNETES_EVENTS.contains(event.type())) { LOGGER.log(Level.FINER, "Unsupported Kubernetes event type {0}", - new Object[] { event.type().toString() }); + new Object[]{event.type().toString()}); return; } if (event.type().equals(EventType.DELETED)) { LOGGER.log(Level.FINE, "EndpointSlice {0} was deleted", - new Object[] { event.endpointSlice().metadata().name() }); + new Object[]{event.endpointSlice().metadata().name()}); return; } @@ -96,10 +167,10 @@ public void onEvent(Event event) { return; } - LOGGER.log(Level.FINER, "Resolving addresses for service {0}", new Object[] { params.service() }); + LOGGER.log(Level.FINER, "Resolving addresses for service {0}", new Object[]{params.service()}); buildAddresses(event.endpointSlice()).ifPresentOrElse(a -> listener.onAddresses(a, Attributes.EMPTY), () -> LOGGER.log(Level.FINE, "No usable addresses found for Kubernetes service {0}", - new Object[] { params.service() })); + new Object[]{params.service()})); } @Override @@ -120,6 +191,9 @@ public void onCompleted() { }); } + /** + * Shuts down the resolver and releases resources. + */ @Override public void shutdown() { if (defaultExecutorUsed && executor instanceof ExecutorService executor) { @@ -127,11 +201,23 @@ public void shutdown() { } } + /** + * Returns the authority of the service being resolved. + * + * @return an empty string as this resolver does not use service authority + */ @Override public String getServiceAuthority() { return ""; } + /** + * Builds a list of gRPC {@link EquivalentAddressGroup} from the given + * {@link EndpointSlice}. + * + * @param endpointSlice the EndpointSlice to process + * @return an optional list of resolved addresses + */ private Optional+ * This provider supports target URIs as defined by {@link ResolverTarget}, including: + *
+ * If the namespace is not explicitly provided in the URI, the underlying + * {@link KubernetesNameResolver} will first attempt to read the current pod's + * namespace from the mounted file at + * {@code /var/run/secrets/kubernetes.io/serviceaccount/namespace}. If this file + * is not found or cannot be read, it will default to using the {@code default} + * namespace. + *
+ *+ * If the port is not specified in the URI, the resolver will use any of the ports + * defined in the Kubernetes EndpointSlice. Alternatively, a port can be specified + * by its name in the URI (e.g., {@code kubernetes:///myservice:grpc}), in which + * case the resolver will look for an EndpointSlice port with that name. If a + * numerical port is provided, that port will be used. + *
+ */ public class KubernetesNameResolverProvider extends NameResolverProvider { private String scheme = "kubernetes"; + /** + * Constructs a new provider with a custom scheme. + * + * @param schema the URI scheme this provider should support (e.g., "kubernetes") + */ public KubernetesNameResolverProvider(String schema) { this.scheme = schema; } + /** + * Constructs a new provider with the default scheme ("kubernetes"). + */ public KubernetesNameResolverProvider() { } + /** + * Indicates whether this provider is available for use. + * + * @return always returns {@code true} + */ @Override protected boolean isAvailable() { return true; } + /** + * Returns the priority of this provider. + * + * @return the priority value (5) + */ @Override protected int priority() { return 5; } + /** + * Returns a new {@link NameResolver} for the given target URI and arguments, if the URI scheme matches. + * + * @param targetUri the URI to resolve + * @param args the resolver arguments + * @return a new {@link KubernetesNameResolver}, or {@code null} if the scheme does not match + */ @Override public NameResolver newNameResolver(URI targetUri, Args args) { if (targetUri.getScheme().equals(this.scheme)) { @@ -38,6 +99,14 @@ public NameResolver newNameResolver(URI targetUri, Args args) { return null; } + /** + * Builds a {@link KubernetesNameResolver} using the provided executor and target parameters. + * + * @param executor the executor for offloading tasks + * @param params the parsed target parameters + * @return a new {@link KubernetesNameResolver} + * @throws RuntimeException if an I/O error occurs + */ private NameResolver buildResolver(Executor executor, ResolverTarget params) { try { if (executor != null) { @@ -49,6 +118,11 @@ private NameResolver buildResolver(Executor executor, ResolverTarget params) { } } + /** + * Returns the default scheme supported by this provider. + * + * @return the default scheme + */ @Override public String getDefaultScheme() { return this.scheme; diff --git a/lib/src/main/java/io/github/lothar1998/kuberesolver/ResolverTarget.java b/lib/src/main/java/io/github/lothar1998/kuberesolver/ResolverTarget.java index 9d7a8b2..c50697c 100644 --- a/lib/src/main/java/io/github/lothar1998/kuberesolver/ResolverTarget.java +++ b/lib/src/main/java/io/github/lothar1998/kuberesolver/ResolverTarget.java @@ -4,10 +4,38 @@ import javax.annotation.Nullable; import javax.annotation.Nonnull; +/** + * Represents a parsed Kubernetes target including service name, namespace, and optional port. + * Service name is always required but namespace and port are optional. + * Used by {@link KubernetesNameResolver} to extract target information from a URI. + */ public record ResolverTarget(@Nullable String namespace, @Nonnull String service, @Nullable String port) { + /** + * Parses a {@link URI} into a {@link ResolverTarget}, extracting the service, namespace, and port. + * Supports formats like: + *+ * Implementations must provide the request and HTTP client logic appropriate for secure or insecure access. + */ public abstract sealed class EndpointSliceWatcher permits InsecureEndpointSliceWatcher, SecureEndpointSliceWatcher { private static final String KUBERNETES_WATCH_ENDPOINT_SLICES_URL_PATTERN = "%s/apis/discovery.k8s.io/v1/watch/namespaces/%s/endpointslices?labelSelector=kubernetes.io/service-name=%s"; @@ -24,11 +31,25 @@ public abstract sealed class EndpointSliceWatcher permits InsecureEndpointSliceW private final String host; private final String namespace; + /** + * Constructs a new watcher for a given Kubernetes API server and namespace. + * + * @param host the base URL of the Kubernetes API server + * @param namespace the Kubernetes namespace to watch for endpoint slices + */ public EndpointSliceWatcher(String host, String namespace) { this.host = host; this.namespace = namespace; } + /** + * Starts watching for EndpointSlice events associated with a given service name. + * Events are streamed and passed to the provided subscriber. + * + * @param serviceName the name of the Kubernetes service + * @param subscriber the subscriber that receives events, errors, and completion signals + * @throws UnexpectedStatusCodeException if the response status from the Kubernetes API is not 200 + */ public void watch(String serviceName, Subscriber subscriber) throws UnexpectedStatusCodeException { try { var request = getRequest(serviceName); @@ -55,24 +76,69 @@ public void watch(String serviceName, Subscriber subscriber) throws UnexpectedSt } } + /** + * Constructs an HTTP request to watch the EndpointSlices of the given service. + * + * @param serviceName the name of the Kubernetes service + * @return the constructed {@link HttpRequest} + * @throws Exception if an error occurs while constructing the request + */ protected abstract HttpRequest getRequest(String serviceName) throws Exception; + /** + * Returns the HTTP client used to make requests to the Kubernetes API. + * + * @return an {@link HttpClient} instance + * @throws Exception if the client cannot be constructed + */ protected abstract HttpClient getClient() throws Exception; + /** + * Constructs the full URI for watching EndpointSlices of the specified service. + * + * @param serviceName the name of the Kubernetes service + * @return the constructed {@link URI} + * @throws URISyntaxException if the URI is invalid + * @throws MalformedURLException if the URL is invalid + */ protected URI getURI(String serviceName) throws URISyntaxException, MalformedURLException { var url = new URL(String.format(KUBERNETES_WATCH_ENDPOINT_SLICES_URL_PATTERN, host, namespace, serviceName)); return url.toURI(); } + /** + * Callback interface for receiving streamed EndpointSlice watch events. + */ public interface Subscriber { + /** + * Called when a new EndpointSlice event is received. + * + * @param event the event data + */ void onEvent(Event event); + /** + * Called when an error occurs during watch processing. + * + * @param throwable the exception or error + */ void onError(Throwable throwable); + /** + * Called when the watch stream completes successfully. + */ void onCompleted(); } + /** + * Exception thrown when a non-200 HTTP response is received from the Kubernetes API. + */ public static class UnexpectedStatusCodeException extends RuntimeException { + /** + * Constructs the exception with a message describing the unexpected status code. + * + * @param message the error message + */ public UnexpectedStatusCodeException(String message) { super(message); } diff --git a/lib/src/main/java/io/github/lothar1998/kuberesolver/kubernetes/InClusterEndpointSliceWatcher.java b/lib/src/main/java/io/github/lothar1998/kuberesolver/kubernetes/InClusterEndpointSliceWatcher.java index 9e34637..27eb001 100644 --- a/lib/src/main/java/io/github/lothar1998/kuberesolver/kubernetes/InClusterEndpointSliceWatcher.java +++ b/lib/src/main/java/io/github/lothar1998/kuberesolver/kubernetes/InClusterEndpointSliceWatcher.java @@ -9,6 +9,16 @@ import java.nio.file.Paths; import java.util.Optional; +/** + * A Kubernetes EndpointSlice watcher that runs from within a Kubernetes cluster. + *
+ * This implementation reads in-cluster configuration including the API server host and port + * from environment variables, and authentication details (token, CA cert, and namespace) + * from service account files mounted in the pod. + *
+ * It is suitable for production usage within a Kubernetes cluster and assumes the standard + * in-cluster configuration paths and environment variables are present. + */ public final class InClusterEndpointSliceWatcher extends SecureEndpointSliceWatcher { private static final String KUBERNETES_SERVICE_HOST = "KUBERNETES_SERVICE_HOST"; @@ -18,14 +28,31 @@ public final class InClusterEndpointSliceWatcher extends SecureEndpointSliceWatc private static final String KUBERNETES_SERVICE_ACCOUNT_CA_CERT_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/ca.crt"; private static final String KUBERNETES_NAMESPACE_PATH = "/var/run/secrets/kubernetes.io/serviceaccount/namespace"; + /** + * Constructs the watcher by inferring the namespace from the in-cluster namespace file. + * + * @throws IOException if reading the namespace file fails + */ public InClusterEndpointSliceWatcher() throws IOException { this(getNamespace()); } + /** + * Constructs the watcher using the provided namespace and other in-cluster configuration. + * + * @param namespace the Kubernetes namespace to watch + */ public InClusterEndpointSliceWatcher(String namespace) { super(getHost(), namespace, getAuthConfigProvider()); } + /** + * Builds the Kubernetes API host URL using environment variables + * {@code KUBERNETES_SERVICE_HOST} and {@code KUBERNETES_SERVICE_PORT}. + * + * @return the complete Kubernetes API server URL + * @throws RuntimeException if required environment variables are not set + */ private static String getHost() { var address = Optional.of(System.getenv(KUBERNETES_SERVICE_HOST)).orElseThrow( () -> new RuntimeException(String.format("%s env variable not set", KUBERNETES_SERVICE_HOST))); @@ -36,6 +63,12 @@ private static String getHost() { return String.format("https://%s:%s", address, port); } + /** + * Reads the namespace from the in-cluster service account namespace file. + * + * @return the current namespace, or "default" if the file is not found + * @throws IOException if file reading fails + */ private static String getNamespace() throws IOException { try { return Files.readString(Paths.get(KUBERNETES_NAMESPACE_PATH), StandardCharsets.UTF_8); @@ -44,6 +77,11 @@ private static String getNamespace() throws IOException { } } + /** + * Provides the authentication configuration based on in-cluster service account files. + * + * @return an {@link AuthConfigProvider} instance that supplies CA certificate and token streams + */ private static AuthConfigProvider getAuthConfigProvider() { return new AuthConfigProvider() { diff --git a/lib/src/main/java/io/github/lothar1998/kuberesolver/kubernetes/InsecureEndpointSliceWatcher.java b/lib/src/main/java/io/github/lothar1998/kuberesolver/kubernetes/InsecureEndpointSliceWatcher.java index 9e4400d..404d9f7 100644 --- a/lib/src/main/java/io/github/lothar1998/kuberesolver/kubernetes/InsecureEndpointSliceWatcher.java +++ b/lib/src/main/java/io/github/lothar1998/kuberesolver/kubernetes/InsecureEndpointSliceWatcher.java @@ -4,12 +4,31 @@ import java.net.http.HttpRequest; import java.net.http.HttpClient.Version; +/** + * A watcher for Kubernetes EndpointSlices that communicates over HTTP without TLS (insecure). + * This class extends {@link EndpointSliceWatcher} and provides an {@link HttpClient} + * configured for HTTP/1.1 and basic JSON requests. + *
+ * It is intended for development or trusted network environments where secure communication
+ * is not required.
+ */
public final class InsecureEndpointSliceWatcher extends EndpointSliceWatcher {
+ /**
+ * Constructs an insecure EndpointSliceWatcher with the specified Kubernetes API host and namespace.
+ *
+ * @param host the hostname or IP of the Kubernetes API server
+ * @param namespace the Kubernetes namespace to watch for EndpointSlices
+ */
public InsecureEndpointSliceWatcher(String host, String namespace) {
super(host, namespace);
}
+ /**
+ * Creates an {@link HttpClient} configured to use HTTP/1.1 without TLS.
+ *
+ * @return an insecure HTTP/1.1 {@link HttpClient} instance
+ */
@Override
protected HttpClient getClient() {
return HttpClient.newBuilder()
@@ -17,6 +36,13 @@ protected HttpClient getClient() {
.build();
}
+ /**
+ * Builds an HTTP GET request to fetch EndpointSlice information for the given service name.
+ *
+ * @param serviceName the name of the Kubernetes service
+ * @return an {@link HttpRequest} configured for the service's EndpointSlice
+ * @throws Exception if URI creation fails
+ */
@Override
protected HttpRequest getRequest(String serviceName) throws Exception {
return HttpRequest.newBuilder(getURI(serviceName))
diff --git a/lib/src/main/java/io/github/lothar1998/kuberesolver/kubernetes/SecureEndpointSliceWatcher.java b/lib/src/main/java/io/github/lothar1998/kuberesolver/kubernetes/SecureEndpointSliceWatcher.java
index 3bf4fcd..86ebbbc 100644
--- a/lib/src/main/java/io/github/lothar1998/kuberesolver/kubernetes/SecureEndpointSliceWatcher.java
+++ b/lib/src/main/java/io/github/lothar1998/kuberesolver/kubernetes/SecureEndpointSliceWatcher.java
@@ -11,15 +11,36 @@
import javax.net.ssl.SSLContext;
import javax.net.ssl.TrustManagerFactory;
+/**
+ * A secure implementation of {@link EndpointSliceWatcher} that uses TLS and token-based authentication
+ * to communicate with the Kubernetes API server.
+ *
+ * This watcher sets up a custom {@link SSLContext} with a provided CA certificate and uses a bearer token
+ * for authorization headers. It is suitable for production environments requiring secure communication.
+ */
public sealed class SecureEndpointSliceWatcher extends EndpointSliceWatcher permits InClusterEndpointSliceWatcher {
private final AuthConfigProvider authConfig;
+ /**
+ * Constructs a SecureEndpointSliceWatcher with the specified Kubernetes API host, namespace, and authentication configuration.
+ *
+ * @param host the Kubernetes API host
+ * @param namespace the namespace to watch for EndpointSlices
+ * @param authConfig the provider for CA certificate and token used for authentication
+ */
public SecureEndpointSliceWatcher(String host, String namespace, AuthConfigProvider authConfig) {
super(host, namespace);
this.authConfig = authConfig;
}
+ /**
+ * Creates an {@link HttpClient} configured with a custom {@link SSLContext} that uses the CA certificate
+ * provided by the {@link AuthConfigProvider}.
+ *
+ * @return an HTTP client configured for secure communication with the Kubernetes API
+ * @throws Exception if SSL context setup fails
+ */
@Override
protected HttpClient getClient() throws Exception {
var cf = CertificateFactory.getInstance("X.509");
@@ -45,6 +66,14 @@ protected HttpClient getClient() throws Exception {
.build();
}
+ /**
+ * Builds a secure HTTP GET request with authorization and content-type headers to retrieve
+ * EndpointSlice information for a given service.
+ *
+ * @param serviceName the name of the Kubernetes service
+ * @return a configured {@link HttpRequest} instance
+ * @throws Exception if the request setup fails
+ */
@Override
protected HttpRequest getRequest(String serviceName) throws Exception {
return HttpRequest.newBuilder(getURI(serviceName))
@@ -54,6 +83,12 @@ protected HttpRequest getRequest(String serviceName) throws Exception {
.build();
}
+ /**
+ * Reads the bearer token from the input stream provided by the {@link AuthConfigProvider}.
+ *
+ * @return the token as a string
+ * @throws Exception if reading the token fails
+ */
private String getToken() throws Exception {
try (var token = authConfig.getToken()) {
var bytes = token.readAllBytes();
@@ -61,6 +96,9 @@ private String getToken() throws Exception {
}
}
+ /**
+ * Provider interface for authentication configuration including CA certificate and token stream.
+ */
public interface AuthConfigProvider {
InputStream getCaCert() throws Exception;
diff --git a/lib/src/test/java/io/github/lothar1998/kuberesolver/ResolverTargetTest.java b/lib/src/test/java/io/github/lothar1998/kuberesolver/ResolverTargetTest.java
index 5f4b195..982b33b 100644
--- a/lib/src/test/java/io/github/lothar1998/kuberesolver/ResolverTargetTest.java
+++ b/lib/src/test/java/io/github/lothar1998/kuberesolver/ResolverTargetTest.java
@@ -35,6 +35,8 @@ private static Stream