Advice how to handle Multi with Kubernetes Client

[kjq]

I have a question on how to handle this scenario from a reactive RESTful endpoint. Usually I can kludge my way through but this one is stumping me.

What I am trying to do is:

1. Get a multi of items from a DB
2. Transform each item to another item
3. Call the Kubernetes client and get a list of deployments per item by tag for each item
4. Return the transformed results as a list

I know this code is completely wrong (there are all kinds of things I was trying to do and now it is just in a complete state of chaos) but it sort of demonstrates what I would like to do:

public Uni<List<InstanceInfo>> getStatus(@BeanParam final ClusterParams params) {
    return tenantClient.getClient(params)
        .chain(c -> this.clusterService.getInstances(params.getInstanceId(), c)
            .map(i -> clusterMapper.mapInstanceResponse(i))
            .invoke(r -> {
                this.getDeploymentsByTag(r.getTag(), r.getClient()).forEach(d -> {
                    final Deployment deployment = d.get();
                    final boolean ready = Readiness.isDeploymentReady(deployment);
                    final InstanceInfo instance = new InstanceInfo();
                    instance.setDomain(r.getDomain());
                    instance.setName(deployment.getMetadata().getName());
                    instance.setReady(ready);
                    r.getInstances().add(instance);
                });
            })
            .runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
            .onItem().transformToMultiAndMerge(i -> Multi.createFrom().iterable(i.getInstances()))
            .collect().asList());
}

Stream<RollableScalableResource<Deployment>> getDeploymentsByTag(final String tag, final String client) {
    return kubernetesClient.apps().deployments()
        .inNamespace(client)
        .withLabel(TAG_LABEL, tag)
        .resources();
}

This is the test I am writing against it if it helps give more context.

@Test
public void should_get_status() {

    clusterInstance.setTag("my-tag-1");

    List<InstanceInfo> actual = given()
        .pathParam("client", "my-client")
        .pathParam("instance", "my-tag-1")
        .contentType(ContentType.JSON)
        .when()
        .get("/{client}/instances/{instance}/status")
        .then()
        .statusCode(200)
        .extract().body()
        .jsonPath().getList(".", InstanceInfo.class);
    assertNotNull(actual);
}

As a final note, I am sure I am doing this wrong (even when I had it working) because I get a blocked thread when running natively in Quarkus where locally my test does not throw any issues.

[kjq]

Adding more context, I am trying to rewrite this original method someone else wrote. It worked but didnt look right (and it only handled a single item not multiple):

    @GET
    @Blocking
    @Path("/{client}/instances/{instance}/status")
    @Produces(APPLICATION_JSON)
    public Uni<InstanceResponse> getStatus(@BeanParam final ClusterParams params) {
        // Can't do the kubectlClient lookup inside chained lambda call. It's fine
        // outside of it though
        final List<Deployment> deployments = kubernetesClient.apps().deployments()
            .inNamespace(params.getClient()).list().getItems();
        log.debug("Found {} deployments in {}", deployments.size(), params.getClient());

        return tenantClient.getClient(params)
            .chain(c -> this.clusterService.getInstance(params.getInstanceId(), c))
            .chain(i -> {
                checkDeploymentStatus(deployments, i);
                return Uni.createFrom().item(i);
            })
            .onFailure().retry().atMost(STATUS_MAX_TRIES)
            .map(i -> this.clusterMapper.mapInstanceResponse(i));
    }

[kjq]

This is my first stab at it. It works from the test case but when deployed/native I get a thread blocked on what appears to be the "list" from the Kubernetes client.

So i am confused how to handle the call to the Kubernetes client "list" of the deployments "reactively"

@GET
@Path("/{client}/instances/{instance}/status")
@Produces(APPLICATION_JSON)
public Uni<List<InstanceInfo>> getStatus(@BeanParam final ClusterParams params) {
    return tenantClient.getClient(params)
        .chain(c -> this.clusterService.getInstances(params.getInstanceId(), c)
            .map(i -> this.clusterMapper.mapInstanceResponse(i)).toUni()
            .chain(r -> Uni.createFrom().item(this.getDeploymentsByTag(r.getTag(), r.getClient())))
            .onItem().transformToMulti(d -> Multi.createFrom().iterable(d.getItems()))
            .map(d -> {
                final boolean ready = Readiness.isDeploymentReady(d);
                final InstanceInfo instance = new InstanceInfo();
                instance.setDomain(c.getUri());
                instance.setName(d.getMetadata().getName());
                instance.setReady(ready);
                return instance;
            })
            //.runSubscriptionOn(Infrastructure.getDefaultWorkerPool())
            .collect().asList());
}

DeploymentList getDeploymentsByTag(final String tag, final String client) {
    return kubernetesClient.apps().deployments()
        .inNamespace(client)
        .withLabel(TAG_LABEL, tag)
        .list();
}

I seem to have two issues. Without the runOnSubscription it blocks on the JVM also. I get these errors (without runOnSubscription):

[WARN] io.vertx.core.impl.VertxImpl (vert.x-eventloop-thread-3) You're already on a Vert.x context, are you sure you want to create a new Vertx instance?
[WARN] io.vertx.core.impl.BlockedThreadChecker (vertx-blocked-thread-checker) Thread Thread[vert.x-eventloop-thread-3,5,main] has been blocked for 2809 ms, time limit is 2000 ms: io.vertx.core.VertxException: Thread blocked
	at java.base@17.0.7/jdk.internal.misc.Unsafe.park(Native Method)
	at java.base@17.0.7/java.util.concurrent.locks.LockSupport.park(Unknown Source)
	at java.base@17.0.7/java.util.concurrent.CompletableFuture$Signaller.block(Unknown Source)
	at java.base@17.0.7/java.util.concurrent.ForkJoinPool.unmanagedBlock(Unknown Source)
	at java.base@17.0.7/java.util.concurrent.ForkJoinPool.managedBlock(Unknown Source)
	at java.base@17.0.7/java.util.concurrent.CompletableFuture.waitingGet(Unknown Source)
	at java.base@17.0.7/java.util.concurrent.CompletableFuture.get(Unknown Source)
	at io.fabric8.kubernetes.client.dsl.internal.OperationSupport.waitForResult(OperationSupport.java:502)
OperationSupport.java:502
	at io.fabric8.kubernetes.client.dsl.internal.BaseOperation.list(BaseOperation.java:420)
BaseOperation.java:420
	at io.fabric8.kubernetes.client.dsl.internal.BaseOperation.list(BaseOperation.java:388)
BaseOperation.java:388
	at io.fabric8.kubernetes.client.dsl.internal.BaseOperation.list(BaseOperation.java:92)

Looking at the way Stork/Kubernetes discovery does it - would this be the right approach with the emitter?

[kjq]

Here is my latest pass and it seems to work! Is this the right way or is there a simpler way to do this?

I could not get it without injecting Vertx (an i literally clipped it from the Stork/Kubernetes Discovery code).

@GET
@Path("/{client}/instances/{instance}/status")
@Produces(APPLICATION_JSON)
public Uni<List<InstanceInfo>> getStatus(@BeanParam final ClusterParams params) {
    return tenantClient.getClient(params)
        .chain(c -> this.clusterService.getInstances(params.getInstanceId(), c)
            .map(i -> this.clusterMapper.mapInstanceResponse(i)).toUni()
            .chain(r -> getDeployments(r.getTag(), r.getClient()))
            .onItem().transformToMulti(d -> Multi.createFrom().iterable(d.getItems()))
            .map(d -> {
                final boolean ready = Readiness.isDeploymentReady(d);
                final InstanceInfo instance = new InstanceInfo();
                instance.setDomain(c.getUri());
                instance.setName(d.getMetadata().getName());
                instance.setReady(ready);
                log.debug("Returning {} instance info", instance.getName());
                return instance;
            })
            .collect().asList());
}

private Uni<DeploymentList> getDeployments(final String tag, final String client) {
    return Uni.createFrom().emitter(emitter -> {
        vertx.executeBlocking(future -> {
            final DeploymentList list = kubernetesClient.apps().deployments()
                .inNamespace(client)
                .withLabel(TAG_LABEL, tag)
                .list();
            future.complete(list);
        }, result -> {
            if (result.succeeded()) {
                final DeploymentList list = (DeploymentList)result.result();
                emitter.complete(list);
            }
            else {
                emitter.fail(result.cause());
            }
        });
    });
}