如何使用Fabric8 Client从K8s Pod读取增量日志/日志文件？

一、Pod容器日志的增量读取 #

你之前用sinceSeconds(10)的问题在于，每次调用都会拉取过去10秒的全部日志，而非从上一次读取的结束位置开始。要实现增量读取，核心是记录上次读取的时间戳，下次查询时只获取该时间点之后的新日志。

具体实现： #

1. 维护一个变量记录上次读取的结束时间（比如lastTimestamp），初始值设为当前时间。
2. 每次查询日志时，用sinceTime(lastTimestamp)替代sinceSeconds，确保只拉取新日志。
3. 读取完成后，更新lastTimestamp——如果日志行自带时间戳，建议解析最后一行的时间来更新（更精准）；如果没有，就用当前时间作为下次查询的起始点。

示例代码： #

import io.fabric8.kubernetes.client.KubernetesClient;
import io.fabric8.kubernetes.client.dsl.LogWatch;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.time.Instant;

public class PodLogIncrementalReader {
    private static Instant lastTimestamp = Instant.now();
    private final KubernetesClient client;
    private final String namespace;
    private final String podName;

    public PodLogIncrementalReader(KubernetesClient client, String namespace, String podName) {
        this.client = client;
        this.namespace = namespace;
        this.podName = podName;
    }

    public void fetchNewLogs() throws IOException {
        try (LogWatch logWatch = client.pods()
                .inNamespace(namespace)
                .withName(podName)
                .sinceTime(lastTimestamp)
                .watchLog()) {

            BufferedReader reader = new BufferedReader(new InputStreamReader(logWatch.getOutput()));
            String line;
            while ((line = reader.readLine()) != null) {
                System.out.println("> " + line);
                // 假设日志格式是 "2024-05-20T12:34:56.789Z [INFO] ..."，解析时间戳更新lastTimestamp
                // String logTime = line.split(" ")[0];
                // lastTimestamp = Instant.parse(logTime);
            }
            // 无法解析日志时间时，用当前时间作为下次起始点
            lastTimestamp = Instant.now();
        }
    }
}

二、Pod内指定日志文件的增量读取 #

直接调用file().read()会读取整个文件，必然重复。要实现增量读取，需要记录上次读取的文件字节偏移量，下次从该位置开始读取新内容。

具体实现： #

1. 维护一个变量记录上次读取的偏移量（比如lastOffset），初始值为0。
2. 每次读取前，先获取文件当前总大小，判断是否有新内容（总大小 > lastOffset）。
3. 通过exec执行dd命令从指定偏移量读取新内容，避免读取历史数据。
4. 读取完成后，更新lastOffset为当前文件总大小。

示例代码： #

import io.fabric8.kubernetes.client.KubernetesClient;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

public class PodFileIncrementalReader {
    private static long lastOffset = 0;
    private final KubernetesClient client;
    private final String namespace;
    private final String podName;
    private final String logFilePath;

    public PodFileIncrementalReader(KubernetesClient client, String namespace, String podName, String logFilePath) {
        this.client = client;
        this.namespace = namespace;
        this.podName = podName;
        this.logFilePath = logFilePath;
    }

    public void fetchNewFileContent() throws IOException {
        // 先获取文件当前总大小
        String sizeCmd = String.format("stat -c %%s %s", logFilePath);
        String sizeOutput = client.pods()
                .inNamespace(namespace)
                .withName(podName)
                .exec(sizeCmd)
                .getOutput();
        long currentSize = Long.parseLong(sizeOutput.trim());

        if (currentSize <= lastOffset) {
            return; // 没有新内容
        }

        // 从上次偏移量开始读取新内容
        String readCmd = String.format("dd if=%s skip=%d bs=1", logFilePath, lastOffset);
        try (InputStreamReader isr = new InputStreamReader(client.pods()
                .inNamespace(namespace)
                .withName(podName)
                .exec(readCmd)
                .getInputStream());
             BufferedReader reader = new BufferedReader(isr)) {

            String line;
            while ((line = reader.readLine()) != null) {
                System.out.println("> " + line);
            }
            lastOffset = currentSize;
        }
    }
}

额外注意： #

• 如果日志文件会被滚动（比如用logrotate），要额外检查文件inode是否变化——如果inode变了，说明文件被轮转，需要重置lastOffset为0，重新读取新文件。
• 如果Pod重启，容器日志的lastTimestamp需要重置，否则会漏掉新Pod的日志。

内容的提问来源于stack exchange，提问作者Akash Shinde