[WIP] mooncake-operator

[liang09255]

Description

Module

• Transfer Engine (mooncake-transfer-engine)
• Mooncake Store (mooncake-store)
• Mooncake EP (mooncake-ep)
• Mooncake PG (mooncake-pg)
• Integration (mooncake-integration)
• P2P Store (mooncake-p2p-store)
• Python Wheel (mooncake-wheel)
• Common (mooncake-common)
• Mooncake RL (mooncake-rl)
• CI/CD
• Docs
• Other

Type of Change

• Bug fix
• New feature
• Refactor
• Breaking change
• Documentation update
• Performance improvement
• Other

How Has This Been Tested?

Test commands:

# Example: bash scripts/run_ci_test.sh

Test results:

• Unit tests pass
• Integration tests pass (if applicable)
• Manual testing done (describe below)

Checklist

• I have performed a self-review of my own code
• I have formatted my code using ./scripts/code_format.sh
• I have run pre-commit run --all-files and all hooks pass
• I have updated the documentation (if applicable)
• I have added tests to prove my changes are effective
• For changes >500 LOC: I have filed an RFC issue

AI Assistance Disclosure

• No AI tools were used
• AI tools were used (specify below)

[gemini-code-assist]

Code Review

This pull request introduces a Next.js Web UI and a Kubernetes Operator (mooncake-operator) to manage Mooncake clusters, alongside vLLM live-migration support and C++ transfer engine enhancements like mock RDMA and automatic RDMA-to-TCP failover. The review feedback is highly actionable and identifies several critical issues: a bug in the operator controller attempting to update a deleted ConfigMap, a body stream reuse error in the benchmark page, missing read timeouts in the proxy server, unhandled non-2xx HTTP status codes in the UI API handler, a failover threshold logic bug in the C++ multi-transport, an accidentally committed backup file, an in-memory session memory leak in the UI server, and missing query parameter stripping in the custom Python HTTP server.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

In reconcileConfigMap, if the created ConfigMap has different data than desired, the code deletes the ConfigMap and then immediately calls r.Update on it. Since the resource has been deleted from the API server, r.Update will fail with a NotFound error. Instead of deleting and updating, you should directly update the existing ConfigMap's data and call r.Update.

		if !configMapDataEqual(created.Data, desired.Data) {
			logger.Info("created ConfigMap has wrong data, updating")
			created.Data = desired.Data
			return r.Update(ctx, &created)
		}

[gemini-code-assist]

In the Fetch API, the response body stream can only be read once. Calling res.text() in the catch block of res.json() will throw a TypeError: body stream already read because the stream has already been consumed by res.json(). To safely log or handle non-JSON error responses, you should read the response as text first, and then parse it as JSON.

      let data
      const text = await res.text()
      try {
        data = JSON.parse(text)
      } catch (parseErr: any) {
        throw new Error(`Invalid JSON response (${res.status}): ${parseErr.message}. Raw body (${text.length} bytes): ${text.substring(0, 200)}`)
      }

[gemini-code-assist]

The _proxy_http_request function defines a timeout parameter but only applies a 5-second timeout to the connection phase. The subsequent read loop has no timeout, which can cause the proxy server to hang indefinitely if the target node is unresponsive. You should wrap the entire request execution (or at least the read loop) with asyncio.wait_for using the provided timeout parameter.

async def _proxy_http_request(
    host: str, port: int, method: str, path: str,
    body: dict | None = None, timeout: float = 10.0,
) -> dict:
    """Make an HTTP request to a migration endpoint."""
    try:
        async def _run():
            reader, writer = await asyncio.open_connection(host, port)
            body_bytes = json.dumps(body).encode() if body else b""
            request_line = f"{method} {path} HTTP/1.1\r\n".encode()
            headers = (
                f"Host: {host}:{port}\r\n"
                f"Content-Length: {len(body_bytes)}\r\n"
                f"Content-Type: application/json\r\n"
                f"Connection: close\r\n\r\n"
            ).encode()
            writer.write(request_line + headers + body_bytes)
            await writer.drain()

            while True:
                line = await reader.readuntil(b"\r\n")
                if line == b"\r\n":
                    break
            resp_body = b""
            while True:
                chunk = await reader.read(4096)
                if not chunk:
                    break
                resp_body += chunk
            writer.close()
            await writer.wait_closed()
            return json.loads(resp_body) if resp_body else {}

        return await asyncio.wait_for(_run(), timeout=timeout)
    except Exception as e:
        logger.error("Proxy HTTP request to %s:%d%s failed: %s",
                     host, port, path, e)
        return {"error": str(e)}

[gemini-code-assist]

If a transport fails and its entry does not exist in transport_health_ yet, it is initialized with consecutive_failures = 1 in the else branch, but the >= failover_threshold_ check is skipped. This means that if failover_threshold_ is configured to 1, the first failure will not trigger failover, requiring a second failure instead. You can simplify and fix this by initializing the entry first or checking the threshold in both cases.

                std::lock_guard<std::mutex> lock(health_mutex_);
                auto &health = transport_health_[proto_str];
                health.consecutive_failures++;
                if (health.consecutive_failures >= failover_threshold_) {
                    health.markUnhealthy(recovery_secs_);
                    LOG(WARNING) << "[MultiTransport] " << proto_str
                                << " marked unhealthy after "
                                << health.consecutive_failures
                                << " consecutive failures. "
                                << "Cooldown for " << recovery_secs_ << "s";
                }

[gemini-code-assist]

A backup file resources.go.bak has been accidentally committed to the repository. Please remove this file to keep the repository clean and maintainable.

[gemini-code-assist]

Sessions are stored in an in-memory Map but are never periodically pruned. Expired sessions are only deleted when a user attempts to use them. If users log in and never return, their sessions will leak memory indefinitely. You should add a periodic setInterval to prune expired sessions.

// --- Session management ---
const SESSION_TTL_MS = 24 * 60 * 60 * 1000 // 24 hours
const sessions = new Map() // token -> { createdAt }

// Periodically prune expired sessions to prevent memory leaks
setInterval(() => {
  const now = Date.now()
  for (const [token, entry] of sessions.entries()) {
    if (now - entry.createdAt > SESSION_TTL_MS) {
      sessions.delete(token)
    }
  }
}, 60 * 60 * 1000)

[gemini-code-assist]

The callMasterAPI helper does not check the HTTP response status code. If the master API returns a non-200 error (e.g., 500 or 404), the function will still resolve successfully with the error body, which can cause unexpected behavior or crashes in the caller. You should reject the promise if the status code is not in the 2xx range.

    const req = http.request(options, (res) => {
      let data = ''
      res.on('data', (chunk) => data += chunk)
      res.on('end', () => {
        if (res.statusCode < 200 || res.statusCode >= 300) {
          reject(new Error(`Master API returned status ${res.statusCode}: ${data}`));
          return;
        }
        try {
          resolve(JSON.parse(data))
        } catch (e) {
          resolve(data)
        }
      })
    })

[gemini-code-assist]

The custom HTTP server matches paths using exact matching or path.startswith, but does not strip query parameters. If a query parameter is appended (e.g., by a monitoring tool or proxy), the route matching will fail. You should strip query parameters from the path before routing.

Suggested change

	method, path, _ = request_line.decode().strip().split(" ", 2)
	request_line = await reader.readuntil(b"\r\n")
	method, path, _ = request_line.decode().strip().split(" ", 2)
	if "?" in path:
	path, _ = path.split("?", 1)