Merge pull request #15 from aerospike-community/lyndon/l2-integration-e2e

Phase 2: storage/L2 plugins and live L2 integration test

Author: lyndonbauto

.github/workflows/ci.yml

@@ -51,7 +51,7 @@ jobs:
   integration:
     name: Integration (Aerospike CE)
     runs-on: ubuntu-latest
-    timeout-minutes: 25
+    timeout-minutes: 30
     steps:
       - name: Harden the runner (Audit all outbound calls)
         uses: step-security/harden-runner@9af89fc71515a100421586dfdb3dc9c984fbf411 # v2.19.4
@@ -60,6 +60,13 @@ jobs:
 
       - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
 
+      - name: Checkout LMCache (dev, L2 plugin APIs)
+        uses: actions/checkout@v4
+        with:
+          repository: LMCache/LMCache
+          ref: dev
+          path: LMCache
+
       - name: Set up Python
         uses: actions/setup-python@a309ff8b426b58ec0e2a45f0f869d46889d02405 # v6.2.0
         with:
@@ -69,10 +76,17 @@ jobs:
 
       - name: Install dependencies
         run: |
-          python -m pip install --upgrade pip
+          python -m pip install --upgrade pip wheel
+          # LMCache build-system pins setuptools 77–80 (see LMCache/pyproject.toml).
+          pip install "setuptools>=77.0.3,<81.0.0"
           # CPU-only torch keeps the hosted runner under ~7 GiB RAM (lmcache pulls torch).
           pip install "torch" --index-url https://download.pytorch.org/whl/cpu
-          pip install -e .
+          # torch CPU can downgrade setuptools; restore LMCache's range before editable install.
+          pip install "setuptools>=77.0.3,<81.0.0" --force-reinstall
+          # PyPI 0.4.x lacks L2StoreResult; install dev (pure Python on CI via NO_NATIVE_EXT).
+          NO_NATIVE_EXT=1 pip install -e ./LMCache --no-build-isolation
+          pip install -e . --no-deps
+          pip install "aerospike>=14.0.0,<19.0.0"
           pip install pytest pytest-asyncio
 
       - name: Start Aerospike CE (Docker)

.gitignore

@@ -23,5 +23,7 @@ htmlcov/
 *.swp
 .DS_Store
 .claude/
+# Local LMCache dev clone for L2 integration tests / CI layout (see README).
+LMCache/
 benchmarks/results/
 benchmarks/l2/.env.local

AGENTS.md

@@ -37,7 +37,7 @@ docker/ scripts/
 | ----- | ------- | ----- |
 | Preflight (S0) | `python scripts/preflight.py` | LMCache + Aerospike client symbols |
 | Unit | `pytest tests/unit -q` | No network |
-| Integration | `./scripts/start_aerospike_ce.sh` then `pytest tests/integration -q` | Live CE |
+| Integration | `./scripts/ci_integration_install.sh` then `./scripts/start_aerospike_ce.sh` and `pytest tests/integration -q` | Live CE + LMCache `dev` for L2 E2E (mirrors CI) |
 | Ecosystem bench | `pip install -r benchmarks/requirements.txt` then `python benchmarks/run.py --profile smoke` | Not in CI by default |
 | L2 bench | `./scripts/setup_l2_bench.sh` then `./benchmarks/l2/run.sh` (LMCache `dev` + live CE) | Not in CI by default |
 | Micro bench | `RUN_BENCH=1 pytest benchmarks/micro --benchmark-only` | FakeClient only |

README.md

@@ -113,6 +113,8 @@ pytest tests/integration -v
 
 Large payloads (16–64 MiB): `RUN_LARGE_INTEGRATION=1`.
 
+**L2 plugin** (`test_l2_plugin_*.py`): live tests against Aerospike CE via `PluginL2AdapterConfig`, aligned with LMCache’s meaningful upstream patterns (`test_mock_l2_adapter`, RESP L2 integration, `lmc_external_l2_adapter`) — not LMCache’s full mocked suite. Requires [LMCache](https://github.com/LMCache/LMCache) `dev` in `LMCache/` (or `LMCACHE_SRC`). Install: `./scripts/ci_integration_install.sh`.
+
 ## Benchmarks
 
 Benchmark code lives under **`benchmarks/`** (not included in the PyPI wheel).

scripts/ci_integration_install.sh

@@ -0,0 +1,28 @@
+#!/usr/bin/env bash
+# Mirror the integration job dependency install (for local CI reproduction).
+#
+# Usage (from repo root):
+#   ./scripts/ci_integration_install.sh
+#   ./scripts/start_aerospike_ce.sh && source .aerospike-ci.env && pytest tests/integration -v
+
+set -euo pipefail
+
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+cd "$ROOT"
+
+if [[ ! -d LMCache ]]; then
+  echo "LMCache/ not found; clone dev: git clone --depth 1 -b dev https://github.com/LMCache/LMCache.git LMCache" >&2
+  exit 1
+fi
+
+python -m pip install --upgrade pip wheel
+pip install "setuptools>=77.0.3,<81.0.0"
+pip install "torch" --index-url https://download.pytorch.org/whl/cpu
+pip install "setuptools>=77.0.3,<81.0.0" --force-reinstall
+NO_NATIVE_EXT=1 pip install -e ./LMCache --no-build-isolation
+pip install -e . --no-deps
+# --no-deps skips aerospike; required for start_aerospike_ce.sh host probe and tests.
+pip install "aerospike>=14.0.0,<19.0.0"
+pip install pytest pytest-asyncio
+
+echo "CI integration install complete."

scripts/start_aerospike_ce.sh

@@ -29,9 +29,14 @@ if ! command -v docker >/dev/null 2>&1; then
   exit 1
 fi
 
+PYTHON="${PYTHON:-python}"
+if ! command -v "$PYTHON" >/dev/null 2>&1; then
+  PYTHON=python3
+fi
+
 HOST_PORT="${AEROSPIKE_TEST_PORT:-}"
 if [[ -z "$HOST_PORT" ]]; then
-  HOST_PORT="$(python3 -c 'import socket; s=socket.socket(); s.bind(("", 0)); print(s.getsockname()[1]); s.close()')"
+  HOST_PORT="$("$PYTHON" -c 'import socket; s=socket.socket(); s.bind(("", 0)); print(s.getsockname()[1]); s.close()')"
 fi
 
 CONF_DIR="$(mktemp -d)"
@@ -60,21 +65,28 @@ done
 
 echo "Waiting for host-side client on 127.0.0.1:${HOST_PORT} (up to 90s) ..."
 host_deadline=$((SECONDS + 90))
-until python3 -c "
+_host_probe_err="$(mktemp)"
+until "$PYTHON" -c "
 import aerospike
 c = aerospike.client({'hosts': [('127.0.0.1', ${HOST_PORT})]})
 c.connect()
 info = c.info_random_node('namespace/lmcache')
 c.close()
 assert 'nsup-period=120' in info
-" 2>/dev/null; do
+" 2>"$_host_probe_err"; do
   if (( SECONDS >= host_deadline )); then
     echo "Aerospike CE did not accept host connections in time." >&2
+    if [[ -s "$_host_probe_err" ]]; then
+      echo "Last host probe error:" >&2
+      tail -20 "$_host_probe_err" >&2
+    fi
     docker logs "$CONTAINER_NAME" 2>&1 | tail -80 >&2 || true
+    rm -f "$_host_probe_err"
     exit 1
   fi
   sleep 1
 done
+rm -f "$_host_probe_err"
 
 cat >"$ENV_FILE" <<EOF
 AEROSPIKE_TEST_HOST=127.0.0.1

tests/integration/helpers.py

@@ -3,7 +3,12 @@
 from __future__ import annotations
 
 import asyncio
+import importlib
+import importlib.util
 import os
+import sys
+import types
+from pathlib import Path
 from typing import Any
 
 import torch
@@ -26,6 +31,78 @@ def on_github_actions() -> bool:
     return os.environ.get("GITHUB_ACTIONS", "").lower() == "true"
 
 
+def lmcache_source_root() -> Path | None:
+    """LMCache source tree (CI: ``LMCache/`` in repo; local: sibling ``../LMCache``)."""
+    env = os.environ.get("LMCACHE_SRC", "").strip()
+    if env:
+        root = Path(env)
+        return root if root.is_dir() else None
+    repo = Path(__file__).resolve().parents[2]
+    for candidate in (repo / "LMCache", repo.parent / "LMCache"):
+        if candidate.is_dir():
+            return candidate
+    return None
+
+
+def install_lmcache_test_native_storage_ops_fallback() -> bool:
+    """Use LMCache's test Bitmap stub when ``native_storage_ops`` is not built."""
+    try:
+        mod = importlib.import_module("lmcache.native_storage_ops")
+        if hasattr(mod, "Bitmap") and hasattr(mod, "TTLLock"):
+            return True
+    except Exception:
+        pass
+
+    root = lmcache_source_root()
+    raw_utils = (
+        None
+        if root is None
+        else root / "tests/v1/storage_backend/raw_block_test_utils.py"
+    )
+    if raw_utils is None or not raw_utils.is_file():
+        return False
+
+    spec = importlib.util.spec_from_file_location(
+        "lmcache_raw_block_test_utils", raw_utils
+    )
+    raw_mod = importlib.util.module_from_spec(spec)
+    assert spec.loader is not None
+    spec.loader.exec_module(raw_mod)
+    raw_mod.install_native_storage_ops_fallback()
+    return True
+
+
+def _minimal_native_storage_ops_fallback() -> None:
+    """Tiny Bitmap/TTLLock stub when LMCache source is unavailable."""
+
+    class Bitmap:
+        def __init__(self, size: int, first_n: int = 0) -> None:
+            self._size = int(size)
+            self._bits = {i for i in range(min(int(first_n), self._size))}
+
+        def set(self, index: int) -> None:
+            index = int(index)
+            if index < 0 or index >= self._size:
+                raise IndexError(index)
+            self._bits.add(index)
+
+        def test(self, index: int) -> bool:
+            return int(index) in self._bits
+
+    class TTLLock:
+        pass
+
+    fallback = types.ModuleType("lmcache.native_storage_ops")
+    fallback.Bitmap = Bitmap  # type: ignore[attr-defined]
+    fallback.TTLLock = TTLLock  # type: ignore[attr-defined]
+    sys.modules["lmcache.native_storage_ops"] = fallback
+
+
+def ensure_native_storage_ops_for_l2_tests() -> None:
+    if not install_lmcache_test_native_storage_ops_fallback():
+        _minimal_native_storage_ops_fallback()
+
+
 def aerospike_hosts() -> tuple[tuple[str, int], ...]:
     host = os.environ.get("AEROSPIKE_TEST_HOST", "127.0.0.1")
     port = int(os.environ.get("AEROSPIKE_TEST_PORT", "3000"))

tests/integration/l2_plugin_support.py

@@ -0,0 +1,104 @@
+"""Shared helpers for live AerospikeL2Plugin integration tests (LMCache L2 contract)."""
+
+from __future__ import annotations
+
+import importlib
+import os
+import select
+from collections.abc import Iterator
+
+import pytest
+import torch
+
+from tests.integration.helpers import (
+    aerospike_hosts,
+    ensure_native_storage_ops_for_l2_tests,
+)
+
+ensure_native_storage_ops_for_l2_tests()
+
+l2_mod = importlib.import_module("lmcache_aerospike.l2_plugin")
+if not l2_mod.L2_MP_AVAILABLE:
+    pytest.skip(
+        "LMCache multiprocess L2 APIs (L2StoreResult) not in this lmcache build",
+        allow_module_level=True,
+    )
+
+from lmcache.v1.distributed.api import ObjectKey  # noqa: E402
+from lmcache.v1.distributed.l2_adapters.factory import (  # noqa: E402
+    create_l2_adapter_from_registry,
+)
+from lmcache.v1.distributed.l2_adapters.plugin_l2_adapter import (  # noqa: E402
+    PluginL2AdapterConfig,
+)
+from lmcache.v1.memory_management import (  # noqa: E402
+    MemoryFormat,
+    MemoryObjMetadata,
+    TensorMemoryObj,
+)
+from lmcache.v1.platform import consume_fd  # noqa: E402
+from lmcache.v1.protocol import init_remote_metadata_info  # noqa: E402
+
+init_remote_metadata_info(1)
+
+RUN_INTEGRATION = os.environ.get("RUN_INTEGRATION") == "1"
+L2_IT_SET = os.environ.get("AEROSPIKE_L2_TEST_SET", "kv_chunks_l2_it")
+
+
+def wait_for_event_fd(event_fd: int, timeout: float = 30.0) -> bool:
+    poll = select.poll()
+    poll.register(event_fd, select.POLLIN)
+    if not poll.poll(timeout * 1000):
+        return False
+    try:
+        consume_fd(event_fd)
+    except BlockingIOError:
+        pass
+    return True
+
+
+def object_key(chunk_id: int, model_name: str = "lmcache_aerospike_it") -> ObjectKey:
+    return ObjectKey(
+        chunk_hash=ObjectKey.IntHash2Bytes(chunk_id),
+        model_name=model_name,
+        kv_rank=0,
+    )
+
+
+def memory_obj(size: int = 256, fill_value: float = 1.0) -> TensorMemoryObj:
+    """TensorMemoryObj with shapes/dtypes (LMCache dev metadata for serde)."""
+    raw = torch.empty(size, dtype=torch.float32)
+    raw.fill_(fill_value)
+    meta = MemoryObjMetadata(
+        shape=torch.Size([size]),
+        dtype=torch.float32,
+        address=0,
+        phy_size=size * 4,
+        fmt=MemoryFormat.KV_2LTD,
+        ref_count=1,
+        shapes=[torch.Size([size])],
+        dtypes=[torch.float32],
+    )
+    return TensorMemoryObj(raw, meta, parent_allocator=None)
+
+
+def plugin_config() -> PluginL2AdapterConfig:
+    host, port = aerospike_hosts()[0]
+    return PluginL2AdapterConfig(
+        module_path="lmcache_aerospike.l2_plugin",
+        class_name="AerospikeL2Plugin",
+        adapter_params={
+            "hosts": f"{host}:{port}",
+            "namespace": os.environ.get("AEROSPIKE_TEST_NAMESPACE", "lmcache"),
+            "set_name": L2_IT_SET,
+        },
+    )
+
+
+@pytest.fixture
+def aerospike_l2_adapter() -> Iterator:
+    adapter = create_l2_adapter_from_registry(plugin_config())
+    try:
+        yield adapter
+    finally:
+        adapter.close()