如何模拟返回Socket的Kubernetes API Job监听端点？

这个报错的核心原因是：Kubernetes的Watch接口是长连接流式接口，不是普通的一次性响应接口。你之前的Mock直接返回数据后就关闭了连接，但K8s JS客户端会保持Socket连接等待后续事件推送，连接被主动关闭就会触发socket hang up错误。

实现Mock的核心要求 #

• 必须使用HTTP分块传输编码（Chunked Transfer Encoding），让响应以流式方式发送
• 保持TCP连接不主动关闭，直到客户端主动断开
• 推送的每个事件必须符合Kubernetes WatchEvent的格式

具体实现示例（以Express为例） #

以下是一个可直接运行的Mock端点实现：

const express = require('express');
const app = express();

app.get('/apis/batch/v1/watch/namespaces/:namespace/jobs', (req, res) => {
  // 校验watch参数
  if (req.query.watch !== 'true') {
    return res.status(400).json({ message: 'watch parameter is required' });
  }

  // 设置流式响应必要头
  res.setHeader('Transfer-Encoding', 'chunked');
  res.setHeader('Content-Type', 'application/json');
  res.setHeader('Connection', 'keep-alive');

  // 模拟符合K8s规范的Watch事件
  const mockEvents = [
    {
      type: 'ADDED',
      object: {
        apiVersion: 'batch/v1',
        kind: 'Job',
        metadata: {
          name: 'test-job-1',
          namespace: req.params.namespace,
          uid: 'test-uid-1'
        },
        spec: { template: { spec: { containers: [{ name: 'test', image: 'nginx' }] } } },
        status: { active: 1 }
      }
    },
    {
      type: 'MODIFIED',
      object: {
        apiVersion: 'batch/v1',
        kind: 'Job',
        metadata: {
          name: 'test-job-1',
          namespace: req.params.namespace,
          uid: 'test-uid-1'
        },
        spec: { template: { spec: { containers: [{ name: 'test', image: 'nginx' }] } } },
        status: { active: 0, succeeded: 1 }
      }
    }
  ];

  // 定时推送事件
  let eventIndex = 0;
  const interval = setInterval(() => {
    if (eventIndex >= mockEvents.length) return;
    // 每个事件末尾加换行符（K8s Watch的约定格式）
    const eventStr = JSON.stringify(mockEvents[eventIndex]) + '\n';
    // 发送分块数据并立即刷新
    res.write(eventStr);
    res.flush();
    eventIndex++;
  }, 2000);

  // 监听客户端断开，清理资源
  req.on('close', () => {
    clearInterval(interval);
    res.end();
  });
});

app.listen(3000, () => {
  console.log('Mock K8s API server running on port 3000');
});

关键细节说明 #

• 响应头设置：Transfer-Encoding: chunked是实现流式响应的核心，告诉客户端数据会分块发送
• 事件格式：每个事件必须是标准的K8s WatchEvent结构，包含type（ADDED/MODIFIED/DELETED等）和object（Job资源对象）
• 连接维护：必须监听客户端的close事件，清理定时器并结束响应，避免内存泄漏
• 数据推送：每个事件发送后要调用res.flush()（Express中）确保数据立即发送到客户端，而非缓存

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