java - Kubernetes Watch - 如何更新？（Java API）

Question

我有两个在并行线程中运行的任务，我正在研究为什么该Watch功能不起作用。如果您有任何见解，请告诉我。

任务 1：获取 pod 的状态并显示当前状态。

任务 2：密切关注新事件。这就是我试图更好地理解的内容。

这些任务中的每一个都每 30 秒执行一次scheduledAtFixedRate()。

 

预期行为：

任务 1：我应该获取所有 pod 及其当前状态的列表（这可行）。

任务 2：我应该期待新事件发生时的列表。

 

观察到的行为：

任务1：它工作正常。我每 30 秒更新一次 Pod 的状态。

任务 2：它从第一个请求中转储事件，但它似乎没有更新任何新事件。

 

代码： 任务 1：

@Component
@Scope(value = org.springframework.beans.factory.config.ConfigurableBeanFactory.SCOPE_SINGLETON)
public class Task1 implements Runnable {

    private ScheduledExecutorService scheduledExecutorService;
    private CommandInvoker commandInvoker;
    private static final int INITIAL_DELAY = 15; 
    private static final int POLLING_INTERVAL = 30;

    @Autowired
    public Task1 (CommandInvoker commandInvoker,
                        ScheduledExecutorService scheduledExecutorService) {
        this.commandInvoker = commandInvoker;
        this.scheduledExecutorService = scheduledExecutorService;
        this.scheduledExecutorService.scheduleAtFixedRate(this, INITIAL_DELAY, POLLING_INTERVAL, TimeUnit.SECONDS);
    }

    @Override
    public void run() {
        System.out.println("===== STARTING TASK 1 POD HEALTH CHECK =======");
        commandInvoker.getPodStatus();
    }
}

任务 2：

@Component
@Scope(value = org.springframework.beans.factory.config.ConfigurableBeanFactory.SCOPE_SINGLETON)
public class Task2 implements Runnable {

    private ScheduledExecutorService scheduledExecutorService;
    private CommandInvoker commandInvoker;

    @Autowired
    public Task2(CommandInvoker commandInvoker,
                              ScheduledExecutorService scheduledExecutorService) {
        this.commandInvoker = commandInvoker;
        this.scheduledExecutorService = scheduledExecutorService;
        this.scheduledExecutorService.scheduleAtFixedRate(this, 30, 30, TimeUnit.SECONDS);
    }

    @Override
    public void run() {
        System.out.println("===== STARTING TASK 2 EVENT WATCH UPDATE =======");
        commandInvoker.getWatchUpdates();
    }
}

命令调用者：

@Component
public class CommandInvoker {

    public void getPodStatus() {
        try {
            CoreV1Api api = new CoreV1Api();
            V1PodList list = api.listPodForAllNamespaces(null,
                    null, null,  null, null, null, null, null, null);
            for( V1Pod pod : list.getItems() ) {
                  // THIS WORKS //
            }
        } catch ( ApiException e) {
            throw new WhateverException ("Failed to handle watchlist event", e);
        }
    }

    public void getWatchUpdates() {
        CoreV1Api api = new CoreV1Api();
        try {
            Watch<V1Event> watch = Watch.createWatch(
                    apiClient,
                    api.listEventForAllNamespacesCall(null, null, null, null,
                            null, null, null, null, true, null, null),
                    new TypeToken<Watch.Response<V1Event>>() {}.getType());

            watch.forEach( response  -> {
                V1Event event = response.object;
                // THIS ONLY DUMPS EVENTS FROM FIRST CALL BUT NEVER GETS EXECUTED AGAIN
            });
            // I NEVER REACH HERE BUT I DON'T GET ANY UPDATES
        } catch ( ApiException e) {
            throw new K8ServerException("Failed to handle watchlist event", e);
        }
    }
}

标签： javakuberneteskubernetes-health-check