Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat(k8s): implement wildcard All Namespaces discovery #725

Open
wants to merge 5 commits into
base: main
Choose a base branch
from
Open

feat(k8s): implement wildcard All Namespaces discovery #725

Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
2e67e53
feat(k8s): implement wildcard All Namespaces discovery
andrewazores Nov 26, 2024
908deda
refactor
andrewazores Nov 26, 2024
621538e
log
andrewazores Dec 4, 2024
41a4d4c
forced resync on all namespaces
andrewazores Dec 4, 2024
2169300
refactor
andrewazores Dec 4, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
107 changes: 75 additions & 32 deletions src/main/java/io/cryostat/discovery/KubeApiDiscovery.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,6 +32,7 @@
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import javax.management.remote.JMXServiceURL;

Expand Down Expand Up @@ -70,6 +71,7 @@
@ApplicationScoped
public class KubeApiDiscovery implements ResourceEventHandler<Endpoints> {

private static final String ALL_NAMESPACES = "*";
private static final String NAMESPACE_QUERY_ADDR = "NS_QUERY";
private static final String ENDPOINTS_DISCOVERY_ADDR = "ENDPOINTS_DISC";

Expand Down Expand Up @@ -108,30 +110,42 @@ public class KubeApiDiscovery implements ResourceEventHandler<Endpoints> {
@Override
protected HashMap<String, SharedIndexInformer<Endpoints>> initialize()
throws ConcurrentException {
// TODO: add support for some wildcard indicating a single Informer for any
// namespace that Cryostat has permissions to. This will need some restructuring
// of how the namespaces within the discovery tree are mapped.
var result = new HashMap<String, SharedIndexInformer<Endpoints>>();
kubeConfig
.getWatchNamespaces()
.forEach(
ns -> {
result.put(
ns,
client.endpoints()
.inNamespace(ns)
.inform(
KubeApiDiscovery.this,
informerResyncPeriod.toMillis()));
logger.debugv(
"Started Endpoints SharedInformer for namespace"
+ " \"{0}\" with resync period {1}",
ns, informerResyncPeriod);
});
if (watchAllNamespaces()) {
result.put(
ALL_NAMESPACES,
client.endpoints()
.inAnyNamespace()
.inform(
KubeApiDiscovery.this,
informerResyncPeriod.toMillis()));
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe, add a logger.debugv here?

 logger.debugv(
        "Started Endpoints SharedInformer for all namespace with resync period {0}",
        informerResyncPeriod);

} else {
kubeConfig
.getWatchNamespaces()
.forEach(
ns -> {
result.put(
ns,
client.endpoints()
.inNamespace(ns)
.inform(
KubeApiDiscovery.this,
informerResyncPeriod
.toMillis()));
logger.debugv(
"Started Endpoints SharedInformer for namespace"
+ " \"{0}\" with resync period {1}",
ns, informerResyncPeriod);
});
}
return result;
}
};

private boolean watchAllNamespaces() {
return kubeConfig.getWatchNamespaces().stream().anyMatch(ns -> ALL_NAMESPACES.equals(ns));
}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: make sense to bring this method and ALL_NAMESPACES into KubeConfig class?


void onStart(@Observes StartupEvent evt) {
if (!enabled()) {
return;
Expand All @@ -144,18 +158,26 @@ void onStart(@Observes StartupEvent evt) {

logger.debugv("Starting {0} client", REALM);
safeGetInformers();
resyncWorker.scheduleAtFixedRate(
() -> {
try {
logger.debug("Resyncing");
notify(NamespaceQueryEvent.from(kubeConfig.getWatchNamespaces()));
} catch (Exception e) {
logger.warn(e);
}
},
0,
informerResyncPeriod.toMillis(),
TimeUnit.MILLISECONDS);
// TODO we should not need to force manual re-syncs this way - the Informer is already
// supposed to resync itself.
Comment on lines +160 to +161
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just curious here: Is it now safe to remove this manual resync? If not, in the case of all namespaces, we should also list namespaces and do resync too?

if (!watchAllNamespaces()) {
resyncWorker.scheduleAtFixedRate(
() -> {
try {
logger.debug("Resyncing");
notify(
NamespaceQueryEvent.from(
kubeConfig.getWatchNamespaces().stream()
.filter(ns -> !ALL_NAMESPACES.equals(ns))
.toList()));
} catch (Exception e) {
logger.warn(e);
}
},
0,
informerResyncPeriod.toMillis(),
TimeUnit.MILLISECONDS);
}
}

void onStop(@Observes ShutdownEvent evt) {
Expand Down Expand Up @@ -226,6 +248,15 @@ List<TargetTuple> tuplesFromEndpoints(Endpoints endpoints) {
for (EndpointPort port : subset.getPorts()) {
for (EndpointAddress addr : subset.getAddresses()) {
var ref = addr.getTargetRef();
if (ref == null) {
logger.debugv(
"Endpoints object {0} in {1} with address {2} had a null"
+ " targetRef",
endpoints.getMetadata().getName(),
endpoints.getMetadata().getNamespace(),
addr.getIp());
continue;
}
tts.add(
new TargetTuple(
ref,
Expand Down Expand Up @@ -295,8 +326,20 @@ public void handleQueryEvent(NamespaceQueryEvent evt) {
persistedTargets.add(node.target);
}

Stream<Endpoints> endpoints;
if (watchAllNamespaces()) {
endpoints =
safeGetInformers().get(ALL_NAMESPACES).getStore().list().stream()
.filter(
ep ->
Objects.equals(
ep.getMetadata().getNamespace(),
namespace));
} else {
endpoints = safeGetInformers().get(namespace).getStore().list().stream();
}
Set<Target> observedTargets =
safeGetInformers().get(namespace).getStore().list().stream()
endpoints
.map((endpoint) -> getTargetTuplesFrom(endpoint))
.flatMap(List::stream)
.filter((tuple) -> Objects.nonNull(tuple.objRef))
Expand Down
Loading