使用Kubernetes API进行分页和异步调用

1. Introduction

In this tutorial, we continue to explore the Kubernetes API for Java. This time, we’ll focus on two of its features: paging and asynchronous calls.

2. Paging

In a nutshell, paging allows us to iterate over a large result set in chunks, a.k.a pages – hence the name of this method. In the context of the Kubernetes Java API, this feature is available in all methods that return a list of resources. Those methods always include two optional parameters that we can use to iterate over the results:

limit: maximum number of items returned in a single API call
continue: A continuation token that tells the server the starting point for the returned result set

Using those parameters, we can iterate over an arbitrary number of items without putting too much pressure on the server. Even better, the amount of memory required on the client-side to hold the results is also bounded.

Now, let’s see how to use those parameters to get a list of all available pods in a cluster using this method:

ApiClient client = Config.defaultClient();
CoreV1Api api = new CoreV1Api(client);
String continuationToken = null;
do {
    V1PodList items = api.listPodForAllNamespaces(
      null,
      continuationToken, 
      null,
      null, 
      2, 
      null, 
      null,
      null,
      10,
      false);
    continuationToken = items.getMetadata().getContinue();
    items.getItems()
      .stream()
      .forEach((node) -> System.out.println(node.getMetadata()));
} while (continuationToken != null);

Here, the second parameter to the listPodForAllNamespaces() API call contains the continuation token, and the fifth is the limit parameter. While the limit is usually just a fixed value, continue requires a little extra effort.

For the first call, we send a null value, signaling the server that this is the first call of paged request sequence. Upon receiving the response, we get the new value for the next to continue value to use from the corresponding list metadata field.

This value will be null when there are no more results available, so we use this fact to define the exit condition for the iteration loop.

2.1. Pagination Gotchas

The paging mechanism is quite straightforward, but there are a few details we must keep in mind:

Currently, the API does not support server-side sorting. Given the current lack of storage-level support for sorting, this is unlikely to change anytime soon
All call parameters, except for continue, must be the same between calls
The continue value must be treated as an opaque handle. We should never make any assumptions about its value
Iteration is one-way*.* We cannot go back in the result set using a previously received continue token
Even though the returned list metadata contains a remainingItemCount field, its value is neither reliable nor supported by all implementations

2.2. List Data Consistency

Since a Kubernetes cluster is a very dynamic environment, there’s a possibility that the result set associated with a paginated call sequence gets modified while being read by the client. How does the Kubernetes API behave in this case?

As explained in Kubernetes documentation, list APIs support a resourceVersion parameter which, together with resourceVersionMatch, define how a particular version is selected for inclusion. However, for the paged result set case, the behavior is always the same: “Continue Token, Exact”.

This means that the returned resource versions corresponded to those available when the paginated list call started. While this approach provides consistency, it will not include results modified afterward. For instance, by the time we finish iterating over all pods in a large cluster, some of them may already have terminated.

3. Async Calls

So far, we’ve used the Kubernetes API in a synchronous way, which is fine for simple programs but not very efficient from a resource usage viewpoint, as it blocks the calling thread until we receive a response from the cluster and process it. This behavior will hurt the responsiveness of an application badly if, for instance, we start to make those calls in a GUI thread.

Fortunately, the library supports an asynchronous mode based on callbacks, which returns the control to the caller immediately.

Inspecting the CoreV1Api class, we’ll notice that, for each synchronous xxx() method, there’s also a xxxAsync() variant. For example, the async method for listPodForAllNamespaces() is listPodForAllNamespacesAsync(). The arguments are the same, with the addition of an extra parameter for the callback implementation.

3.1. Callback Details

The callback parameter object must implement the generic interface ApiCallback, which contains just four methods:

onSuccess: Called if and only if the call succeeded. The first argument type is the same that would be returned by the synchronous version
onFailure: Called there was an error calling the server or the reply contains an error code
onUploadProgress: Called during an upload. We can use this callback to provide feedback to a user during a lengthy operation
onDownloadProgress: Same as onUploadProgress, but for downloads

Async calls also don’t return a regular result. Instead, they return an OkHttp’s (the underlying REST client used by the Kubernetes API) Call instance, which works as a handle to the undergoing call. We can use this object to poll the completion state or, if we want, cancel it before completion.

3.2. Async Call Example

As we can imagine, implementing callbacks everywhere requires a lot of boilerplate code. To avoid this, we’ll use an invocation helper that simplifies this task a bit:

// Start async call
CompletableFuture<V1NodeList> p = AsyncHelper.doAsync(api,(capi,cb) ->
  capi.listNodeAsync(null, null, null, null, null, null, null, null, 10, false, cb)
);
p.thenAcceptAsync((nodeList) -> {
    nodeList.getItems()
      .stream()
      .forEach((node) -> System.out.println(node.getMetadata()));
});
// ... do something useful while we wait for results

Here, the helper wraps the asynchronous call invocation and adapts it to a more standard CompletableFuture. This allows us to use it with other libraries, such as those from the Reactor Project. In this example, we’ve added a completion stage that prints all metadata to the standard output.

As usual, when dealing with futures, we must be aware of concurrency issues that may arise. The online version of this code contains some debugging logs that clearly show that, even for this simple code, at least three threads were used:

The main thread*,* which kicks the async call
OkHttp’s threads used to make the actual HTTP call
The completion thread, where the results are processed

4. Conclusion

In this article, we have seen how to use paging and asynchronous calls with the Kubernetes Java API.

As usual, the full source code of the examples can be found over on GitHub.

Persistence

REST

Security