Kubernetes 日志流中断后如何持续读取？

python

2024-03-03 03:08:04

从 Kubernetes 日志流的中断处恢复读取

问题：

在 Kubernetes 中使用日志流 URI 访问 Pod 日志时，如何从流中断处重新请求，直到 Pod 结束？

解决方案：

使用 Range 头

通过在请求头中传递 headers['Range'] = f'bytes={last_byte + 1}-'，可以从上次读取的字节位置开始恢复请求。其中，last_byte 是上次读取的最后一个字节的序号。

使用 Requests 库的 iter_lines() 方法

在 Requests 库中，iter_lines() 方法可用于迭代流式响应中的行。通过持续调用此方法，可以直到流中断，然后使用 try...except 块捕获异常并重新建立连接。

代码示例：

import requests
import time

def stream_logs(log_streaming_endpoint, headers):
    while True:
        try:
            response = requests.get(log_streaming_endpoint, headers=headers, stream=True)
            response_log_streaming_lines = response.iter_lines(chunk_size=1024)
            for line in response_log_streaming_lines:
                log_line = line.decode("utf-8")
                print(log_line)
        except requests.exceptions.RequestException as e:
            print("Error fetching logs:", e)

        # 检查流是否中断
        if not response.ok:
            break
        else:
            # 如果响应成功，清除 'Range' 头
            content_range = response.headers.get('Content-Range')
            if content_range:
                last_byte = int(content_range.split('/')[-1])
                # 设置 'Range' 头以从接收的最后一个字节恢复流
                headers['Range'] = f'bytes={last_byte + 1}-'
            else:
                # 如果 'Content-Range' 头不存在，则从头开始重试
                try:
                    del headers['Range']
                except KeyError:
                    pass  # 'Range' 头不存在，无需删除

            print("Content-Range 头未找到。从头开始重试。")

        time.sleep(1)

# 使用示例
log_streaming_endpoint = "https://example.com/api/v1/logs/pod/my-pod"
headers = {'Authorization': 'Bearer 12345'}

stream_logs(log_streaming_endpoint, headers)