From d772fe6b804cd1415a281dc46a6dd98f7566faee Mon Sep 17 00:00:00 2001 From: robbluo Date: Fri, 4 Sep 2026 11:40:03 +0800 Subject: [PATCH] feat(data-plane): deploy Rust edge and node proxy - deploy the optional Rust data plane while preserving the legacy path by default - add synchronized sandbox throughput tooling and compute-only create capability - preserve configurable monitoring storage and stable sandbox command identity - pin the reviewed openYuanRong and sandboxd component revisions Signed-off-by: robbluo --- .dockerignore | 1 + .gitignore | 1 + AGENTS.md | 39 +- Makefile | 7 + builder/node.Dockerfile | 53 +- builder/scripts/master_entrypoint.sh | 214 ++++++-- builder/scripts/yr_node_bootstrap.sh | 201 +++++-- builder/systemd_services/yuanrong.service | 3 +- deploy/akernel/benchmark/fullscale_create.py | 68 +++ deploy/akernel/benchmark/fullscale_delete.py | 13 + deploy/akernel/benchmark/loadgen.Dockerfile | 16 + deploy/akernel/benchmark/raw_edge_sweep.py | 121 ++++ .../charts/core/scripts/prepare_node.sh | 27 + .../core/templates/edge/edge_service.yaml | 26 + .../templates/frontend/akernel_frontend.yaml | 69 +++ .../core/templates/master/akernel_master.yaml | 22 +- .../charts/core/templates/node/configmap.yaml | 8 + .../charts/core/templates/node/daemonset.yaml | 39 +- deploy/akernel/charts/core/values.yaml | 42 +- .../monitor/templates/prometheus/pvc.yaml | 4 +- .../templates/prometheus/statefulset.yaml | 2 +- deploy/akernel/charts/monitor/values.yaml | 1 + deploy/scripts/build-image.sh | 78 +++ deploy/standalone/README.md | 47 +- deploy/standalone/config/sandboxd_config.toml | 3 + deploy/standalone/start.sh | 142 +++-- deploy/standalone/stop.sh | 5 +- .../_backends/openyuanrong_sandbox.py | 180 ++++++ sdk/python/benchmarks/__init__.py | 1 + sdk/python/benchmarks/cluster_throughput.py | 517 ++++++++++++++++++ sdk/python/benchmarks/edge_raw_throughput.py | 218 ++++++++ .../benchmarks/rrt_direct_throughput.py | 166 ++++++ sdk/python/tests/unit/test_backends.py | 63 +++ .../unit/test_cluster_throughput_benchmark.py | 250 +++++++++ src/sandboxd | 2 +- src/yuanrong | 2 +- 36 files changed, 2468 insertions(+), 183 deletions(-) create mode 100644 deploy/akernel/benchmark/fullscale_create.py create mode 100644 deploy/akernel/benchmark/fullscale_delete.py create mode 100644 deploy/akernel/benchmark/loadgen.Dockerfile create mode 100644 deploy/akernel/benchmark/raw_edge_sweep.py create mode 100644 deploy/akernel/charts/core/templates/edge/edge_service.yaml create mode 100644 sdk/python/benchmarks/__init__.py create mode 100644 sdk/python/benchmarks/cluster_throughput.py create mode 100644 sdk/python/benchmarks/edge_raw_throughput.py create mode 100644 sdk/python/benchmarks/rrt_direct_throughput.py create mode 100644 sdk/python/tests/unit/test_cluster_throughput_benchmark.py diff --git a/.dockerignore b/.dockerignore index 9550d54..d6c3a5d 100644 --- a/.dockerignore +++ b/.dockerignore @@ -33,6 +33,7 @@ src/yuanrong/ **/*.egg-info/ # Local image and archive outputs. +.artifacts/ builder/*.img builder/*.tar builder/*.tar.gz diff --git a/.gitignore b/.gitignore index 4f99cb4..fd15213 100644 --- a/.gitignore +++ b/.gitignore @@ -16,6 +16,7 @@ __pycache__/ /sdk/python/dist/ # Local build artifacts. +.artifacts/ builder/*.img builder/*.tar builder/*.tar.gz diff --git a/AGENTS.md b/AGENTS.md index f656f42..dfd3fca 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -117,11 +117,12 @@ The build creates only the selected image reference; it does not add a second The build helper performs two Docker builds. `builder/runtime.Dockerfile` creates `yr-runtime-rootfs.img`; the default `rrt` profile contains the -pinned openYuanRong RRT binary without Python. Set +pinned openYuanRong RRT binary without Python and verifies that it is byte-for- +byte identical to the binary in the matching `openyuanrong_rrt` wheel. Set `RUNTIME_PROFILE=python` to include the optional Python 3.10 through 3.14 -runtimes and `openyuanrong_sdk`. `builder/node.Dockerfile` then compiles the -node components and produces the AKernel all-in-one image using the selected -runtime image and its matching service configuration. +runtimes, `openyuanrong_sdk`, and the RRT wheel. `builder/node.Dockerfile` then +compiles the node components and produces the AKernel all-in-one image using +the selected runtime image and its matching service configuration. The control-plane and RRT release version is independent of the optional actor-based `openyuanrong_sdk` installed in the Python runtime profile. This @@ -161,6 +162,29 @@ To test an unreleased RRT binary, provide both `RRT_RUNTIME_URL` and `RRT_RUNTIME_SHA256` to `make build`. The runtime build verifies the binary before packaging it into the selected runtime root filesystem. +The Rust data plane is packaged separately from the core wheel. Every image +build must provide its SHA-256 plus exactly one source: a local authenticated +Buildkite download or a public URL. A local wheel is exposed to Docker through +a dedicated read-only BuildKit context and is not copied into the primary build +context or a persistent image layer: + +```bash +make build \ + OPEN_YR_DATA_PLANE_WHEEL_PATH=/absolute/path/openyuanrong_data_plane.whl \ + OPEN_YR_DATA_PLANE_WHEEL_SHA256= +``` + +Use `OPEN_YR_DATA_PLANE_WHEEL_URL` instead of the path after the artifact has a +public URL. The all-in-one image installs both wheels through Python packaging, +keeps `/usr/bin/yr` as the legacy native CLI for Kubernetes entrypoints, and +uses `/usr/local/bin/yr` to start the Rust Node Proxy and Edge Frontend in +standalone mode. + +On a development host whose proxy only listens on loopback, set +`AKERNEL_BUILD_NETWORK=host` while building. The helper then gives BuildKit host +network access and forwards the standard proxy environment variables as Docker +proxy build arguments. The default remains Docker's isolated build network. + Inspect the selected local versions without building an image: ```bash @@ -392,6 +416,13 @@ sandbox port URLs and reverse tunnels; exec and file transfer continue to use `AKERNEL_SERVER_ADDRESS`. Standalone uses `akerneldev/all-in-one:latest` by default; pass `IMAGE` to test a locally built or differently tagged image. +In standalone mode Traefik terminates neither YuanRong protocol nor TLS. Its +HTTPS TCP entrypoint passes TLS through to the Rust Edge Frontend on port 8443, +and its HTTP entrypoint forwards to the Edge plain listener on port 8080. Edge +routes control traffic to the core frontend; sandbox traffic follows the Rust +Edge Frontend to Node Proxy path. Their local readiness endpoints are 18080 +and 18443 respectively. + Standalone GPU testing additionally requires NVIDIA Container Toolkit on the host and `AKERNEL_ENABLE_GPU=true`. sandboxd uses the read-only cgroup node-resource provider in standalone mode; Kubernetes deployments retain the diff --git a/Makefile b/Makefile index 100218d..823e590 100644 --- a/Makefile +++ b/Makefile @@ -12,6 +12,9 @@ OPEN_YR_CORE_WHEEL_URL ?= OPEN_YR_CORE_WHEEL_SHA256 ?= RRT_RUNTIME_URL ?= RRT_RUNTIME_SHA256 ?= +OPEN_YR_DATA_PLANE_WHEEL_PATH ?= +OPEN_YR_DATA_PLANE_WHEEL_URL ?= +OPEN_YR_DATA_PLANE_WHEEL_SHA256 ?= TOKEN_TTL ?= $(if $(TTL),$(TTL),24h) TENANT ?= default ROLE ?= developer @@ -55,6 +58,7 @@ help: @echo " make build AKERNEL_ENABLE_FIRECRACKER=false Exclude Firecracker" @echo " make build AKERNEL_ENABLE_RUNC=true Include the optional runc payload" @echo " make build RRT_RUNTIME_URL=... RRT_RUNTIME_SHA256=... Override RRT artifact" + @echo " make build OPEN_YR_DATA_PLANE_WHEEL_PATH= Use a local Rust data-plane wheel" @echo " make versions Show locally selected component versions" @echo " make push Push the configured all-in-one image" @echo " make plan Terraform plan" @@ -107,6 +111,9 @@ build: if [[ -n "$(OPEN_YR_CORE_WHEEL_SHA256)" ]]; then args+=(--open-yr-core-wheel-sha256 "$(OPEN_YR_CORE_WHEEL_SHA256)"); fi; \ if [[ -n "$(RRT_RUNTIME_URL)" ]]; then args+=(--rrt-runtime-url "$(RRT_RUNTIME_URL)"); fi; \ if [[ -n "$(RRT_RUNTIME_SHA256)" ]]; then args+=(--rrt-runtime-sha256 "$(RRT_RUNTIME_SHA256)"); fi; \ + if [[ -n "$(OPEN_YR_DATA_PLANE_WHEEL_PATH)" ]]; then args+=(--open-yr-data-plane-wheel-path "$(OPEN_YR_DATA_PLANE_WHEEL_PATH)"); fi; \ + if [[ -n "$(OPEN_YR_DATA_PLANE_WHEEL_URL)" ]]; then args+=(--open-yr-data-plane-wheel-url "$(OPEN_YR_DATA_PLANE_WHEEL_URL)"); fi; \ + if [[ -n "$(OPEN_YR_DATA_PLANE_WHEEL_SHA256)" ]]; then args+=(--open-yr-data-plane-wheel-sha256 "$(OPEN_YR_DATA_PLANE_WHEEL_SHA256)"); fi; \ ./deploy/scripts/build-image.sh "$${args[@]}" .PHONY: versions diff --git a/builder/node.Dockerfile b/builder/node.Dockerfile index 82171c0..9667ac9 100644 --- a/builder/node.Dockerfile +++ b/builder/node.Dockerfile @@ -1,3 +1,5 @@ +# syntax=docker/dockerfile:1.7 + # Copyright (c) 2026 Ant Group Corporation. # # SPDX-License-Identifier: Apache-2.0 @@ -16,6 +18,9 @@ ARG OPEN_YR_CORE_WHEEL_SHA256= ARG OPEN_YR_RELEASE_BASE_URL=https://openyuanrong.obs.cn-southwest-2.myhuaweicloud.com/release ARG OPEN_YR_CORE_AMD64_SHA256=65c1f27e7e700a253a2e907dea0273e85f1c76610e48c93544caa6bcc07ac3af ARG OPEN_YR_CORE_ARM64_SHA256=29d25c3388c2913346035ee8df9b8159e702de7a8e6f783e3e218ab773896333 +ARG OPEN_YR_DATA_PLANE_WHEEL_URL= +ARG OPEN_YR_DATA_PLANE_WHEEL_SHA256= +ARG OPEN_YR_DATA_PLANE_WHEEL_NAME= ARG GVISOR_DOWNLOAD_IMAGE=ubuntu:24.04 ARG GVISOR_RELEASE ARG GVISOR_AMD64_URL @@ -223,6 +228,9 @@ ARG AKERNEL_REVISION ARG OPEN_YR_VERSION ARG OPEN_YR_CORE_WHEEL_URL ARG OPEN_YR_CORE_WHEEL_SHA256 +ARG OPEN_YR_DATA_PLANE_WHEEL_URL +ARG OPEN_YR_DATA_PLANE_WHEEL_SHA256 +ARG OPEN_YR_DATA_PLANE_WHEEL_NAME ARG OPEN_YR_RELEASE_BASE_URL ARG OPEN_YR_CORE_AMD64_SHA256 ARG OPEN_YR_CORE_ARM64_SHA256 @@ -297,10 +305,11 @@ RUN ln -snf /usr/share/zoneinfo/$TZ /etc/localtime && \ ENV YR_INSTALLATION_DIR=/home/yuanrong -# Install the complete, language-runtime-free openYuanRong control plane from -# its checksum-pinned core wheel. A URL and checksum pair may override the -# release asset when validating an unreleased daily build. -RUN set -eux; \ +# Install the complete, language-runtime-free openYuanRong control plane and +# the standalone Rust data-plane executables. The data-plane wheel can arrive +# either from a public URL or from a BuildKit file mount supplied by the build +# helper for authenticated Buildkite artifacts. +RUN --mount=type=bind,from=open_yr_data_plane_wheel,source=.,target=/run/open_yr_data_plane_wheel,ro set -eux; \ case "${TARGETARCH:-}" in \ amd64) wheel_arch=x86_64; wheel_platform=amd64; release_sha="${OPEN_YR_CORE_AMD64_SHA256}" ;; \ arm64) wheel_arch=aarch64; wheel_platform=arm64; release_sha="${OPEN_YR_CORE_ARM64_SHA256}" ;; \ @@ -325,21 +334,41 @@ RUN set -eux; \ test -z "${OPEN_YR_CORE_WHEEL_SHA256}"; \ fi; \ wheel="/tmp/${wheel_name}"; \ - target=/tmp/openyuanrong-core; \ curl -fSL --retry 10 --retry-delay 2 --retry-all-errors \ "${wheel_url}" -o "${wheel}"; \ echo "${wheel_sha} ${wheel}" | sha256sum -c -; \ + data_plane_wheel_name="${OPEN_YR_DATA_PLANE_WHEEL_NAME}"; \ + if [ -n "${OPEN_YR_DATA_PLANE_WHEEL_URL}" ]; then \ + data_plane_wheel_name="$(python3 -c 'import os, sys, urllib.parse; print(os.path.basename(urllib.parse.unquote(urllib.parse.urlparse(sys.argv[1]).path)))' "${OPEN_YR_DATA_PLANE_WHEEL_URL}")"; \ + fi; \ + case "${data_plane_wheel_name}" in openyuanrong_data_plane-*.whl) ;; *) echo "the data-plane source must reference a valid openyuanrong_data_plane wheel filename" >&2; exit 1 ;; esac; \ + data_plane_wheel="/tmp/${data_plane_wheel_name}"; \ + if [ -n "${OPEN_YR_DATA_PLANE_WHEEL_URL}" ]; then \ + curl -fSL --retry 10 --retry-delay 2 --retry-all-errors \ + "${OPEN_YR_DATA_PLANE_WHEEL_URL}" -o "${data_plane_wheel}"; \ + else \ + test -f "/run/open_yr_data_plane_wheel/${data_plane_wheel_name}"; \ + cp "/run/open_yr_data_plane_wheel/${data_plane_wheel_name}" "${data_plane_wheel}"; \ + fi; \ + echo "${OPEN_YR_DATA_PLANE_WHEEL_SHA256} ${data_plane_wheel}" | sha256sum -c -; \ python3 -m pip install \ --break-system-packages \ --no-cache-dir \ - --no-deps \ - --target "${target}" \ - "${wheel}"; \ - test -x "${target}/yr/functionsystem/bin/yr"; \ + --index-url "${PIP_INDEX_URL}" \ + --timeout 120 \ + --retries 10 \ + "${wheel}" \ + "${data_plane_wheel}"; \ + yr_package_dir="$(python3 -c 'from pathlib import Path; import yr; print(Path(yr.__file__).resolve().parent)')"; \ + python_bin_dir="$(python3 -c 'import sysconfig; print(sysconfig.get_path("scripts"))')"; \ + test -x "${yr_package_dir}/functionsystem/bin/yr"; \ + test -x "${yr_package_dir}/data_plane/bin/yr-node-proxy"; \ + test -x "${yr_package_dir}/data_plane/bin/yr-edge-frontend"; \ + test -x "${yr_package_dir}/data_plane/bin/yr-data-plane-forward"; \ + test -x "${python_bin_dir}/yr"; \ mkdir -p "${YR_INSTALLATION_DIR}"; \ - cp -a "${target}/yr/." "${YR_INSTALLATION_DIR}/"; \ - rm -rf "${target}" "${wheel}"; \ - ln -sfn "${YR_INSTALLATION_DIR}/functionsystem/bin/yr" /usr/bin/yr + rm -f "${wheel}" "${data_plane_wheel}"; \ + ln -sfn "${yr_package_dir}/functionsystem/bin/yr" /usr/bin/yr COPY --from=runtime-image /yr-runtime-rootfs.img ${YR_INSTALLATION_DIR}/yr-runtime-rootfs.img diff --git a/builder/scripts/master_entrypoint.sh b/builder/scripts/master_entrypoint.sh index 2a795ab..1dde7d0 100644 --- a/builder/scripts/master_entrypoint.sh +++ b/builder/scripts/master_entrypoint.sh @@ -49,64 +49,172 @@ else echo "otelcol watchdog skipped" fi -# Set enable_traefik_provider based on TRAEFIK_MODE -if [ "${TRAEFIK_MODE:-etcd}" = "http" ]; then - ENABLE_TRAEFIK_PROVIDER=true -else - ENABLE_TRAEFIK_PROVIDER=false -fi - if [ -z "${LITEBUS_DATA_KEY:-}" ]; then echo "LITEBUS_DATA_KEY is required for akernel master/frontend" >&2 exit 1 fi -YR_BIN="${YR_BIN:-/usr/bin/yr}" -if [ ! -x "${YR_BIN}" ]; then - echo "yr binary not found or not executable: ${YR_BIN}" >&2 +YR_PYTHON_CLI="${YR_PYTHON_CLI:-/usr/local/bin/yr}" +if [ ! -x "${YR_PYTHON_CLI}" ]; then + echo "openYuanRong Python CLI not found: ${YR_PYTHON_CLI}" >&2 exit 1 fi -exec "${YR_BIN}" start --master --block true \ - -e -c 0 -m 8000 -s 4096 -n $HOSTNAME \ - -d $DEPLOY_PATH \ - --fs_health_check_retry_interval 1 \ - --schedule_relaxed 20 \ - --enable_faas_frontend ${ENABLE_FAAS_FRONTEND:-true} \ - --enable_function_scheduler ${ENABLE_FUNCTION_SCHEDULER:-false} \ - --enable_meta_service ${ENABLE_META_SERVICE:-true} \ - --enable_iam_server ${ENABLE_IAM_SERVER:-true} \ - --iam_token_expired_time_span 604800 \ - --ssl_base_path=/home/yuanrong/.cert/ \ - --frontend_ssl_enable=true \ - --frontend_client_auth_type NoClientCert \ - --enable_function_token_auth ${ENABLE_FUNCTION_TOKEN_AUTH:-true} \ - --enable_inherit_env false \ - --npu_collection_mode off \ - --port_policy FIX \ - --system_timeout 300000 \ - --enable_distributed_master false \ - --etcd_mode outter \ - --etcd_addr_list $ETCD_ADDRESS \ - --etcd_port ${ETCD_PORT} \ - --etcd_peer_port 2378 \ - --enable_metrics ${ENABLE_METRICS} \ - --metrics_config_file "/home/yuanrong/metrics/metrics_config.json" \ - --enable_trace ${ENABLE_TRACE} \ - --trace_config "$(cat /home/yuanrong/trace/trace_config.json)" \ - --ds_rpc_thread_num 128 \ - --function_proxy_merge_process_enable true \ - --enable_traefik_provider=${ENABLE_TRAEFIK_PROVIDER} \ - --traefik_http_entry_point=${TRAEFIK_HTTP_ENTRYPOINT:-websecure} \ - --traefik_enable_tls=${TRAEFIK_ENABLE_TLS:-false} \ - --traefik_forward_timeout_ms=3000 \ - --frontend_lease_bypass true \ - --iam_ssl_enable true \ - --ssl_root_file ca.crt \ - --ssl_cert_file module.crt \ - --ssl_key_file module.key \ - --iam_local_listen_port 31113 \ - --iam_local_ip 127.0.0.1 \ - --enable_direct_routing false \ - --enable_sandbox_router true \ - ${META_SERVICE_ADDRESS:+--meta_service_address $META_SERVICE_ADDRESS} +resolve_host() { + local host="$1" + local attempts="${YR_DNS_RESOLVE_ATTEMPTS:-90}" + local delay="${YR_DNS_RESOLVE_INTERVAL_SECONDS:-2}" + local attempt + local resolved + + for ((attempt = 1; attempt <= attempts; attempt++)); do + if resolved="$(python3 - "${host}" 2>/dev/null <<'PY' +import socket +import sys + +print(socket.gethostbyname(sys.argv[1])) +PY + )" && [ -n "${resolved}" ]; then + printf '%s\n' "${resolved}" + return 0 + fi + sleep "${delay}" + done + echo "failed to resolve ${host} after ${attempts} attempts" >&2 + return 1 +} + +toml_etcd_addresses() { + local addr_list="$1" + local result="" sep="" entry host port resolved_host + + IFS=',' read -ra entries <<< "${addr_list}" + for entry in "${entries[@]}"; do + host="${entry%:*}" + port="${entry##*:}" + resolved_host="$(resolve_host "${host}")" + result="${result}${sep}{ip=\"${resolved_host}\",peer_port=${port},port=${port}}" + sep="," + done + printf '[%s]' "${result}" +} + +toml_string_list() { + python3 - "$1" <<'PY' +import json +import sys + +print(json.dumps([item.strip() for item in sys.argv[1].split(",") if item.strip()])) +PY +} + +is_true() { + case "${1:-}" in + 1|true|TRUE|yes|YES|on|ON) return 0 ;; + *) return 1 ;; + esac +} + +host_ip="${INSTANCE_IP:-$(hostname -i | awk '{print $1}')}" +etcd_addr_list="${YR_ETCD_ADDR_LIST:-${ETCD_ADDRESS:-127.0.0.1}:${ETCD_PORT:-2379}}" +etcd_addresses="$(toml_etcd_addresses "${etcd_addr_list}")" +services_path="${YR_SERVICES_PATH:-/home/yuanrong/deploy/process/services.yaml}" +common_args=( + --function-proxy-merge-process-enable + --port-policy FIX + --block true + -s "values.host_ip=\"${host_ip}\"" + -s "values.local_ip=\"${host_ip}\"" + -s "values.cpu_num=${YR_CPU_NUM_MILLICORES:-1000}" + -s "values.memory_num=${YR_MEMORY_NUM_MB:-1536}" + -s "values.shared_memory_num=${YR_SHARED_MEMORY_NUM_MB:-512}" + -s 'ds_worker.health_check.timeout=300' + -s 'ds_worker.args.heartbeat_interval_ms=1000' + -s 'ds_worker.args.node_timeout_s=20' + -s 'ds_worker.args.node_dead_timeout_s=60' + -s 'ds_worker.args.client_dead_timeout_s=60' + -s "ds_worker.args.cluster_name=\"${YR_DATA_SYSTEM_CLUSTER_NAME:-}\"" + -s 'values.etcd.enable_multi_master=true' + -s "values.etcd.address=${etcd_addresses}" + -s 'values.fs.tls.enable=false' + -s "function_proxy.args.services_path=\"${services_path}\"" + -s 'function_proxy.args.advertise_frontend_proxy_create=false' + -s 'function_proxy.args.enable_inherit_env=true' + -s 'function_proxy.args.force_low_reliability_instance=true' +) + +if [ "${AKERNEL_ROLE:-master}" = "frontend" ]; then + master_host="${YR_MASTER_ADDRESS:-${META_SERVICE_ADDRESS%:*}}" + master_ip="$(resolve_host "${master_host}")" + meta_service_address="${META_SERVICE_ADDRESS:-${master_ip}:31111}" + + edge_args=() + if is_true "${ENABLE_EDGE_FRONTEND:-false}"; then + tls_dir="${AKERNEL_COMPONENT_CERT_DIR:-/home/yuanrong/.cert}" + tls_cert="${YR_DATA_PLANE_EDGE_FRONTEND_TLS_CERT:-${tls_dir}/module.crt}" + tls_key="${YR_DATA_PLANE_EDGE_FRONTEND_TLS_KEY:-${tls_dir}/module.key}" + for required in "${tls_cert}" "${tls_key}"; do + if [ ! -s "${required}" ]; then + echo "required Rust Edge certificate is missing: ${required}" >&2 + exit 1 + fi + done + + allowed_client_cidrs="$(toml_string_list "${YR_DATA_PLANE_EDGE_FRONTEND_ALLOWED_CLIENT_CIDRS:-}")" + default_control_plane_routes="exact:/,exact:/healthz,prefix:/terminal,prefix:/api/instances,prefix:/api/jobs,prefix:/api/sandbox,prefix:/functions,prefix:/api-docs,prefix:/admin/v1/functions,prefix:/serverless/v1/functions,prefix:/serverless/v1/stream,prefix:/serverless/v1/componentshealth,prefix:/serverless/v1/posix,prefix:/frontend/v1/instance,prefix:/datasystem/v1,prefix:/serverless/v2,prefix:/app/v1,prefix:/client/v1/lease,prefix:/invocations,prefix:/global-scheduler" + control_plane_routes="$(toml_string_list "${YR_DATA_PLANE_EDGE_FRONTEND_CONTROL_PLANE_ROUTES:-${default_control_plane_routes}}")" + edge_args=( + -s 'mode.agent.edge_frontend=true' + -s "values.edge_frontend.tls_bind=\"${YR_DATA_PLANE_EDGE_FRONTEND_TLS_BIND:-0.0.0.0:8443}\"" + -s "values.edge_frontend.plain_bind=\"${YR_DATA_PLANE_EDGE_FRONTEND_PLAIN_BIND:-0.0.0.0:8080}\"" + -s "values.edge_frontend.health_bind=\"${YR_DATA_PLANE_EDGE_FRONTEND_HEALTH_BIND:-0.0.0.0:18080}\"" + -s "values.edge_frontend.frontend_address=\"${YR_DATA_PLANE_EDGE_FRONTEND_CONTROL_PLANE_ADDRESS:-${host_ip}:8888}\"" + -s "values.edge_frontend.control_plane_routes=${control_plane_routes}" + -s "values.edge_frontend.tls_cert=\"${tls_cert}\"" + -s "values.edge_frontend.tls_key=\"${tls_key}\"" + -s "values.edge_frontend.validate_iam=${YR_DATA_PLANE_EDGE_FRONTEND_VALIDATE_IAM:-true}" + -s "values.edge_frontend.iam_address=\"${YR_DATA_PLANE_EDGE_FRONTEND_IAM_ADDRESS:-${host_ip}:31112}\"" + -s "values.edge_frontend.node_security_mode=\"${YR_DATA_PLANE_EDGE_FRONTEND_NODE_SECURITY_MODE:-network}\"" + -s "values.edge_frontend.allowed_client_cidrs=${allowed_client_cidrs}" + -s "values.edge_frontend.allow_any_client=${YR_DATA_PLANE_EDGE_FRONTEND_ALLOW_ANY_CLIENT:-false}" + -s "values.edge_frontend.log_level=\"${YR_DATA_PLANE_EDGE_FRONTEND_LOG_LEVEL:-info}\"" + ) + fi + + exec "${YR_PYTHON_CLI}" start \ + --log-dir-prefix /home/yuanrong/sessions/frontend \ + "${common_args[@]}" \ + -s 'mode.agent.frontend=true' \ + -s 'mode.agent.iam_server=true' \ + -s "values.function_master.ip=\"${master_ip}\"" \ + -s "values.meta_service.ip=\"${master_ip}\"" \ + -s 'values.meta_service.port=31111' \ + -s 'values.frontend.port=8888' \ + -s 'values.frontend.ssl_enable=false' \ + -s 'values.frontend.client_auth_type="NoClientCert"' \ + -s "values.frontend.enable_function_token_auth=${ENABLE_FUNCTION_TOKEN_AUTH:-true}" \ + -s "values.frontend.enable_func_token_auth=${ENABLE_FUNCTION_TOKEN_AUTH:-true}" \ + -s 'values.frontend.frontend_lease_bypass=true' \ + -s "values.frontend.sandbox_router_enable=${ENABLE_SANDBOX_ROUTER:-false}" \ + -s "values.frontend.meta_service_address=\"${meta_service_address}\"" \ + -s "values.frontend.iam_server_address=\"${host_ip}:31112\"" \ + -s 'values.iam_server.port=31112' \ + "${edge_args[@]}" +fi + +exec "${YR_PYTHON_CLI}" start \ + --master \ + --log-dir-prefix /home/yuanrong/sessions/master \ + "${common_args[@]}" \ + -s 'mode.master.etcd=false' \ + -s 'mode.master.ds_master=false' \ + -s "mode.master.frontend=${ENABLE_FAAS_FRONTEND:-false}" \ + -s "mode.master.function_scheduler=${ENABLE_FUNCTION_SCHEDULER:-false}" \ + -s "mode.master.meta_service=${ENABLE_META_SERVICE:-true}" \ + -s "mode.master.iam_server=${ENABLE_IAM_SERVER:-false}" \ + -s 'values.function_master.global_scheduler_port=22770' \ + -s 'values.meta_service.port=31111' \ + -s 'values.iam_server.port=31112' \ + -s 'function_master.args.enable_traefik_provider=false' \ + -s 'function_master.args.system_timeout=300000' \ + -s "function_master.args.services_path=\"${services_path}\"" diff --git a/builder/scripts/yr_node_bootstrap.sh b/builder/scripts/yr_node_bootstrap.sh index 7c68326..fc9361d 100755 --- a/builder/scripts/yr_node_bootstrap.sh +++ b/builder/scripts/yr_node_bootstrap.sh @@ -3,6 +3,8 @@ # Copyright (c) 2026 Ant Group Corporation. # # SPDX-License-Identifier: Apache-2.0 +set -eo pipefail + ulimit -n 32768 export YR_RUNTIME_BACKEND=sandboxd @@ -44,6 +46,92 @@ echo "Using ${YR_NODE_IP} as the YuanRong node address" CHECKPOINT_DIR="/home/akernel/checkpoints" mkdir -p "${CHECKPOINT_DIR}" +resolve_host() { + local host="$1" + local attempts="${YR_DNS_RESOLVE_ATTEMPTS:-90}" + local delay="${YR_DNS_RESOLVE_INTERVAL_SECONDS:-2}" + local attempt + local resolved + + for ((attempt = 1; attempt <= attempts; attempt++)); do + if resolved="$(python3 - "${host}" 2>/dev/null <<'PY' +import socket +import sys + +print(socket.gethostbyname(sys.argv[1])) +PY + )" && [ -n "${resolved}" ]; then + printf '%s\n' "${resolved}" + return 0 + fi + sleep "${delay}" + done + echo "failed to resolve ${host} after ${attempts} attempts" >&2 + return 1 +} + +toml_etcd_addresses() { + local addr_list="$1" + local result="" + local sep="" + local entry host port resolved_host + + IFS=',' read -ra entries <<< "${addr_list}" + for entry in "${entries[@]}"; do + host="${entry%:*}" + port="${entry##*:}" + resolved_host="$(resolve_host "${host}")" + result="${result}${sep}{ip=\"${resolved_host}\",peer_port=${port},port=${port}}" + sep="," + done + printf '[%s]' "${result}" +} + +if [[ "${YR_DATA_PLANE_NODE_PROXY_ENABLED:-false}" =~ ^(1|true|TRUE|yes|YES|on|ON)$ ]]; then + YR_PYTHON_CLI="${YR_PYTHON_CLI:-/usr/local/bin/yr}" + if [ ! -x "${YR_PYTHON_CLI}" ]; then + echo "openYuanRong Python CLI not found: ${YR_PYTHON_CLI}" >&2 + exit 1 + fi + ETCD_ADDR_LIST="${YR_ETCD_ADDR_LIST:-${ETCD_ADDRESS}:${ETCD_PORT}}" + ETCD_ADDRESSES="$(toml_etcd_addresses "${ETCD_ADDR_LIST}")" + MASTER_HOST="${YR_MASTER_ADDRESS:-akernel-master}" + MASTER_IP="$(resolve_host "${MASTER_HOST}")" + + exec "${YR_PYTHON_CLI}" start \ + --function-proxy-merge-process-enable \ + --data-system-enable true \ + --port-policy FIX \ + --block true \ + --log-dir-prefix /home/yuanrong/sessions/node \ + -s "values.host_ip=\"${YR_NODE_IP}\"" \ + -s "values.local_ip=\"${YR_NODE_IP}\"" \ + -s "values.shared_memory_num=${YR_SHARED_MEMORY_NUM_MB:-512}" \ + -s 'ds_worker.health_check.timeout=300' \ + -s 'ds_worker.args.heartbeat_interval_ms=1000' \ + -s 'ds_worker.args.node_timeout_s=20' \ + -s 'ds_worker.args.node_dead_timeout_s=60' \ + -s 'ds_worker.args.client_dead_timeout_s=60' \ + -s "ds_worker.args.cluster_name=\"${YR_DATA_SYSTEM_CLUSTER_NAME:-}\"" \ + -s "values.function_master.ip=\"${MASTER_IP}\"" \ + -s 'values.etcd.enable_multi_master=true' \ + -s "values.etcd.address=${ETCD_ADDRESSES}" \ + -s 'values.fs.tls.enable=false' \ + -s 'values.node_proxy.enabled=true' \ + -s "values.node_proxy.bind=\"${YR_DATA_PLANE_NODE_PROXY_BIND:-0.0.0.0:9443}\"" \ + -s "values.node_proxy.advertise_address=\"${YR_NODE_PROXY_ADDRESS:-${YR_NODE_IP}:9443}\"" \ + -s "values.node_proxy.health_bind=\"${YR_DATA_PLANE_NODE_PROXY_HEALTH_BIND:-127.0.0.1:18443}\"" \ + -s 'values.node_proxy.allowed_target_cidrs=["172.17.0.0/16"]' \ + -s "values.node_proxy.allow_any_edge=${YR_DATA_PLANE_NODE_PROXY_ALLOW_ANY_EDGE:-true}" \ + -s "values.node_proxy.edge_security_mode=\"${YR_DATA_PLANE_EDGE_FRONTEND_NODE_SECURITY_MODE:-network}\"" \ + -s 'values.node_proxy.activity_interval_sec=2' \ + -s 'function_proxy.args.services_path="/home/yuanrong/deploy/process/services.yaml"' \ + -s 'function_proxy.args.metrics_collector_type="external"' \ + -s 'function_proxy.args.enable_traefik_registry=false' \ + -s 'function_proxy.args.enable_inherit_env=true' \ + -s 'function_proxy.args.force_low_reliability_instance=true' +fi + # Select the legacy etcd registry or the FunctionMaster HTTP provider. if [ "${TRAEFIK_MODE:-etcd}" = "etcd" ]; then ENABLE_TRAEFIK_REGISTRY=${ENABLE_TRAEFIK_REGISTRY:-true} @@ -53,7 +141,7 @@ else ENABLE_TRAEFIK_PROVIDER=true fi -if [ "x${AKS_LOCAL_MODE}" == "xtrue" ]; then +if [ "${AKS_LOCAL_MODE:-}" = "true" ]; then if [ -z "${LITEBUS_DATA_KEY:-}" ] && [ -r /home/akernel/iam-seed ]; then LITEBUS_DATA_KEY="$(tr -d '[:space:]' < /home/akernel/iam-seed)" export LITEBUS_DATA_KEY @@ -62,57 +150,68 @@ if [ "x${AKS_LOCAL_MODE}" == "xtrue" ]; then echo "LITEBUS_DATA_KEY is required in standalone mode" >&2 exit 1 fi - /usr/bin/yr start --master \ - --ip_address "${YR_NODE_IP}" \ - --port_policy FIX \ - --enable_function_scheduler=false \ - --enable_faas_frontend=true \ - --enable_meta_service=true \ - --enable_iam_server=true \ - --iam_token_expired_time_span 604800 \ - --ssl_base_path=/home/yuanrong/.cert/ \ - --frontend_ssl_enable=true \ - --frontend_client_auth_type NoClientCert \ - --enable_function_token_auth true \ - --ds_node_timeout_s 30 \ - --ds_client_dead_timeout_s 60 \ - --ds_heartbeat_interval_ms 1000 \ - --ds_node_dead_timeout_s 120 \ - --system_timeout 60000 \ + YR_PYTHON_CLI="${YR_PYTHON_CLI:-/usr/local/bin/yr}" + if [ ! -x "${YR_PYTHON_CLI}" ]; then + echo "openYuanRong Python CLI not found: ${YR_PYTHON_CLI}" >&2 + exit 1 + fi + + exec "${YR_PYTHON_CLI}" start \ + --master \ + --function-proxy-merge-process-enable \ + --port-policy FIX \ --block true \ - --etcd_port ${ETCD_PORT:-2379} \ - --etcd_peer_port ${ETCD_PEER_PORT:-2378} \ - --enable_inherit_env false \ - --npu_collection_mode off \ - --enable_distributed_master false \ - --metrics_collector_type external \ - --enable_traefik_registry=${ENABLE_TRAEFIK_REGISTRY} \ - --enable_traefik_provider=${ENABLE_TRAEFIK_PROVIDER} \ - --traefik_enable_tls=${TRAEFIK_ENABLE_TLS:-false} \ - --traefik_etcd_prefix=traefik \ - --traefik_lease_ttl=300000 \ - --traefik_http_entrypoint=${TRAEFIK_HTTP_ENTRYPOINT:-websecure} \ - --traefik_http_entry_point=${TRAEFIK_HTTP_ENTRYPOINT:-websecure} \ - --enable_metrics ${ENABLE_METRICS} \ - --metrics_config_file "/home/yuanrong/metrics/metrics_config.json" \ - --enable_trace ${ENABLE_TRACE} \ - --trace_config "$(cat /home/yuanrong/trace/trace_config.json)" \ - --log_root "${YR_LOG_PATH}" \ - --function_proxy_merge_process_enable true \ - --fc_agent_mgr_retry_times 30 \ - --fc_agent_mgr_retry_cycle 60000 \ - --iam_ssl_enable true \ - --ssl_root_file ca.crt \ - --ssl_cert_file module.crt \ - --ssl_key_file module.key \ - --iam_local_listen_port 31113 \ - --iam_local_ip 127.0.0.1 \ - --frontend_lease_bypass true \ - --force_low_reliability_instance true \ - --snapshot_storage_mode local_only \ - --checkpoint_dir "${CHECKPOINT_DIR}" \ - --enable_sandbox_router true \ - --enable_direct_routing false + --log-dir-prefix /home/yuanrong/sessions/master \ + -s "values.host_ip=\"${YR_NODE_IP}\"" \ + -s "values.local_ip=\"${YR_NODE_IP}\"" \ + -s "values.etcd.address=[{ip=\"${YR_NODE_IP}\",peer_port=${ETCD_PEER_PORT:-2378},port=${ETCD_PORT:-2379}}]" \ + -s 'values.fs.tls.enable=false' \ + -s 'values.node_proxy.enabled=true' \ + -s 'values.node_proxy.bind="0.0.0.0:9443"' \ + -s "values.node_proxy.advertise_address=\"${YR_NODE_IP}:9443\"" \ + -s 'values.node_proxy.health_bind="127.0.0.1:18443"' \ + -s 'values.node_proxy.allowed_target_cidrs=["10.88.0.0/16"]' \ + -s 'values.node_proxy.allow_any_edge=true' \ + -s 'values.node_proxy.edge_security_mode="network"' \ + -s 'values.node_proxy.activity_interval_sec=2' \ + -s 'mode.master.frontend=true' \ + -s 'mode.master.edge_frontend=true' \ + -s 'mode.master.meta_service=true' \ + -s 'mode.master.iam_server=true' \ + -s 'mode.master.function_scheduler=false' \ + -s 'values.frontend.port=8888' \ + -s 'values.frontend.ssl_enable=false' \ + -s 'values.frontend.client_auth_type="NoClientCert"' \ + -s 'values.frontend.enable_function_token_auth=true' \ + -s 'values.frontend.enable_func_token_auth=true' \ + -s 'values.frontend.frontend_lease_bypass=true' \ + -s 'values.frontend.sandbox_router_enable=false' \ + -s "values.frontend.meta_service_address=\"${YR_NODE_IP}:31182\"" \ + -s "values.frontend.iam_server_address=\"${YR_NODE_IP}:31112\"" \ + -s 'values.edge_frontend.tls_bind="0.0.0.0:8443"' \ + -s 'values.edge_frontend.plain_bind="0.0.0.0:8080"' \ + -s 'values.edge_frontend.health_bind="0.0.0.0:18080"' \ + -s "values.edge_frontend.frontend_address=\"${YR_NODE_IP}:8888\"" \ + -s 'values.edge_frontend.control_plane_routes=["exact:/","exact:/healthz","prefix:/terminal","prefix:/api/instances","prefix:/api/jobs","prefix:/api/sandbox","prefix:/functions","prefix:/api-docs","prefix:/admin/v1/functions","prefix:/serverless/v1/functions","prefix:/serverless/v1/stream","prefix:/serverless/v1/componentshealth","prefix:/serverless/v1/posix","prefix:/frontend/v1/instance","prefix:/datasystem/v1","prefix:/serverless/v2","prefix:/app/v1","prefix:/client/v1/lease","prefix:/invocations","prefix:/global-scheduler"]' \ + -s 'values.edge_frontend.tls_cert="/home/yuanrong/.cert/module.crt"' \ + -s 'values.edge_frontend.tls_key="/home/yuanrong/.cert/module.key"' \ + -s 'values.edge_frontend.validate_iam=true' \ + -s "values.edge_frontend.iam_address=\"${YR_NODE_IP}:31112\"" \ + -s 'values.edge_frontend.node_security_mode="network"' \ + -s 'values.edge_frontend.allowed_client_cidrs=[]' \ + -s 'values.edge_frontend.allow_any_client=true' \ + -s 'values.edge_frontend.log_level="info"' \ + -s 'values.meta_service.port=31182' \ + -s 'values.meta_service.ssl_enable=false' \ + -s 'values.iam_server.port=31112' \ + -s 'function_master.args.services_path="/home/yuanrong/deploy/process/services.yaml"' \ + -s 'function_proxy.args.services_path="/home/yuanrong/deploy/process/services.yaml"' \ + -s 'function_master.args.enable_traefik_provider=false' \ + -s 'function_master.args.system_timeout=60000' \ + -s 'function_proxy.args.system_timeout=60000' \ + -s 'function_proxy.args.metrics_collector_type="external"' \ + -s 'function_proxy.args.enable_inherit_env=true' \ + -s 'function_proxy.args.force_low_reliability_instance=true' else /usr/bin/yr start \ --ip_address "${YR_NODE_IP}" \ diff --git a/builder/systemd_services/yuanrong.service b/builder/systemd_services/yuanrong.service index 13a5486..a9d64ea 100644 --- a/builder/systemd_services/yuanrong.service +++ b/builder/systemd_services/yuanrong.service @@ -4,7 +4,7 @@ Description=yuanrong.service [Service] #Type=simple PIDFile=/run/yuanrong.pid -PassEnvironment=ETCD_PORT ETCD_PEER_PORT ETCD_ADDRESS HOSTNAME AKS_LOCAL_MODE AKERNEL_NODE_IP INSTANCE_IP LITEBUS_DATA_KEY YR_LOG_PATH YR_INSTALLATION_DIR YR_RRT_CONTROL_SOCKET_PATH ENABLE_METRICS ENABLE_TRACE TRAEFIK_MODE TRAEFIK_ENABLE_TLS TRAEFIK_HTTP_ENTRYPOINT +PassEnvironment=ETCD_PORT ETCD_PEER_PORT ETCD_ADDRESS HOSTNAME AKS_LOCAL_MODE AKERNEL_NODE_IP INSTANCE_IP LITEBUS_DATA_KEY YR_LOG_PATH YR_INSTALLATION_DIR YR_RRT_CONTROL_SOCKET_PATH ENABLE_METRICS ENABLE_TRACE TRAEFIK_MODE TRAEFIK_ENABLE_TLS TRAEFIK_HTTP_ENTRYPOINT YR_ETCD_ADDR_LIST YR_MASTER_ADDRESS YR_DATA_SYSTEM_CLUSTER_NAME YR_DATA_PLANE_NODE_PROXY_ENABLED YR_DATA_PLANE_NODE_PROXY_BIND YR_DATA_PLANE_NODE_PROXY_HEALTH_BIND YR_DATA_PLANE_NODE_PROXY_ALLOW_ANY_EDGE YR_DATA_PLANE_EDGE_FRONTEND_NODE_SECURITY_MODE YR_CPU_NUM_MILLICORES YR_MEMORY_NUM_MB YR_SHARED_MEMORY_NUM_MB Environment="CONTAINER_EP=unix:///run/sandboxd/sandboxd.sock" Environment="RUNTIME_HOME_DIR=/home/yuanrong/runtime" Environment="YR_NOSET_CUDA_VISIBLE_DEVICES=1" @@ -14,6 +14,7 @@ ExecStart=/usr/bin/bash /home/yuanrong/yr_node_bootstrap.sh ExecReload=/bin/kill -15 $MAINPID KillMode=control-group Restart=always +RestartSec=5s Delegate=yes UMask=000 TasksMax=infinity diff --git a/deploy/akernel/benchmark/fullscale_create.py b/deploy/akernel/benchmark/fullscale_create.py new file mode 100644 index 0000000..c694664 --- /dev/null +++ b/deploy/akernel/benchmark/fullscale_create.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python3 +"""Create node-pinned 1-core sandboxes for a full-cluster regression.""" + +from __future__ import annotations + +import os +from concurrent.futures import ThreadPoolExecutor, as_completed + +import yr_sandbox +import yr_sandbox.sandbox_api as sandbox_api + + +sandbox_api._NODE_ID_LABEL = "HOST_IP" + + +def targets() -> list[tuple[str, str, int]]: + result = [] + for raw in os.environ["AKERNEL_BENCH_TARGETS"].split(","): + group, host_ip, count = raw.split(":", 2) + result.append((group, host_ip, int(count))) + return result + + +def create_one(group: str, host_ip: str, index: int) -> tuple[str, str]: + sandbox = yr_sandbox.Sandbox( + runtime="runsc", + cpu=1000, + memory=96, + cpu_limit=1000, + mem_limit=512, + idle_timeout=3600, + schedule_timeout=60, + create_timeout=180, + name=f"regression-{group}-{index:02d}", + node_id=host_ip, + detached=True, + ) + return group, sandbox.id + + +def main() -> None: + jobs = [] + created: list[tuple[str, str]] = [] + try: + with ThreadPoolExecutor(max_workers=6) as executor: + for group, host_ip, count in targets(): + for index in range(count): + jobs.append(executor.submit(create_one, group, host_ip, index)) + for future in as_completed(jobs): + item = future.result() + created.append(item) + print(f"created group={item[0]} id={item[1]}", flush=True) + except BaseException: + for future in jobs: + future.cancel() + with ThreadPoolExecutor(max_workers=8) as executor: + list(executor.map(lambda item: yr_sandbox.Sandbox.delete(item[1]), created)) + raise + + created.sort() + with open("/state/placements", "w", encoding="utf-8") as output: + for group, sandbox_id in created: + output.write(f"{group} {sandbox_id}\n") + print(f"created_total={len(created)}", flush=True) + + +if __name__ == "__main__": + main() diff --git a/deploy/akernel/benchmark/fullscale_delete.py b/deploy/akernel/benchmark/fullscale_delete.py new file mode 100644 index 0000000..29e1f53 --- /dev/null +++ b/deploy/akernel/benchmark/fullscale_delete.py @@ -0,0 +1,13 @@ +#!/usr/bin/env python3 +"""Delete every sandbox recorded by fullscale_create.py.""" + +from concurrent.futures import ThreadPoolExecutor + +import yr_sandbox + + +with open("/state/placements", encoding="utf-8") as source: + sandbox_ids = [line.split(maxsplit=1)[1].strip() for line in source if line.strip()] +with ThreadPoolExecutor(max_workers=8) as executor: + results = list(executor.map(yr_sandbox.Sandbox.delete, sandbox_ids)) +print(f"deleted_total={len(results)}") diff --git a/deploy/akernel/benchmark/loadgen.Dockerfile b/deploy/akernel/benchmark/loadgen.Dockerfile new file mode 100644 index 0000000..270f83b --- /dev/null +++ b/deploy/akernel/benchmark/loadgen.Dockerfile @@ -0,0 +1,16 @@ +# syntax=docker/dockerfile:1.7 + +FROM python:3.12-slim + +ENV PYTHONDONTWRITEBYTECODE=1 \ + PYTHONUNBUFFERED=1 + +WORKDIR /opt/akernel-benchmark + +COPY sdk/python /tmp/akernel-sdk +RUN python -m pip install --no-cache-dir /tmp/akernel-sdk && \ + rm -rf /tmp/akernel-sdk + +COPY sdk/python/benchmarks/cluster_throughput.py ./cluster_throughput.py + +ENTRYPOINT ["python", "/opt/akernel-benchmark/cluster_throughput.py"] diff --git a/deploy/akernel/benchmark/raw_edge_sweep.py b/deploy/akernel/benchmark/raw_edge_sweep.py new file mode 100644 index 0000000..2ee6b6a --- /dev/null +++ b/deploy/akernel/benchmark/raw_edge_sweep.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python3 +"""Run multiple raw Edge benchmark processes over selected sandbox groups.""" + +from __future__ import annotations + +import argparse +import json +import subprocess +import sys +import time + + +def parse_limits(value: str) -> dict[str, int]: + if not value: + return {} + result = {} + for item in value.split(","): + group, count = item.split("=", 1) + result[group] = int(count) + return result + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--groups", required=True) + parser.add_argument("--group-limits", default="") + parser.add_argument("--connections-per-sandbox", type=int, default=4) + parser.add_argument("--shards", type=int, required=True) + parser.add_argument("--operations-per-connection", type=int, default=3000) + parser.add_argument("--host", required=True) + parser.add_argument("--port", type=int, default=443) + parser.add_argument("--action", default="ping") + args = parser.parse_args() + + selected_groups = set(args.groups.split(",")) + limits = parse_limits(args.group_limits) + by_group: dict[str, list[str]] = {} + with open("/state/placements", encoding="utf-8") as source: + for line in source: + group, sandbox_id = line.split(maxsplit=1) + if group in selected_groups: + by_group.setdefault(group, []).append(sandbox_id.strip()) + + sandboxes = [] + for group in sorted(selected_groups): + values = sorted(by_group.get(group, [])) + sandboxes.extend(values[: limits.get(group, len(values))]) + if not sandboxes: + raise SystemExit("no matching sandboxes") + + connections = [ + sandbox_id + for sandbox_id in sandboxes + for _ in range(args.connections_per_sandbox) + ] + shards = [[] for _ in range(min(args.shards, len(connections)))] + for index, sandbox_id in enumerate(connections): + shards[index % len(shards)].append(sandbox_id) + + processes = [] + started = time.perf_counter() + for shard in shards: + command = [ + sys.executable, + "/bench/edge_raw_throughput.py", + "--host", + args.host, + "--port", + str(args.port), + "--action", + args.action, + "--operations-per-sandbox", + str(args.operations_per_connection), + *shard, + ] + processes.append( + subprocess.Popen(command, stdout=subprocess.PIPE, text=True) + ) + + results = [] + for process in processes: + stdout, _ = process.communicate() + if process.returncode != 0: + raise SystemExit(f"benchmark shard failed with {process.returncode}") + results.append(json.loads(stdout)) + wall_seconds = time.perf_counter() - started + + operations = sum(item["operations"] for item in results) + weighted_mean = sum( + item["latency_ms"]["mean"] * item["operations"] for item in results + ) / operations + print( + json.dumps( + { + "action": args.action, + "groups": sorted(selected_groups), + "sandboxes": len(sandboxes), + "connections": len(connections), + "connections_per_sandbox": args.connections_per_sandbox, + "shards": len(shards), + "operations": operations, + "wall_seconds": wall_seconds, + "throughput_ops_s": operations / wall_seconds, + "shard_throughput_ops_s": [ + item["throughput_ops_s"] for item in results + ], + "latency_ms": { + "weighted_mean": weighted_mean, + "max_shard_p50": max(item["latency_ms"]["p50"] for item in results), + "max_shard_p95": max(item["latency_ms"]["p95"] for item in results), + "max_shard_p99": max(item["latency_ms"]["p99"] for item in results), + }, + }, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/deploy/akernel/charts/core/scripts/prepare_node.sh b/deploy/akernel/charts/core/scripts/prepare_node.sh index 6340469..26ffc57 100644 --- a/deploy/akernel/charts/core/scripts/prepare_node.sh +++ b/deploy/akernel/charts/core/scripts/prepare_node.sh @@ -5,5 +5,32 @@ # SPDX-License-Identifier: Apache-2.0 set -euo pipefail +install -d -m 0755 /home/akernel/filestore + +# sandboxd's bridge network requires the host br_netfilter module. The node +# container is privileged and mounts the host's matching module tree read-only. +if ! grep -q '^br_netfilter ' /proc/modules; then + modprobe br_netfilter +fi + +# Keep host scheduling and memory behavior consistent across CCE images. HCE +# disables scheduler autogrouping and enables THP globally; Ubuntu defaults are +# different and caused otherwise identical worker nodes to benchmark +# differently. These files are host-global even though this script runs from +# the privileged node pod. +if [ -w /proc/sys/kernel/sched_autogroup_enabled ]; then + echo 0 > /proc/sys/kernel/sched_autogroup_enabled +fi +thp_enabled=/host-sys/kernel/mm/transparent_hugepage/enabled +if [ -w "$thp_enabled" ]; then + echo always > "$thp_enabled" +fi +if [ -w /proc/sys/net/bridge/bridge-nf-call-iptables ]; then + echo 1 > /proc/sys/net/bridge/bridge-nf-call-iptables +fi +if [ -w /proc/sys/net/bridge/bridge-nf-call-ip6tables ]; then + echo 1 > /proc/sys/net/bridge/bridge-nf-call-ip6tables +fi + mkdir -p /etc/k8s_secrets mount --bind /run/secrets/kubernetes.io/serviceaccount /etc/k8s_secrets diff --git a/deploy/akernel/charts/core/templates/edge/edge_service.yaml b/deploy/akernel/charts/core/templates/edge/edge_service.yaml new file mode 100644 index 0000000..435c935 --- /dev/null +++ b/deploy/akernel/charts/core/templates/edge/edge_service.yaml @@ -0,0 +1,26 @@ +{{- if and .Values.dataPlane.enabled .Values.dataPlane.edge.enabled }} +apiVersion: v1 +kind: Service +metadata: + name: akernel-edge + namespace: {{ .Release.Namespace }} + labels: + app: akernel-edge + {{- with .Values.dataPlane.edge.service.annotations }} + annotations: + {{- toYaml . | nindent 4 }} + {{- end }} +spec: + type: {{ .Values.dataPlane.edge.service.type }} + ports: + - name: tls + port: {{ .Values.dataPlane.edge.service.tlsPort }} + targetPort: edge-tls + protocol: TCP + - name: plain + port: {{ .Values.dataPlane.edge.service.plainPort }} + targetPort: edge-plain + protocol: TCP + selector: + app: akernel-frontend +{{- end }} diff --git a/deploy/akernel/charts/core/templates/frontend/akernel_frontend.yaml b/deploy/akernel/charts/core/templates/frontend/akernel_frontend.yaml index 2b02967..9be646f 100644 --- a/deploy/akernel/charts/core/templates/frontend/akernel_frontend.yaml +++ b/deploy/akernel/charts/core/templates/frontend/akernel_frontend.yaml @@ -102,6 +102,51 @@ spec: value: "false" - name: ENABLE_IAM_SERVER value: "true" + - name: ENABLE_SANDBOX_ROUTER + value: {{ not .Values.dataPlane.enabled | quote }} + {{- if .Values.dataPlane.enabled }} + # The frontend has no sandboxd/node-proxy. Advertising worker + # capacity here lets the global scheduler place sandboxes on a + # control-plane-only agent, where runtime creation fails. + - name: YR_CPU_NUM_MILLICORES + value: "0" + - name: YR_MEMORY_NUM_MB + value: "0" + {{- end }} + {{- if and .Values.dataPlane.enabled .Values.dataPlane.edge.enabled }} + - name: ENABLE_EDGE_FRONTEND + value: "true" + - name: INSTANCE_IP + valueFrom: + fieldRef: + fieldPath: status.podIP + - name: YR_ETCD_ADDR_LIST + value: {{ printf "akernel-etcd.%s.svc.cluster.local:%v" .Release.Namespace (get $frontendEtcd "port" | default "2379") | quote }} + - name: YR_DATA_SYSTEM_CLUSTER_NAME + value: {{ .Values.dataPlane.clusterName | quote }} + - name: YR_DATA_PLANE_EDGE_FRONTEND_TLS_BIND + value: {{ printf "0.0.0.0:%v" .Values.dataPlane.edge.tlsPort | quote }} + - name: YR_DATA_PLANE_EDGE_FRONTEND_PLAIN_BIND + value: {{ printf "0.0.0.0:%v" .Values.dataPlane.edge.plainPort | quote }} + - name: YR_DATA_PLANE_EDGE_FRONTEND_HEALTH_BIND + value: {{ printf "0.0.0.0:%v" .Values.dataPlane.edge.healthPort | quote }} + - name: YR_DATA_PLANE_EDGE_FRONTEND_CONTROL_PLANE_ADDRESS + value: "$(INSTANCE_IP):8888" + - name: YR_DATA_PLANE_EDGE_FRONTEND_IAM_ADDRESS + value: "$(INSTANCE_IP):31112" + - name: YR_DATA_PLANE_EDGE_FRONTEND_VALIDATE_IAM + value: {{ .Values.dataPlane.edge.validateIam | quote }} + - name: YR_DATA_PLANE_EDGE_FRONTEND_ACCESS_LOG_ENABLED + value: {{ .Values.dataPlane.edge.accessLogEnabled | quote }} + - name: YR_DATA_PLANE_EDGE_FRONTEND_AUDIT_LOG_ENABLED + value: {{ .Values.dataPlane.edge.auditLogEnabled | quote }} + - name: YR_DATA_PLANE_EDGE_FRONTEND_ALLOWED_CLIENT_CIDRS + value: {{ join "," .Values.dataPlane.edge.allowedClientCidrs | quote }} + - name: YR_DATA_PLANE_EDGE_FRONTEND_ALLOW_ANY_CLIENT + value: {{ .Values.dataPlane.edge.allowAnyClient | quote }} + - name: YR_DATA_PLANE_EDGE_FRONTEND_NODE_SECURITY_MODE + value: {{ .Values.dataPlane.node.securityMode | quote }} + {{- end }} - name: META_SERVICE_ADDRESS value: {{ .Values.frontend.master.metaServiceAddress | default (printf "akernel-master.%s.svc.cluster.local:31111" .Release.Namespace) | quote }} - name: IAM_SERVER_ADDRESS @@ -131,20 +176,44 @@ spec: ports: - name: http containerPort: 8888 + - name: iam + containerPort: 31112 + {{- if and .Values.dataPlane.enabled .Values.dataPlane.edge.enabled }} + - name: edge-tls + containerPort: {{ .Values.dataPlane.edge.tlsPort }} + protocol: TCP + - name: edge-plain + containerPort: {{ .Values.dataPlane.edge.plainPort }} + protocol: TCP + - name: edge-health + containerPort: {{ .Values.dataPlane.edge.healthPort }} + protocol: TCP + {{- end }} livenessProbe: httpGet: path: /healthz + {{- if and .Values.dataPlane.enabled .Values.dataPlane.edge.enabled }} + port: edge-health + scheme: HTTP + {{- else }} port: 8888 scheme: HTTPS + {{- end }} initialDelaySeconds: 15 periodSeconds: 20 timeoutSeconds: 5 failureThreshold: 3 readinessProbe: httpGet: + {{- if and .Values.dataPlane.enabled .Values.dataPlane.edge.enabled }} + path: /readyz + port: edge-health + scheme: HTTP + {{- else }} path: /healthz port: 8888 scheme: HTTPS + {{- end }} initialDelaySeconds: 5 periodSeconds: 10 timeoutSeconds: 3 diff --git a/deploy/akernel/charts/core/templates/master/akernel_master.yaml b/deploy/akernel/charts/core/templates/master/akernel_master.yaml index de37591..db4a5ee 100644 --- a/deploy/akernel/charts/core/templates/master/akernel_master.yaml +++ b/deploy/akernel/charts/core/templates/master/akernel_master.yaml @@ -90,10 +90,22 @@ spec: secretKeyRef: name: {{ include "core.litebusSecretName" . }} key: litebus-data-key + {{- if .Values.dataPlane.enabled }} + # The master participates in the YuanRong control plane but has no + # sandboxd/node-proxy, so it must not advertise worker capacity. + - name: YR_CPU_NUM_MILLICORES + value: "0" + - name: YR_MEMORY_NUM_MB + value: "0" + {{- end }} - name: ETCD_ADDRESS value: {{ get $masterEtcd "host" | default (printf "akernel-etcd.%s.svc.cluster.local" .Release.Namespace) | quote }} - name: ETCD_PORT value: {{ get $masterEtcd "port" | default "2379" | quote }} + {{- if .Values.dataPlane.enabled }} + - name: YR_DATA_SYSTEM_CLUSTER_NAME + value: {{ .Values.dataPlane.clusterName | quote }} + {{- end }} {{- if .Values.monitoring.prometheusEndpoint }} - name: PROMETHEUS_ENDPOINT value: {{ .Values.monitoring.prometheusEndpoint | quote }} @@ -113,7 +125,7 @@ spec: - name: ENABLE_TRACE value: {{ ne (.Values.monitoring.tempoEndpoint | default "") "" | ternary "true" "false" | quote }} - name: TRAEFIK_MODE - value: {{ .Values.traefik.mode | default "http" | quote }} + value: {{ .Values.dataPlane.enabled | ternary "disabled" (.Values.traefik.mode | default "http") | quote }} - name: TRAEFIK_ENABLE_TLS value: {{ .Values.traefik.enableWebEntrypoint | default false | ternary false (.Values.traefik.enableTLS | default false) | quote }} - name: TRAEFIK_HTTP_ENTRYPOINT @@ -132,6 +144,14 @@ spec: periodSeconds: 20 timeoutSeconds: 5 failureThreshold: 3 + {{- if .Values.dataPlane.enabled }} + startupProbe: + tcpSocket: + port: 22770 + periodSeconds: 10 + timeoutSeconds: 5 + failureThreshold: 36 + {{- end }} readinessProbe: tcpSocket: port: 22770 diff --git a/deploy/akernel/charts/core/templates/node/configmap.yaml b/deploy/akernel/charts/core/templates/node/configmap.yaml index bfc8989..092ac22 100644 --- a/deploy/akernel/charts/core/templates/node/configmap.yaml +++ b/deploy/akernel/charts/core/templates/node/configmap.yaml @@ -15,6 +15,14 @@ data: sandboxd_config.toml: | {{- $sandboxdConfig := .Values.node.config.sandboxd.config }} {{- $runcMarker := "# AKERNEL_RUNTIME_RUNC" }} +{{- $defaultCgroupRoot := "cgroup_root_name=\"/akernel\"" }} +{{- $cgroupRootName := .Values.node.config.sandboxd.cgroupRootName | default "/akernel" }} +{{- if ne $cgroupRootName "/akernel" }} +{{- if not (contains $defaultCgroupRoot $sandboxdConfig) }} +{{- fail "custom node.config.sandboxd.cgroupRootName requires the default cgroup_root_name marker in node.config.sandboxd.config" }} +{{- end }} +{{- $sandboxdConfig = replace $defaultCgroupRoot (printf "cgroup_root_name=\"%s\"" $cgroupRootName) $sandboxdConfig }} +{{- end }} {{- if .Values.node.config.sandboxd.enableRunc }} {{- if not (contains $runcMarker $sandboxdConfig) }} {{- fail "node.config.sandboxd.enableRunc requires the # AKERNEL_RUNTIME_RUNC marker in node.config.sandboxd.config" }} diff --git a/deploy/akernel/charts/core/templates/node/daemonset.yaml b/deploy/akernel/charts/core/templates/node/daemonset.yaml index ba6a974..16190e3 100644 --- a/deploy/akernel/charts/core/templates/node/daemonset.yaml +++ b/deploy/akernel/charts/core/templates/node/daemonset.yaml @@ -106,6 +106,24 @@ spec: fieldRef: apiVersion: v1 fieldPath: status.podIP + {{- if .Values.dataPlane.enabled }} + - name: YR_DATA_PLANE_NODE_PROXY_ENABLED + value: "true" + - name: YR_MASTER_ADDRESS + value: {{ printf "akernel-master.%s.svc.cluster.local" .Release.Namespace | quote }} + - name: YR_ETCD_ADDR_LIST + value: {{ printf "akernel-etcd.%s.svc.cluster.local:%v" .Release.Namespace (get $nodeEtcd "port" | default "2379") | quote }} + - name: YR_DATA_SYSTEM_CLUSTER_NAME + value: {{ .Values.dataPlane.clusterName | quote }} + - name: YR_DATA_PLANE_NODE_PROXY_BIND + value: {{ printf "0.0.0.0:%v" .Values.dataPlane.node.port | quote }} + - name: YR_DATA_PLANE_NODE_PROXY_HEALTH_BIND + value: {{ printf "127.0.0.1:%v" .Values.dataPlane.node.healthPort | quote }} + - name: YR_DATA_PLANE_NODE_PROXY_ALLOW_ANY_EDGE + value: {{ .Values.dataPlane.node.allowAnyEdge | quote }} + - name: YR_DATA_PLANE_EDGE_FRONTEND_NODE_SECURITY_MODE + value: {{ .Values.dataPlane.node.securityMode | quote }} + {{- end }} {{- if .Values.monitoring.prometheusEndpoint }} - name: PROMETHEUS_ENDPOINT value: {{ .Values.monitoring.prometheusEndpoint | quote }} @@ -125,11 +143,17 @@ spec: - name: ENABLE_TRACE value: {{ ne (.Values.monitoring.tempoEndpoint | default "") "" | ternary "true" "false" | quote }} - name: TRAEFIK_MODE - value: {{ .Values.traefik.mode | default "http" | quote }} + value: {{ .Values.dataPlane.enabled | ternary "disabled" (.Values.traefik.mode | default "http") | quote }} - name: TRAEFIK_ENABLE_TLS value: {{ .Values.traefik.enableWebEntrypoint | default false | ternary false (.Values.traefik.enableTLS | default false) | quote }} - name: TRAEFIK_HTTP_ENTRYPOINT value: {{ .Values.traefik.enableWebEntrypoint | default false | ternary "web" "websecure" | quote }} + {{- if .Values.dataPlane.enabled }} + ports: + - name: node-proxy + containerPort: {{ .Values.dataPlane.node.port }} + protocol: TCP + {{- end }} securityContext: privileged: true volumeMounts: @@ -158,6 +182,11 @@ spec: subPath: config.json - mountPath: /dev/shm name: shm + - mountPath: /lib/modules + name: host-modules + readOnly: true + - mountPath: /host-sys/kernel/mm/transparent_hugepage + name: host-thp - mountPath: /etc/systemd/system/system.slice.d/service-resource-control.conf name: akernel-node-config subPath: service-resource-control.conf @@ -226,6 +255,14 @@ spec: emptyDir: medium: Memory sizeLimit: 500Mi + - name: host-modules + hostPath: + path: /lib/modules + type: Directory + - name: host-thp + hostPath: + path: /sys/kernel/mm/transparent_hugepage + type: Directory {{- if or .Values.monitoring.prometheusEndpoint .Values.monitoring.lokiEndpoint .Values.monitoring.tempoEndpoint }} - name: otel-collector-config configMap: diff --git a/deploy/akernel/charts/core/values.yaml b/deploy/akernel/charts/core/values.yaml index ac28507..86072f9 100644 --- a/deploy/akernel/charts/core/values.yaml +++ b/deploy/akernel/charts/core/values.yaml @@ -97,6 +97,11 @@ master: frontend: enabled: true replicas: 2 + # sandboxd creates runtime processes outside Kubernetes' resource accounting. + # For high-throughput data-plane deployments, use required node affinity and + # a matching taint toleration to keep the shared Edge off sandbox workers. + affinity: {} + tolerations: [] service: type: ClusterIP annotations: {} @@ -119,6 +124,37 @@ frontend: metaServiceAddress: "" iamServerAddress: "" +# Rust edge frontend (co-located in the frontend container) plus the per-node +# Rust proxy. Disabled by default so +# existing releases retain the legacy Traefik/sandboxRouter topology until +# explicitly migrated. +dataPlane: + enabled: false + clusterName: "" + edge: + enabled: true + tlsPort: 8443 + plainPort: 8080 + healthPort: 18080 + validateIam: true + # High-volume per-request access records may be disabled independently of + # security audit records when the Edge binary supports split log controls. + accessLogEnabled: true + auditLogEnabled: true + allowAnyClient: false + allowedClientCidrs: + - 127.0.0.0/8 + service: + type: ClusterIP + tlsPort: 443 + plainPort: 80 + annotations: {} + node: + port: 9443 + healthPort: 18443 + allowAnyEdge: true + securityMode: network + node: secret: create: true @@ -131,7 +167,9 @@ node: pullPolicy: "" resources: limits: - ephemeral-storage: 1Gi + # Runtime/session logs share the container writable layer. Keep enough + # headroom for a full-node sandbox benchmark without kubelet eviction. + ephemeral-storage: 10Gi requests: ephemeral-storage: 1Gi annotations: {} @@ -445,6 +483,8 @@ node: # The node image must also be built with AKERNEL_ENABLE_RUNC=true. # Sandboxd probes runc and runc-shim before advertising the runtime. enableRunc: false + # Must be unique when multiple AKernel releases share a physical node. + cgroupRootName: "/akernel" config: |- rootDir="/home/akernel/sandboxd/root" storeDir="/home/akernel/sandboxd/store" diff --git a/deploy/akernel/charts/monitor/templates/prometheus/pvc.yaml b/deploy/akernel/charts/monitor/templates/prometheus/pvc.yaml index 3a3ba97..50c5094 100644 --- a/deploy/akernel/charts/monitor/templates/prometheus/pvc.yaml +++ b/deploy/akernel/charts/monitor/templates/prometheus/pvc.yaml @@ -1,8 +1,10 @@ apiVersion: v1 kind: PersistentVolumeClaim metadata: - name: prometheus-pvc-ssd + name: {{ .Values.prometheusServer.persistence.claimName | default "prometheus-pvc-ssd" }} namespace: {{ .Release.Namespace }} + annotations: + helm.sh/resource-policy: keep labels: {{- include "monitor.labels" . | nindent 4 }} spec: diff --git a/deploy/akernel/charts/monitor/templates/prometheus/statefulset.yaml b/deploy/akernel/charts/monitor/templates/prometheus/statefulset.yaml index 942b4c6..391cc12 100644 --- a/deploy/akernel/charts/monitor/templates/prometheus/statefulset.yaml +++ b/deploy/akernel/charts/monitor/templates/prometheus/statefulset.yaml @@ -116,4 +116,4 @@ spec: name: prometheus-config - name: prometheus-storage persistentVolumeClaim: - claimName: prometheus-pvc-ssd + claimName: {{ .Values.prometheusServer.persistence.claimName | default "prometheus-pvc-ssd" }} diff --git a/deploy/akernel/charts/monitor/values.yaml b/deploy/akernel/charts/monitor/values.yaml index ebca533..6183460 100644 --- a/deploy/akernel/charts/monitor/values.yaml +++ b/deploy/akernel/charts/monitor/values.yaml @@ -187,6 +187,7 @@ prometheusServer: ephemeral-storage: "1Gi" persistence: + claimName: prometheus-pvc-ssd storageClass: "" size: 20Gi accessMode: ReadWriteOnce diff --git a/deploy/scripts/build-image.sh b/deploy/scripts/build-image.sh index 53acd8d..1a65752 100755 --- a/deploy/scripts/build-image.sh +++ b/deploy/scripts/build-image.sh @@ -31,6 +31,9 @@ open_yr_core_wheel_url="${OPEN_YR_CORE_WHEEL_URL:-}" open_yr_core_wheel_sha256="${OPEN_YR_CORE_WHEEL_SHA256:-}" rrt_runtime_url="${RRT_RUNTIME_URL:-}" rrt_runtime_sha256="${RRT_RUNTIME_SHA256:-}" +open_yr_data_plane_wheel_path="${OPEN_YR_DATA_PLANE_WHEEL_PATH:-}" +open_yr_data_plane_wheel_url="${OPEN_YR_DATA_PLANE_WHEEL_URL:-}" +open_yr_data_plane_wheel_sha256="${OPEN_YR_DATA_PLANE_WHEEL_SHA256:-}" print_component_versions=0 component_revision() { @@ -111,6 +114,18 @@ while [[ $# -gt 0 ]]; do rrt_runtime_sha256="$2" shift 2 ;; + --open-yr-data-plane-wheel-path) + open_yr_data_plane_wheel_path="$2" + shift 2 + ;; + --open-yr-data-plane-wheel-url) + open_yr_data_plane_wheel_url="$2" + shift 2 + ;; + --open-yr-data-plane-wheel-sha256) + open_yr_data_plane_wheel_sha256="$2" + shift 2 + ;; --print-component-versions) print_component_versions=1 shift @@ -138,6 +153,25 @@ esac require_cmd docker +docker_build_common_args=() +if [[ -n "${AKERNEL_BUILD_NETWORK:-}" ]]; then + case "${AKERNEL_BUILD_NETWORK}" in + default|host|none) ;; + *) die "AKERNEL_BUILD_NETWORK must be default, host, or none" ;; + esac + docker_build_common_args+=(--network "${AKERNEL_BUILD_NETWORK}") + if [[ "${AKERNEL_BUILD_NETWORK}" == "host" ]]; then + for proxy_name in \ + HTTP_PROXY HTTPS_PROXY ALL_PROXY NO_PROXY \ + http_proxy https_proxy all_proxy no_proxy; do + proxy_value="${!proxy_name:-}" + if [[ -n "${proxy_value}" ]]; then + docker_build_common_args+=(--build-arg "${proxy_name}=${proxy_value}") + fi + done + fi +fi + if [[ -n "${env_name}" && -f "$(state_dir "${env_name}")/config.env" ]]; then load_env_config "${env_name}" repository="${repository:-${IMAGE_REPOSITORY}}" @@ -194,9 +228,45 @@ if [[ -n "${rrt_runtime_url}" || -n "${rrt_runtime_sha256}" ]]; then ) fi +if [[ -n "${open_yr_data_plane_wheel_path}" && -n "${open_yr_data_plane_wheel_url}" ]]; then + die "OPEN_YR_DATA_PLANE_WHEEL_PATH and OPEN_YR_DATA_PLANE_WHEEL_URL are mutually exclusive" +fi +if [[ -z "${open_yr_data_plane_wheel_path}" && -z "${open_yr_data_plane_wheel_url}" ]]; then + die "the Rust data-plane wheel is required; set OPEN_YR_DATA_PLANE_WHEEL_PATH or OPEN_YR_DATA_PLANE_WHEEL_URL" +fi +if [[ -z "${open_yr_data_plane_wheel_sha256}" ]]; then + die "OPEN_YR_DATA_PLANE_WHEEL_SHA256 is required with the Rust data-plane wheel" +fi + +node_build_contexts=(--build-context "open_yr_data_plane_wheel=.") +data_plane_context_dir="" +open_yr_data_plane_wheel_name="" +if [[ -n "${open_yr_data_plane_wheel_path}" ]]; then + if [[ ! -f "${open_yr_data_plane_wheel_path}" ]]; then + die "Rust data-plane wheel does not exist: ${open_yr_data_plane_wheel_path}" + fi + actual_data_plane_sha256="$(sha256sum "${open_yr_data_plane_wheel_path}" | awk '{print $1}')" + if [[ "${actual_data_plane_sha256}" != "${open_yr_data_plane_wheel_sha256}" ]]; then + die "Rust data-plane wheel checksum mismatch: expected ${open_yr_data_plane_wheel_sha256}, got ${actual_data_plane_sha256}" + fi + open_yr_data_plane_wheel_name="$(basename "${open_yr_data_plane_wheel_path}")" + case "${open_yr_data_plane_wheel_name}" in + openyuanrong_data_plane-*.whl) ;; + *) die "Rust data-plane path must retain its openyuanrong_data_plane wheel filename" ;; + esac + data_plane_context_dir="$(mktemp -d)" + trap 'rm -rf -- "${data_plane_context_dir}"' EXIT + ln "${open_yr_data_plane_wheel_path}" \ + "${data_plane_context_dir}/${open_yr_data_plane_wheel_name}" + node_build_contexts=( + --build-context "open_yr_data_plane_wheel=${data_plane_context_dir}" + ) +fi + info "building ${runtime_image} with runtime profile ${runtime_profile}" docker build \ -f builder/runtime.Dockerfile \ + "${docker_build_common_args[@]}" \ --target "runtime-${runtime_profile}" \ "${runtime_build_args[@]}" \ -t "${runtime_image}" \ @@ -241,8 +311,16 @@ if [[ -n "${open_yr_core_wheel_url}" || -n "${open_yr_core_wheel_sha256}" ]]; th --build-arg "OPEN_YR_CORE_WHEEL_SHA256=${open_yr_core_wheel_sha256}" ) fi + +node_build_args+=( + --build-arg "OPEN_YR_DATA_PLANE_WHEEL_URL=${open_yr_data_plane_wheel_url}" + --build-arg "OPEN_YR_DATA_PLANE_WHEEL_SHA256=${open_yr_data_plane_wheel_sha256}" + --build-arg "OPEN_YR_DATA_PLANE_WHEEL_NAME=${open_yr_data_plane_wheel_name}" +) docker build \ -f builder/node.Dockerfile \ + "${docker_build_common_args[@]}" \ + "${node_build_contexts[@]}" \ "${node_build_args[@]}" \ -t "${all_in_one_image}" \ . diff --git a/deploy/standalone/README.md b/deploy/standalone/README.md index b24912b..bbfe1bc 100644 --- a/deploy/standalone/README.md +++ b/deploy/standalone/README.md @@ -8,9 +8,26 @@ two containers on the default container bridge: - `akernel-traefik` runs the official Traefik image as the external gateway. Keeping the gateway in a separate network namespace allows sandboxd's normal -`PREROUTING` rules to handle gateway traffic. The all-in-one frontend sends -traffic from the node network namespace, so the standalone sandboxd config -also enables its local-output DNAT support. +`PREROUTING` rules to handle gateway traffic. Traefik passes HTTPS/TLS through +to the Rust Edge Frontend on port 8443 and forwards plain HTTP to Edge on port +8080. Edge sends control requests to the core frontend and sandbox data traffic +through the Rust Node Proxy. The all-in-one components send traffic from the +node network namespace, so the standalone sandboxd config also enables its +local-output DNAT support. + +Building the all-in-one image requires the separately published Rust data-plane +wheel. Download it first, then provide its absolute path and SHA-256: + +```bash +make build IMAGE_REPOSITORY=akernel-local IMAGE_TAG=rust-data-plane \ + OPEN_YR_DATA_PLANE_WHEEL_PATH=/absolute/path/openyuanrong_data_plane.whl \ + OPEN_YR_DATA_PLANE_WHEEL_SHA256= +``` + +The local wheel is mounted into the Docker build through a dedicated read-only +BuildKit context and is not retained in an image layer. If a public artifact +URL becomes available, use `OPEN_YR_DATA_PLANE_WHEEL_URL` instead of +`OPEN_YR_DATA_PLANE_WHEEL_PATH`. The default runtime is gVisor `runsc`. The bundled image also contains Kata Containers and Firecracker. Both `Sandbox(runtime="kata")` and @@ -18,6 +35,10 @@ Containers and Firecracker. Both `Sandbox(runtime="kata")` and virtualization on the Docker host. Nodes without KVM remain usable with runsc and do not advertise either VM runtime to the scheduler. +`runsc` uses the `systrap` platform by default. Hosts that expose `/dev/kvm` +may select gVisor's KVM platform for standalone startup with +`AKERNEL_RUNSC_PLATFORM=kvm ./start.sh`. + See the maintained [runtime selection example](../../sdk/python/examples/sandbox_runtime.py) for client usage. @@ -167,8 +188,9 @@ This will: - Start the privileged AKernel all-in-one container - Start an independent Traefik container for the HTTPS API and HTTP sandbox port-forwarding gateway -- Configure Traefik to poll FunctionMaster's HTTP provider for per-sandbox - tunnel routes, including custom tunnel ports +- Configure Traefik with TLS passthrough and HTTP forwarding to the Rust Edge + Frontend +- Wait for the Rust Edge Frontend readiness endpoint before exposing it - Generate a deployment-specific IAM signing seed and a 24-hour SDK token - Generate a sandboxd config using `AKERNEL_NAT_BACKEND` (`iptables` by default) @@ -247,9 +269,20 @@ needed: TRAEFIK_IMAGE="traefik:v3.6.8" ./start.sh ``` +To run an isolated second standalone instance on the same Docker host, give it +distinct container names and a distinct data directory. Pass the same names to +`stop.sh`: + +```bash +AKERNEL_NODE_CONTAINER_NAME=akernel-rustdp-node \ +AKERNEL_TRAEFIK_CONTAINER_NAME=akernel-rustdp-traefik \ +AKERNEL_STANDALONE_DATA_DIR=/absolute/path/to/rustdp-data \ +IMAGE=akernel-local:rust-data-plane ./start.sh +``` + ### Data Directory Location -By default, data is stored in `./data`. To change this, edit `start.sh`: +By default, data is stored in `./data`. Override it without editing the script: ```bash -DATA_DIR="/path/to/your/data" +AKERNEL_STANDALONE_DATA_DIR="/path/to/your/data" ./start.sh ``` diff --git a/deploy/standalone/config/sandboxd_config.toml b/deploy/standalone/config/sandboxd_config.toml index ea41dd1..58a1693 100644 --- a/deploy/standalone/config/sandboxd_config.toml +++ b/deploy/standalone/config/sandboxd_config.toml @@ -39,6 +39,9 @@ filestore_xfs_enabled=false loop_device_dir="/dev" overlay_tmpfs_size="10G" +[plugin.runtime.runsc] +platform="systrap" + [plugin.runtime.basic_spec] runsc="/home/akernel/images/config.json" runc="" diff --git a/deploy/standalone/start.sh b/deploy/standalone/start.sh index 1fff5d2..79d4ad4 100755 --- a/deploy/standalone/start.sh +++ b/deploy/standalone/start.sh @@ -10,18 +10,18 @@ set -e # Configuration SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" CONFIG_DIR="${SCRIPT_DIR}/config" -DATA_DIR="${SCRIPT_DIR}/data" -FRONTEND_PORT="8888" +DATA_DIR="${AKERNEL_STANDALONE_DATA_DIR:-${SCRIPT_DIR}/data}" ETCD_PORT="${ETCD_PORT:-2379}" ETCD_PEER_PORT="${ETCD_PEER_PORT:-2378}" -NODE_CONTAINER_NAME="akernel-node" -TRAEFIK_CONTAINER_NAME="akernel-traefik" +NODE_CONTAINER_NAME="${AKERNEL_NODE_CONTAINER_NAME:-akernel-node}" +TRAEFIK_CONTAINER_NAME="${AKERNEL_TRAEFIK_CONTAINER_NAME:-akernel-traefik}" IMAGE="${IMAGE:-akerneldev/all-in-one:latest}" TRAEFIK_IMAGE="${TRAEFIK_IMAGE:-traefik:v3.6.8}" IAM_SEED_FILE="${DATA_DIR}/iam-seed" TOKEN_FILE="${DATA_DIR}/token" SANDBOXD_CONFIG_FILE="${DATA_DIR}/sandboxd/config.toml" AKERNEL_NAT_BACKEND="${AKERNEL_NAT_BACKEND:-iptables}" +AKERNEL_RUNSC_PLATFORM="${AKERNEL_RUNSC_PLATFORM:-systrap}" AKERNEL_ENABLE_RUNC="${AKERNEL_ENABLE_RUNC:-false}" LITEBUS_DATA_KEY="" @@ -104,6 +104,21 @@ check_prerequisites() { ;; esac + case "${AKERNEL_RUNSC_PLATFORM}" in + systrap) + ;; + kvm) + if [[ ! -c /dev/kvm ]]; then + log_error "AKERNEL_RUNSC_PLATFORM=kvm requires /dev/kvm" + exit 1 + fi + ;; + *) + log_error "AKERNEL_RUNSC_PLATFORM must be 'systrap' or 'kvm'" + exit 1 + ;; + esac + # Create data directory mkdir -p "${DATA_DIR}" log_info "Data directory: ${DATA_DIR}" @@ -241,6 +256,7 @@ configure_network() { local sed_args=( -E -e "s/^[[:space:]]*nat_backend[[:space:]]*=.*/nat_backend=\"${AKERNEL_NAT_BACKEND}\"/" + -e "s/^[[:space:]]*platform[[:space:]]*=.*/platform=\"${AKERNEL_RUNSC_PLATFORM}\"/" ) case "${AKERNEL_NAT_BACKEND}" in @@ -257,6 +273,11 @@ configure_network() { log_error "Missing nat_backend in ${CONFIG_DIR}/sandboxd_config.toml" exit 1 fi + if ! grep -q '^[[:space:]]*platform[[:space:]]*=' \ + "${CONFIG_DIR}/sandboxd_config.toml"; then + log_error "Missing runsc platform in ${CONFIG_DIR}/sandboxd_config.toml" + exit 1 + fi if [[ "${AKERNEL_ENABLE_RUNC}" == "true" ]]; then if ! grep -q '^[[:space:]]*# AKERNEL_RUNTIME_RUNC[[:space:]]*$' \ "${CONFIG_DIR}/sandboxd_config.toml"; then @@ -267,6 +288,8 @@ configure_network() { -e 's|^[[:space:]]*# AKERNEL_RUNTIME_RUNC[[:space:]]*$|runc="/usr/local/bin/runc"|' ) fi + + log_info "Using the runsc ${AKERNEL_RUNSC_PLATFORM} platform" sed "${sed_args[@]}" "${CONFIG_DIR}/sandboxd_config.toml" > "${config_tmp}" mv "${config_tmp}" "${SANDBOXD_CONFIG_FILE}" @@ -283,25 +306,26 @@ configure_network() { prepare_host_network_modules() { local modprobe_bin - modprobe_bin="$(command -v modprobe || true)" - if [[ -z "${modprobe_bin}" ]]; then - log_error "modprobe is required to load AKernel host network modules" - exit 1 - fi - - if [[ "$(id -u)" -eq 0 ]]; then - "${modprobe_bin}" tun - elif sudo -n "${modprobe_bin}" tun; then - : - else - log_error "Unable to load tun; run this script as root or allow passwordless sudo for modprobe" - exit 1 + if [[ ! -c /dev/net/tun ]]; then + modprobe_bin="$(command -v modprobe || true)" + if [[ -z "${modprobe_bin}" ]]; then + log_error "modprobe is required to load the missing tun module" + exit 1 + fi + if [[ "$(id -u)" -eq 0 ]]; then + "${modprobe_bin}" tun + elif sudo -n "${modprobe_bin}" tun; then + : + else + log_error "Unable to load tun; run this script as root or allow passwordless sudo for modprobe" + exit 1 + fi fi if [[ ! -c /dev/net/tun ]]; then log_error "tun loaded but /dev/net/tun is unavailable" exit 1 fi - log_info "Loaded host tun module for pooled TAP networking" + log_info "Host tun device is ready for pooled TAP networking" if [[ "${AKERNEL_NAT_BACKEND}" != "iptables" ]]; then return 0 @@ -352,9 +376,6 @@ prepare_host_network_modules() { # from the gateway enters this network namespace through PREROUTING. start_node_container() { log_info "Starting container: ${NODE_CONTAINER_NAME}" - # FunctionMaster's HTTP provider publishes the per-sandbox routes required - # by reverse tunnels; the legacy etcd mode cannot publish those routes. - "${DOCKER_PREFIX[@]}" ${DOCKER_CMD} run -d \ --name "${NODE_CONTAINER_NAME}" \ --privileged \ @@ -368,7 +389,7 @@ start_node_container() { -e ETCD_PORT="${ETCD_PORT}" \ -e ETCD_PEER_PORT="${ETCD_PEER_PORT}" \ -e NODE_NAME="$(hostname)" \ - -e POD_NAME=akernel-node-local \ + -e POD_NAME="${NODE_CONTAINER_NAME}-local" \ -e POD_NAMESPACE=default \ -e TZ=Asia/Shanghai \ -e ENABLE_TRACE="${ENABLE_TRACE:-false}" \ @@ -419,38 +440,39 @@ write_traefik_config() { mkdir -p "${traefik_dir}" cat > "${traefik_dir}/dynamic.yml" < /dev/null; then + log_info "Rust Edge Frontend is ready" + return 0 + fi + + if ! "${DOCKER_PREFIX[@]}" ${DOCKER_CMD} inspect \ + --format '{{.State.Running}}' "${NODE_CONTAINER_NAME}" 2> /dev/null \ + | grep -q true; then + log_error "AKernel node exited while the Rust data plane was starting" + return 1 + fi + + if [[ ${i} -eq ${retries} ]]; then + log_error "Rust Edge Frontend did not become ready" + "${DOCKER_PREFIX[@]}" ${DOCKER_CMD} exec "${NODE_CONTAINER_NAME}" \ + systemctl --no-pager --full status yuanrong.service || true + return 1 + fi + sleep ${delay} + done +} + wait_for_gateway() { local traefik_ip="$1" local retries=60 @@ -538,9 +587,8 @@ if [[ -z "${NODE_IP}" ]]; then exit 1 fi write_traefik_config "${NODE_IP}" -TRAEFIK_PROVIDER_ENDPOINT="http://${NODE_IP}:22770/global-scheduler/traefik/config" -log_info "Using FunctionMaster route provider: ${TRAEFIK_PROVIDER_ENDPOINT}" -start_traefik_container "${TRAEFIK_PROVIDER_ENDPOINT}" +wait_for_data_plane +start_traefik_container TRAEFIK_IP="$(container_ip "${TRAEFIK_CONTAINER_NAME}")" if [[ -z "${TRAEFIK_IP}" ]]; then log_error "Could not determine the Traefik container IP" diff --git a/deploy/standalone/stop.sh b/deploy/standalone/stop.sh index eb19800..685621b 100755 --- a/deploy/standalone/stop.sh +++ b/deploy/standalone/stop.sh @@ -7,7 +7,10 @@ set -e -CONTAINER_NAMES=("akernel-traefik" "akernel-node") +CONTAINER_NAMES=( + "${AKERNEL_TRAEFIK_CONTAINER_NAME:-akernel-traefik}" + "${AKERNEL_NODE_CONTAINER_NAME:-akernel-node}" +) # Container runtime command (docker or pouch) DOCKER_CMD="" diff --git a/sdk/python/akernel_sdk/_backends/openyuanrong_sandbox.py b/sdk/python/akernel_sdk/_backends/openyuanrong_sandbox.py index 8f577d8..73299b2 100644 --- a/sdk/python/akernel_sdk/_backends/openyuanrong_sandbox.py +++ b/sdk/python/akernel_sdk/_backends/openyuanrong_sandbox.py @@ -17,7 +17,10 @@ from __future__ import annotations import inspect +import math import os +import time +import uuid from collections.abc import Mapping from typing import Any @@ -153,6 +156,130 @@ class _CommandsDriver: def __init__(self, commands: Any) -> None: self._commands = commands self._handles: dict[int, Any] = {} + self._command_ids: dict[int, str] = {} + self._stable_commands: bool | None = None + + def _stable_client(self) -> tuple[Any, str] | None: + client = getattr(self._commands, "_client", None) + sandbox_id = getattr(self._commands, "_sid", None) + if client is None or not isinstance(sandbox_id, str) or not sandbox_id: + return None + return client, sandbox_id + + def _uses_stable_commands(self) -> bool: + if self._stable_commands is not None: + return self._stable_commands + stable_client = self._stable_client() + if stable_client is None: + self._stable_commands = False + return False + client, sandbox_id = stable_client + try: + response = client.invoke(sandbox_id, "process.capabilities", {}) + except Exception: + # Do not cache transient transport failures. The native client can + # still serve older RRTs, and the next operation may retry discovery. + return False + capabilities = response.get("capabilities") + self._stable_commands = isinstance(capabilities, list) and ( + "stable-command-id" in capabilities + ) + return self._stable_commands + + def _invoke_stable( + self, + action: str, + values: Mapping[str, Any], + *, + timeout: int | None = None, + ) -> Mapping[str, Any]: + stable_client = self._stable_client() + if stable_client is None: + raise RuntimeError("stable command transport is unavailable") + client, sandbox_id = stable_client + if timeout is None: + response = client.invoke(sandbox_id, action, dict(values)) + else: + response = client.invoke( + sandbox_id, + action, + dict(values), + timeout=timeout, + ) + if not isinstance(response, Mapping): + raise RuntimeError(f"{action} returned an invalid response") + return response + + @staticmethod + def _stable_result(value: Mapping[str, Any]) -> CommandResult: + exit_code = value.get("exit_code") + return CommandResult( + stdout=str(value.get("stdout") or ""), + stderr=str(value.get("stderr") or value.get("error") or ""), + exit_code=int(exit_code) if isinstance(exit_code, int) else -1, + ) + + def _wait_stable(self, pid: int, timeout: int | None) -> CommandResult: + command_id = self._command_ids.get(pid) + identity: dict[str, Any] = {"pid": pid} + if command_id is not None: + identity["command_id"] = command_id + deadline = None if timeout is None else time.monotonic() + timeout + while True: + if deadline is None: + wait_timeout = 10.0 + else: + remaining = deadline - time.monotonic() + if remaining <= 0: + self._invoke_stable("process.kill", identity) + return CommandResult( + stdout="", + stderr=f"Command timed out after {timeout} seconds", + exit_code=-1, + ) + wait_timeout = min(10.0, remaining) + response = self._invoke_stable( + "process.poll", + {**identity, "wait_timeout": wait_timeout}, + timeout=max(1, math.ceil(wait_timeout) + 1), + ) + status = str(response.get("status") or "").upper() + if status in {"PENDING", "RUNNING"}: + continue + if status in {"SUCCEEDED", "FAILED", "TIMED_OUT", "KILLED"}: + return self._stable_result(response) + raise RuntimeError( + str(response.get("error") or f"unknown command status {status!r}") + ) + + def _start_stable( + self, + cmd: str, + *, + envs: Mapping[str, str] | None, + cwd: str | None, + stdin: bool, + timeout: int | None = None, + ) -> int: + command_id = f"cmd-{uuid.uuid4().hex}" + values: dict[str, Any] = { + "command_id": command_id, + "cmd": cmd, + "envs": dict(envs) if envs is not None else None, + "cwd": cwd, + "want_stdin": stdin, + } + if timeout is not None: + values["timeout"] = timeout + response = self._invoke_stable("process.start", values) + error = response.get("error") + if error: + raise RuntimeError(str(error)) + pid = response.get("pid") + if not isinstance(pid, int) or pid <= 0: + raise RuntimeError(f"process.start returned invalid pid {pid!r}") + self._command_ids[pid] = command_id + return pid def run( self, @@ -163,6 +290,15 @@ def run( timeout: int, ) -> CommandResult: try: + if self._uses_stable_commands(): + pid = self._start_stable( + cmd, + envs=envs, + cwd=cwd, + stdin=False, + timeout=timeout, + ) + return self._wait_stable(pid, timeout) value = self._commands.run( cmd, envs=dict(envs) if envs is not None else None, @@ -182,6 +318,13 @@ def start( stdin: bool, ) -> int: try: + if self._uses_stable_commands(): + return self._start_stable( + cmd, + envs=envs, + cwd=cwd, + stdin=stdin, + ) handle = self._commands.run( cmd, background=True, @@ -196,6 +339,11 @@ def start( return pid def wait(self, pid: int, timeout: int | None) -> CommandResult: + if self._uses_stable_commands(): + try: + return self._wait_stable(pid, timeout) + except Exception as error: + raise _convert_error(f"wait for process {pid}", error) from error handle = self._handles.get(pid) if handle is None: raise BackendOperationError(f"no command handle for pid {pid}") @@ -206,18 +354,50 @@ def wait(self, pid: int, timeout: int | None) -> CommandResult: def kill(self, pid: int) -> bool: try: + if self._uses_stable_commands(): + values: dict[str, Any] = {"pid": pid} + if command_id := self._command_ids.get(pid): + values["command_id"] = command_id + return bool(self._invoke_stable("process.kill", values).get("killed")) return bool(self._commands.kill(pid)) except Exception as error: raise _convert_error(f"kill process {pid}", error) from error def send_stdin(self, pid: int, data: str, eof: bool) -> None: try: + if self._uses_stable_commands(): + values: dict[str, Any] = { + "pid": pid, + "data": data, + "eof": eof, + } + if command_id := self._command_ids.get(pid): + values["command_id"] = command_id + response = self._invoke_stable("process.send_stdin", values) + if error := response.get("error"): + raise RuntimeError(str(error)) + return self._commands.send_stdin(pid, data, eof) except Exception as error: raise _convert_error(f"send stdin to process {pid}", error) from error def list(self) -> list[CommandInfo]: try: + if self._uses_stable_commands(): + values = self._invoke_stable("process.list", {}) + processes = values.get("processes") + if not isinstance(processes, list): + return [] + return [ + CommandInfo( + pid=int(value.get("pid") or 0), + command=str(value.get("cmd") or ""), + running=str(value.get("status") or "").upper() + in {"PENDING", "RUNNING"}, + ) + for value in processes + if isinstance(value, Mapping) + ] return [_command_info(value) for value in self._commands.list()] except Exception as error: raise _convert_error("list processes", error) from error diff --git a/sdk/python/benchmarks/__init__.py b/sdk/python/benchmarks/__init__.py new file mode 100644 index 0000000..e3eff61 --- /dev/null +++ b/sdk/python/benchmarks/__init__.py @@ -0,0 +1 @@ +"""Manual AKernel SDK benchmark programs.""" diff --git a/sdk/python/benchmarks/cluster_throughput.py b/sdk/python/benchmarks/cluster_throughput.py new file mode 100644 index 0000000..e191948 --- /dev/null +++ b/sdk/python/benchmarks/cluster_throughput.py @@ -0,0 +1,517 @@ +# Copyright (c) 2026 Ant Group Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Measure AKernel sandbox create, direct-invoke, and command throughput. + +The modes deliberately isolate different paths: + +* ``create`` measures only ``Sandbox(...)`` readiness. Sandboxes stay alive + until the measured burst completes, and deletion is outside the timer. +* ``invoke`` uses a pre-created pool and calls ``files.exists("/")``. +* ``exec`` uses a pre-created pool and calls ``commands.run("/bin/true")``. + +The program emits one JSON document. Tokens are read only by the SDK from the +environment and are never included in the result. +""" + +from __future__ import annotations + +import argparse +import json +import math +import os +import platform +import socket +import sys +import threading +import time +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor, as_completed +from dataclasses import asdict, dataclass, field +from pathlib import Path +from typing import Any, Protocol, TypeVar + +from akernel_sdk.sandbox import Sandbox + + +class SandboxLike(Protocol): + """Minimal sandbox contract required by this benchmark.""" + + @property + def id(self) -> str: ... + + @property + def files(self) -> Any: ... + + @property + def commands(self) -> Any: ... + + def kill(self) -> None: ... + + +@dataclass(frozen=True) +class SandboxConfig: + """Resource settings shared by every measured sandbox.""" + + runtime: str = "runsc" + cpu: int = 100 + memory: int = 128 + cpu_limit: int = 500 + mem_limit: int = 512 + idle_timeout: int = 300 + schedule_timeout: int = 60 + + def kwargs(self) -> dict[str, Any]: + return asdict(self) + + +@dataclass +class Sample: + """One operation result.""" + + latency_seconds: float + value: Any = None + error: str | None = None + + +@dataclass +class BenchmarkResult: + """Serializable result for one benchmark mode.""" + + mode: str + operations: int + concurrency: int + success: int + failed: int + wall_seconds: float + throughput_ops_s: float + latency_ms: dict[str, float] + errors: list[str] = field(default_factory=list) + cleanup_failed: int = 0 + cleanup_seconds: float = 0.0 + sandbox_count: int = 0 + + +T = TypeVar("T") +_KEPT_SANDBOXES: list[SandboxLike] = [] + + +def _percentile(sorted_values: list[float], percentile: float) -> float: + """Return a nearest-rank percentile from a sorted, non-empty sample.""" + + index = max(0, math.ceil(percentile * len(sorted_values)) - 1) + return sorted_values[index] + + +def latency_summary(samples_seconds: list[float]) -> dict[str, float]: + """Summarize successful operation latency in milliseconds.""" + + if not samples_seconds: + return {} + values = sorted(value * 1000 for value in samples_seconds) + return { + "min": values[0], + "p50": _percentile(values, 0.50), + "p90": _percentile(values, 0.90), + "p95": _percentile(values, 0.95), + "p99": _percentile(values, 0.99), + "max": values[-1], + "mean": sum(values) / len(values), + } + + +def _error_text(error: BaseException) -> str: + """Return a bounded, single-line error suitable for benchmark output.""" + + return " ".join(f"{type(error).__name__}: {error}".split())[:500] + + +def run_parallel( + operation: Callable[[int], T], operations: int, concurrency: int +) -> tuple[list[Sample], float]: + """Run a fixed number of operations with bounded closed-loop concurrency.""" + + if operations < 1: + raise ValueError("operations must be at least 1") + if concurrency < 1: + raise ValueError("concurrency must be at least 1") + + samples: list[Sample] = [] + + def measured(index: int) -> Sample: + started = time.perf_counter() + try: + value = operation(index) + return Sample(time.perf_counter() - started, value=value) + except Exception as error: # Each failed request is benchmark data. + return Sample( + time.perf_counter() - started, + error=_error_text(error), + ) + + wall_started = time.perf_counter() + with ThreadPoolExecutor(max_workers=concurrency) as executor: + futures = [executor.submit(measured, index) for index in range(operations)] + for future in as_completed(futures): + samples.append(future.result()) + return samples, time.perf_counter() - wall_started + + +def _result( + mode: str, + operations: int, + concurrency: int, + samples: list[Sample], + wall_seconds: float, + *, + cleanup_failed: int = 0, + cleanup_seconds: float = 0.0, + sandbox_count: int = 0, +) -> BenchmarkResult: + successful = [sample for sample in samples if sample.error is None] + errors = [sample.error for sample in samples if sample.error is not None] + return BenchmarkResult( + mode=mode, + operations=operations, + concurrency=concurrency, + success=len(successful), + failed=len(errors), + wall_seconds=wall_seconds, + throughput_ops_s=len(successful) / wall_seconds if wall_seconds else 0.0, + latency_ms=latency_summary( + [sample.latency_seconds for sample in successful] + ), + errors=errors[:5], + cleanup_failed=cleanup_failed, + cleanup_seconds=cleanup_seconds, + sandbox_count=sandbox_count, + ) + + +def cleanup_sandboxes( + sandboxes: list[SandboxLike], concurrency: int +) -> int: + """Best-effort parallel sandbox cleanup; return the failure count.""" + + if not sandboxes: + return 0 + + def kill(sandbox: SandboxLike) -> None: + sandbox.kill() + + with ThreadPoolExecutor(max_workers=max(1, concurrency)) as executor: + futures = [executor.submit(kill, sandbox) for sandbox in sandboxes] + return sum(1 for future in futures if future.exception() is not None) + + +def benchmark_create( + factory: Callable[[], SandboxLike], + operations: int, + concurrency: int, + start_at_epoch: float = 0, + pre_cleanup_delay: float = 0, +) -> BenchmarkResult: + """Measure a create burst while holding successful sandboxes alive.""" + + wait_for_synchronized_start(start_at_epoch) + samples, wall_seconds = run_parallel( + lambda _index: factory(), operations, concurrency + ) + sandboxes = [ + sample.value for sample in samples if sample.error is None + ] + if pre_cleanup_delay > 0: + time.sleep(pre_cleanup_delay) + cleanup_started = time.perf_counter() + cleanup_failed = cleanup_sandboxes(sandboxes, concurrency) + cleanup_seconds = time.perf_counter() - cleanup_started + return _result( + "create", + operations, + concurrency, + samples, + wall_seconds, + cleanup_failed=cleanup_failed, + cleanup_seconds=cleanup_seconds, + sandbox_count=len(sandboxes), + ) + + +def wait_for_synchronized_start(start_at_epoch: float) -> None: + """Wait for a shared epoch, rejecting workers that missed it badly.""" + + if start_at_epoch <= 0: + return + start_delay = start_at_epoch - time.time() + if start_delay < -1: + raise RuntimeError( + f"missed synchronized start by {-start_delay:.3f} seconds" + ) + if start_delay > 0: + time.sleep(start_delay) + + +def create_pool( + factory: Callable[[], SandboxLike], sandbox_count: int, concurrency: int +) -> list[SandboxLike]: + """Create the unmeasured sandbox pool used by invoke and exec modes.""" + + samples, _ = run_parallel(lambda _index: factory(), sandbox_count, concurrency) + sandboxes = [ + sample.value for sample in samples if sample.error is None + ] + errors = [sample.error for sample in samples if sample.error is not None] + if errors: + cleanup_sandboxes(sandboxes, concurrency) + raise RuntimeError(f"failed to create sandbox pool: {errors[0]}") + return sandboxes + + +def benchmark_pool( + mode: str, + factory: Callable[[], SandboxLike], + operations: int, + concurrency: int, + sandbox_count: int, + command_timeout: int, + pool_create_concurrency: int = 0, + start_at_epoch: float = 0, + keep_sandboxes: bool = False, +) -> BenchmarkResult: + """Measure direct invoke or command execution against a pre-created pool.""" + + if mode not in {"invoke", "exec"}: + raise ValueError(f"unsupported pool mode: {mode}") + if sandbox_count < concurrency: + raise ValueError("sandbox-count must be greater than or equal to concurrency") + if pool_create_concurrency < 0: + raise ValueError("pool-create-concurrency cannot be negative") + + create_concurrency = pool_create_concurrency or concurrency + sandboxes = create_pool(factory, sandbox_count, create_concurrency) + locks = [threading.Lock() for _ in sandboxes] + + def invoke(index: int) -> None: + pool_index = index % len(sandboxes) + sandbox = sandboxes[pool_index] + with locks[pool_index]: + if mode == "invoke": + if not sandbox.files.exists("/"): + raise RuntimeError("sandbox root does not exist") + return + result = sandbox.commands.run("/bin/true", timeout=command_timeout) + if result.exit_code != 0: + raise RuntimeError(f"/bin/true exited with {result.exit_code}") + + try: + # Exclude first-use connection setup from the measured interval. + for index, sandbox in enumerate(sandboxes): + if mode == "invoke": + if not sandbox.files.exists("/"): + raise RuntimeError(f"sandbox {index} root does not exist") + else: + result = sandbox.commands.run("/bin/true", timeout=command_timeout) + if result.exit_code != 0: + raise RuntimeError( + f"sandbox {index} warm-up exited with {result.exit_code}" + ) + wait_for_synchronized_start(start_at_epoch) + samples, wall_seconds = run_parallel(invoke, operations, concurrency) + finally: + if keep_sandboxes: + # Retain strong references until main prints the result and exits + # without running SDK destructors. Cleanup is coordinated outside + # this process so concurrent deletion cannot hide measurements. + _KEPT_SANDBOXES.extend(sandboxes) + cleanup_failed = 0 + cleanup_seconds = 0.0 + else: + cleanup_started = time.perf_counter() + cleanup_failed = cleanup_sandboxes(sandboxes, concurrency) + cleanup_seconds = time.perf_counter() - cleanup_started + + return _result( + mode, + operations, + concurrency, + samples, + wall_seconds, + cleanup_failed=cleanup_failed, + cleanup_seconds=cleanup_seconds, + sandbox_count=sandbox_count, + ) + + +def parse_args(argv: list[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--mode", + choices=("create", "invoke", "exec", "all"), + default="all", + ) + parser.add_argument("--operations", type=int, default=20) + parser.add_argument("--concurrency", type=int, default=4) + parser.add_argument("--sandbox-count", type=int, default=4) + parser.add_argument( + "--pool-create-concurrency", + type=int, + default=0, + help="pool setup concurrency; 0 uses --concurrency", + ) + parser.add_argument( + "--start-at-epoch", + type=float, + default=0, + help="wait after pool warm-up until this Unix epoch timestamp", + ) + parser.add_argument( + "--keep-sandboxes", + action="store_true", + help="leave the pool running for externally coordinated cleanup", + ) + parser.add_argument("--runtime", default="runsc") + parser.add_argument("--cpu", type=int, default=100) + parser.add_argument("--memory", type=int, default=128) + parser.add_argument("--cpu-limit", type=int, default=500) + parser.add_argument("--mem-limit", type=int, default=512) + parser.add_argument("--idle-timeout", type=int, default=300) + parser.add_argument("--schedule-timeout", type=int, default=60) + parser.add_argument( + "--node-id", + action="append", + default=[], + help="pin creates to this node; repeat to distribute round-robin", + ) + parser.add_argument( + "--node-label", + default="NODE_ID", + help="scheduler label used for --node-id (external collectors use HOST_IP)", + ) + parser.add_argument("--command-timeout", type=int, default=20) + parser.add_argument( + "--pre-cleanup-delay", + type=float, + default=0, + help="wait after measured create completion before deleting sandboxes", + ) + parser.add_argument("--output", type=Path) + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv) + if not os.environ.get("AKERNEL_SERVER_ADDRESS"): + raise SystemExit("AKERNEL_SERVER_ADDRESS is required") + if not os.environ.get("AKERNEL_TOKEN"): + raise SystemExit("AKERNEL_TOKEN is required") + + config = SandboxConfig( + runtime=args.runtime, + cpu=args.cpu, + memory=args.memory, + cpu_limit=args.cpu_limit, + mem_limit=args.mem_limit, + idle_timeout=args.idle_timeout, + schedule_timeout=args.schedule_timeout, + ) + node_ids = tuple(dict.fromkeys(node_id.strip() for node_id in args.node_id)) + if any(not node_id for node_id in node_ids): + raise SystemExit("--node-id must be non-empty") + node_label = args.node_label.strip() + if not node_label: + raise SystemExit("--node-label must be non-empty") + if node_ids and node_label != "NODE_ID": + if os.environ.get("AKERNEL_BACKEND") != "openyuanrong-sandbox": + raise SystemExit( + "custom --node-label requires AKERNEL_BACKEND=openyuanrong-sandbox" + ) + import yr_sandbox.sandbox_api as sandbox_api + + sandbox_api._NODE_ID_LABEL = node_label + factory_lock = threading.Lock() + next_node = 0 + + def factory() -> SandboxLike: + nonlocal next_node + node_id = None + if node_ids: + with factory_lock: + node_id = node_ids[next_node % len(node_ids)] + next_node += 1 + return Sandbox(**config.kwargs(), node_id=node_id) + + modes = ("create", "invoke", "exec") if args.mode == "all" else (args.mode,) + results: list[BenchmarkResult] = [] + + for mode in modes: + if mode == "create": + result = benchmark_create( + factory, + args.operations, + args.concurrency, + args.start_at_epoch, + args.pre_cleanup_delay, + ) + else: + result = benchmark_pool( + mode, + factory, + args.operations, + args.concurrency, + args.sandbox_count, + args.command_timeout, + args.pool_create_concurrency, + args.start_at_epoch, + args.keep_sandboxes, + ) + results.append(result) + + payload = { + "schema_version": 1, + "run_id": f"{int(time.time())}-{socket.gethostname()}", + "client": { + "hostname": socket.gethostname(), + "python": platform.python_version(), + "platform": platform.platform(), + }, + "sandbox": asdict(config), + "runner": { + "pool_create_concurrency": args.pool_create_concurrency, + "start_at_epoch": args.start_at_epoch, + "keep_sandboxes": args.keep_sandboxes, + "node_ids": list(node_ids), + "node_label": node_label, + "pre_cleanup_delay": args.pre_cleanup_delay, + }, + "results": [asdict(result) for result in results], + } + rendered = json.dumps(payload, indent=2, sort_keys=True) + print(rendered) + if args.output is not None: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(rendered + "\n", encoding="utf-8") + + failed = any( + result.failed or result.cleanup_failed for result in results + ) + exit_code = 1 if failed else 0 + if args.keep_sandboxes: + sys.stdout.flush() + sys.stderr.flush() + os._exit(exit_code) + return exit_code + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/sdk/python/benchmarks/edge_raw_throughput.py b/sdk/python/benchmarks/edge_raw_throughput.py new file mode 100644 index 0000000..1f3a2e9 --- /dev/null +++ b/sdk/python/benchmarks/edge_raw_throughput.py @@ -0,0 +1,218 @@ +#!/usr/bin/env python3 + +# Copyright (c) 2026 Ant Group Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. + +"""Benchmark Edge /direct using raw persistent TLS connections. + +This diagnostic intentionally bypasses HTTPX and the high-level SDK while +retaining TLS, JWT authentication, Edge routing, Node Proxy, and RRT. +""" + +from __future__ import annotations + +import argparse +import json +import math +import os +import socket +import ssl +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from typing import BinaryIO + +_ACTIONS = { + "file.exists": ( + {"action": "file.exists", "args": {"path": "/"}}, + ("exists", True), + ), + "ping": ({"action": "ping", "args": {}}, ("status", "ok")), +} + + +def _percentile(values: list[float], percentile: float) -> float: + index = max(0, math.ceil(percentile * len(values)) - 1) + return sorted(values)[index] + + +def _read_exact(reader: BinaryIO, size: int) -> bytes: + chunks: list[bytes] = [] + remaining = size + while remaining: + chunk = reader.read(remaining) + if not chunk: + raise EOFError(f"response ended with {remaining} bytes remaining") + chunks.append(chunk) + remaining -= len(chunk) + return b"".join(chunks) + + +def _read_response(reader: BinaryIO, expected_status: int) -> bytes: + status = reader.readline() + expected_prefix = f"HTTP/1.1 {expected_status} ".encode() + if not status.startswith(expected_prefix): + raise RuntimeError(f"unexpected response status: {status!r}") + content_length: int | None = None + while True: + line = reader.readline() + if line in (b"\r\n", b""): + break + name, _, value = line.partition(b":") + if name.lower() == b"content-length": + content_length = int(value.strip()) + if content_length is None: + raise RuntimeError("response is missing Content-Length") + return _read_exact(reader, content_length) + + +def benchmark_instance( + instance_id: str, + connection_index: int, + operations: int, + host: str, + port: int, + token: str, + context: ssl.SSLContext, + barrier: threading.Barrier, + missing_route: bool, + edge_only: bool, + body: bytes, + expected_result: tuple[str, object], + edge_generates_request_id: bool, +) -> list[float]: + """Run one keep-alive TLS connection for a sandbox instance.""" + + target_id = f"missing-{instance_id}" if missing_route else instance_id + path = ( + "/__akernel_benchmark_not_found" + if edge_only + else f"/direct/{target_id}/invoke" + ) + expected_status = 404 if edge_only else (503 if missing_route else 200) + request_id_header = ( + b"" + if edge_generates_request_id + else f"X-Request-ID: raw-{os.getpid()}-{connection_index}\r\n".encode() + ) + request = ( + f"POST {path} HTTP/1.1\r\n".encode() + + f"Host: {host}:{port}\r\n".encode() + + b"Content-Type: application/json\r\n" + + f"X-Auth: {token}\r\n".encode() + + request_id_header + + f"Content-Length: {len(body)}\r\n".encode() + + b"Connection: keep-alive\r\n\r\n" + + body + ) + latencies: list[float] = [] + with socket.create_connection((host, port), timeout=10) as tcp: + tcp.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + with context.wrap_socket(tcp, server_hostname=host) as connection: + with connection.makefile("rb") as reader: + connection.sendall(request) + warmup_body = _read_response(reader, expected_status) + if not missing_route and not edge_only and json.loads( + warmup_body + ).get(expected_result[0]) != expected_result[1]: + raise RuntimeError( + f"unexpected warm-up response: {warmup_body!r}" + ) + barrier.wait() + for _ in range(operations): + started = time.perf_counter() + connection.sendall(request) + response_body = _read_response(reader, expected_status) + if not missing_route and not edge_only and json.loads( + response_body + ).get(expected_result[0]) != expected_result[1]: + raise RuntimeError( + f"unexpected response for {instance_id}: {response_body!r}" + ) + latencies.append(time.perf_counter() - started) + return latencies + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--host", required=True) + parser.add_argument("--port", type=int, default=443) + parser.add_argument("--operations-per-sandbox", type=int, default=5_000) + parser.add_argument("--missing-route", action="store_true") + parser.add_argument("--edge-only", action="store_true") + parser.add_argument("--action", choices=sorted(_ACTIONS), default="file.exists") + parser.add_argument( + "--edge-generates-request-id", + action="store_true", + help="Omit X-Request-ID to measure Edge-side UUID generation.", + ) + parser.add_argument("instances", nargs="+") + args = parser.parse_args() + if args.operations_per_sandbox <= 0: + parser.error("--operations-per-sandbox must be positive") + token = os.environ.get("AKERNEL_TOKEN", "").strip() + if not token: + raise SystemExit("AKERNEL_TOKEN is required") + action, expected_result = _ACTIONS[args.action] + body = json.dumps(action, separators=(",", ":")).encode() + + context = ssl.create_default_context() + context.check_hostname = False + context.verify_mode = ssl.CERT_NONE + barrier = threading.Barrier(len(args.instances) + 1) + with ThreadPoolExecutor(max_workers=len(args.instances)) as executor: + futures = [ + executor.submit( + benchmark_instance, + instance_id, + connection_index, + args.operations_per_sandbox, + args.host, + args.port, + token, + context, + barrier, + args.missing_route, + args.edge_only, + body, + expected_result, + args.edge_generates_request_id, + ) + for connection_index, instance_id in enumerate(args.instances) + ] + barrier.wait() + started = time.perf_counter() + samples = [future.result() for future in futures] + wall_seconds = time.perf_counter() - started + latencies_ms = [sample * 1000 for values in samples for sample in values] + operations = len(latencies_ms) + print( + json.dumps( + { + "instances": len(args.instances), + "action": args.action, + "request_id_source": ( + "edge" if args.edge_generates_request_id else "raw-client" + ), + "operations": operations, + "wall_seconds": wall_seconds, + "throughput_ops_s": operations / wall_seconds, + "latency_ms": { + "mean": sum(latencies_ms) / operations, + "p50": _percentile(latencies_ms, 0.50), + "p95": _percentile(latencies_ms, 0.95), + "p99": _percentile(latencies_ms, 0.99), + "max": max(latencies_ms), + }, + }, + indent=2, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/sdk/python/benchmarks/rrt_direct_throughput.py b/sdk/python/benchmarks/rrt_direct_throughput.py new file mode 100644 index 0000000..8e50b3b --- /dev/null +++ b/sdk/python/benchmarks/rrt_direct_throughput.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python3 + +# Copyright (c) 2026 Ant Group Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 + +"""Benchmark local sandbox RRT endpoints without Edge or Node Proxy. + +Run this inside an AKernel node container. Active sandbox bridge addresses are +discovered from sandboxd OCI metadata and each endpoint receives one persistent +HTTP/1.1 connection, matching the Edge backend pool's steady-state shape. +""" + +from __future__ import annotations + +import argparse +import json +import math +import socket +import time +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +from typing import BinaryIO + +_CONFIG_GLOB = "/home/akernel/sandboxd/root/containers/sbox-*/config.json" +_INTERFACE_ANNOTATION = "sandbox.akernel.dev/resource-interface" +_CGROUP_ANNOTATION = "sandbox.akernel.dev/resource-cgroup" +_CPU_CGROUP_ROOT = Path("/sys/fs/cgroup/cpu,cpuacct") +_BODY = json.dumps( + {"action": "file.exists", "args": {"path": "/"}}, + separators=(",", ":"), +).encode() + + +def _percentile(values: list[float], percentile: float) -> float: + index = max(0, math.ceil(percentile * len(values)) - 1) + return sorted(values)[index] + + +def _read_exact(reader: BinaryIO, size: int) -> bytes: + chunks: list[bytes] = [] + remaining = size + while remaining: + chunk = reader.read(remaining) + if not chunk: + raise EOFError(f"response ended with {remaining} bytes remaining") + chunks.append(chunk) + remaining -= len(chunk) + return b"".join(chunks) + + +def _read_response(reader: BinaryIO) -> bytes: + status = reader.readline() + if not status.startswith(b"HTTP/1.1 200 "): + raise RuntimeError(f"unexpected response status: {status!r}") + content_length: int | None = None + while True: + line = reader.readline() + if line in (b"\r\n", b""): + break + name, _, value = line.partition(b":") + if name.lower() == b"content-length": + content_length = int(value.strip()) + if content_length is None: + raise RuntimeError("response is missing Content-Length") + return _read_exact(reader, content_length) + + +def discover_active_endpoints() -> list[str]: + """Return bridge IPs for sandboxd containers with a populated CPU cgroup.""" + + endpoints: list[str] = [] + for config_path in sorted(Path("/").glob(_CONFIG_GLOB.lstrip("/"))): + config = json.loads(config_path.read_text(encoding="utf-8")) + annotations = config.get("annotations") or {} + cgroup = str(annotations.get(_CGROUP_ANNOTATION) or "") + cgroup_procs = _CPU_CGROUP_ROOT / cgroup.lstrip("/") / "cgroup.procs" + try: + active = bool(cgroup_procs.read_text(encoding="utf-8").strip()) + except FileNotFoundError: + continue + if not active: + continue + interface = json.loads(annotations[_INTERFACE_ANNOTATION]) + endpoints.append(str(interface["ip"])) + return endpoints + + +def benchmark_endpoint(ip: str, operations: int, port: int) -> list[float]: + """Run sequential keep-alive requests against one RRT endpoint.""" + + request = ( + b"POST /invoke HTTP/1.1\r\n" + + f"Host: {ip}:{port}\r\n".encode() + + b"Content-Type: application/json\r\n" + + f"Content-Length: {len(_BODY)}\r\n".encode() + + b"Connection: keep-alive\r\n\r\n" + + _BODY + ) + latencies: list[float] = [] + with socket.create_connection((ip, port), timeout=10) as connection: + connection.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) + with connection.makefile("rb") as reader: + for _ in range(operations): + started = time.perf_counter() + connection.sendall(request) + response = json.loads(_read_response(reader)) + if response.get("exists") is not True: + raise RuntimeError(f"unexpected response from {ip}: {response!r}") + latencies.append(time.perf_counter() - started) + return latencies + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--operations-per-sandbox", type=int, default=5_000) + parser.add_argument("--port", type=int, default=50_090) + args = parser.parse_args() + if args.operations_per_sandbox <= 0: + parser.error("--operations-per-sandbox must be positive") + + endpoints = discover_active_endpoints() + if not endpoints: + raise SystemExit("no active local sandbox endpoints found") + + started = time.perf_counter() + with ThreadPoolExecutor(max_workers=len(endpoints)) as executor: + samples = list( + executor.map( + lambda ip: benchmark_endpoint( + ip, args.operations_per_sandbox, args.port + ), + endpoints, + ) + ) + wall_seconds = time.perf_counter() - started + latencies_ms = [sample * 1000 for values in samples for sample in values] + operations = len(latencies_ms) + print( + json.dumps( + { + "endpoints": len(endpoints), + "operations": operations, + "wall_seconds": wall_seconds, + "throughput_ops_s": operations / wall_seconds, + "latency_ms": { + "mean": sum(latencies_ms) / operations, + "p50": _percentile(latencies_ms, 0.50), + "p95": _percentile(latencies_ms, 0.95), + "p99": _percentile(latencies_ms, 0.99), + "max": max(latencies_ms), + }, + }, + indent=2, + sort_keys=True, + ) + ) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/sdk/python/tests/unit/test_backends.py b/sdk/python/tests/unit/test_backends.py index 425a55d..acc226c 100644 --- a/sdk/python/tests/unit/test_backends.py +++ b/sdk/python/tests/unit/test_backends.py @@ -151,6 +151,69 @@ def test_connection_config_maps_to_yr_environment(self): self.assertEqual(os.environ["YR_GATEWAY_TLS"], "0") self.assertEqual(os.environ["YR_TOKEN"], "secret") + def test_stable_command_protocol_adds_id_and_polls_result(self): + client = MagicMock() + + def invoke(_sandbox_id, action, values, **_kwargs): + if action == "process.capabilities": + return {"capabilities": ["stable-command-id"]} + if action == "process.start": + self.assertRegex(values["command_id"], r"^cmd-[0-9a-f]{32}$") + return {"pid": 17, "status": "RUNNING"} + if action == "process.poll": + self.assertEqual(values["pid"], 17) + self.assertRegex(values["command_id"], r"^cmd-[0-9a-f]{32}$") + return { + "pid": 17, + "status": "SUCCEEDED", + "stdout": "ok\n", + "stderr": "", + "exit_code": 0, + } + self.fail(f"unexpected action {action}") + + client.invoke.side_effect = invoke + commands = SimpleNamespace(_client=client, _sid="default-worker") + driver = openyuanrong_sandbox._CommandsDriver(commands) + + self.assertEqual( + driver.run("echo ok", envs=None, cwd=None, timeout=60), + CommandResult("ok\n", "", 0), + ) + + def test_stable_background_command_uses_id_for_followup_actions(self): + client = MagicMock() + + def invoke(_sandbox_id, action, values, **_kwargs): + if action == "process.capabilities": + return {"capabilities": ["stable-command-id"]} + if action == "process.start": + return {"pid": 23, "status": "RUNNING"} + if action == "process.list": + return { + "processes": [{"pid": 23, "cmd": "sleep 30", "status": "RUNNING"}] + } + if action == "process.send_stdin": + self.assertRegex(values["command_id"], r"^cmd-[0-9a-f]{32}$") + return {"error": None} + if action == "process.kill": + self.assertRegex(values["command_id"], r"^cmd-[0-9a-f]{32}$") + return {"killed": True, "error": None} + self.fail(f"unexpected action {action}") + + client.invoke.side_effect = invoke + commands = SimpleNamespace(_client=client, _sid="default-worker") + driver = openyuanrong_sandbox._CommandsDriver(commands) + + pid = driver.start("sleep 30", envs=None, cwd=None, stdin=True) + self.assertEqual(pid, 23) + self.assertEqual( + driver.list(), + [CommandInfo(pid=23, command="sleep 30", running=True)], + ) + driver.send_stdin(pid, "input", False) + self.assertTrue(driver.kill(pid)) + def test_runtime_identifier_without_explicit_rootfs_is_forwarded(self): native = MagicMock() native.id = "default-gvisor-next" diff --git a/sdk/python/tests/unit/test_cluster_throughput_benchmark.py b/sdk/python/tests/unit/test_cluster_throughput_benchmark.py new file mode 100644 index 0000000..bda105f --- /dev/null +++ b/sdk/python/tests/unit/test_cluster_throughput_benchmark.py @@ -0,0 +1,250 @@ +# Copyright (c) 2026 Ant Group Corporation. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import sys +import time +import unittest +from dataclasses import dataclass +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from benchmarks import cluster_throughput as benchmark + + +@dataclass +class FakeResult: + exit_code: int = 0 + + +class FakeFiles: + def __init__(self): + self.calls = 0 + + def exists(self, path): + self.calls += 1 + return path == "/" + + +class FakeCommands: + def __init__(self): + self.calls = 0 + + def run(self, command, timeout=60): + self.calls += 1 + return FakeResult(0 if command == "/bin/true" else 1) + + +class FakeSandbox: + next_id = 0 + + def __init__(self): + FakeSandbox.next_id += 1 + self.id = f"sandbox-{FakeSandbox.next_id}" + self.files = FakeFiles() + self.commands = FakeCommands() + self.killed = False + + def kill(self): + self.killed = True + + +class ClusterThroughputBenchmarkTest(unittest.TestCase): + def test_latency_summary_uses_nearest_rank_percentiles(self): + summary = benchmark.latency_summary([0.001, 0.002, 0.003, 0.004]) + + self.assertEqual(summary["min"], 1.0) + self.assertEqual(summary["p50"], 2.0) + self.assertEqual(summary["p90"], 4.0) + self.assertEqual(summary["p99"], 4.0) + self.assertEqual(summary["max"], 4.0) + + def test_create_holds_sandboxes_until_measured_burst_finishes(self): + created = [] + + def factory(): + sandbox = FakeSandbox() + created.append(sandbox) + time.sleep(0.001) + return sandbox + + result = benchmark.benchmark_create(factory, operations=4, concurrency=2) + + self.assertEqual(result.success, 4) + self.assertEqual(result.failed, 0) + self.assertEqual(result.sandbox_count, 4) + self.assertGreaterEqual(result.cleanup_seconds, 0) + self.assertTrue(all(sandbox.killed for sandbox in created)) + + def test_create_waits_for_synchronized_start(self): + with ( + patch.object(benchmark.time, "time", return_value=97.5), + patch.object(benchmark.time, "sleep") as sleep, + ): + result = benchmark.benchmark_create( + FakeSandbox, + operations=2, + concurrency=1, + start_at_epoch=100, + ) + + sleep.assert_called_once_with(2.5) + self.assertEqual(result.success, 2) + + def test_create_can_wait_before_cleanup(self): + with patch.object(benchmark.time, "sleep") as sleep: + result = benchmark.benchmark_create( + FakeSandbox, + operations=2, + concurrency=1, + pre_cleanup_delay=2, + ) + + sleep.assert_called_once_with(2) + self.assertEqual(result.success, 2) + + def test_cli_accepts_repeated_node_ids(self): + args = benchmark.parse_args( + [ + "--mode=create", + "--node-id=192.0.2.10", + "--node-id=192.0.2.11", + "--node-label=HOST_IP", + ] + ) + + self.assertEqual(args.node_id, ["192.0.2.10", "192.0.2.11"]) + self.assertEqual(args.node_label, "HOST_IP") + + def test_invoke_uses_precreated_pool_and_excludes_warmup(self): + created = [] + + def factory(): + sandbox = FakeSandbox() + created.append(sandbox) + return sandbox + + result = benchmark.benchmark_pool( + "invoke", + factory, + operations=6, + concurrency=2, + sandbox_count=2, + command_timeout=5, + ) + + self.assertEqual(result.success, 6) + self.assertEqual(result.failed, 0) + self.assertEqual(sum(s.files.calls for s in created), 8) + self.assertTrue(all(sandbox.killed for sandbox in created)) + + def test_exec_uses_bin_true(self): + created = [] + + def factory(): + sandbox = FakeSandbox() + created.append(sandbox) + return sandbox + + result = benchmark.benchmark_pool( + "exec", + factory, + operations=4, + concurrency=2, + sandbox_count=2, + command_timeout=5, + ) + + self.assertEqual(result.success, 4) + self.assertEqual(sum(s.commands.calls for s in created), 6) + + def test_pool_rejects_less_sandboxes_than_concurrency(self): + with self.assertRaisesRegex(ValueError, "sandbox-count"): + benchmark.benchmark_pool( + "invoke", + FakeSandbox, + operations=2, + concurrency=2, + sandbox_count=1, + command_timeout=5, + ) + + def test_pool_creation_concurrency_is_independent_from_invoke(self): + created = [] + + def factory(): + sandbox = FakeSandbox() + created.append(sandbox) + return sandbox + + with patch.object( + benchmark, "create_pool", wraps=benchmark.create_pool + ) as create_pool: + result = benchmark.benchmark_pool( + "invoke", + factory, + operations=8, + concurrency=4, + sandbox_count=4, + command_timeout=5, + pool_create_concurrency=2, + ) + + create_pool.assert_called_once_with(factory, 4, 2) + self.assertEqual(result.success, 8) + self.assertTrue(all(sandbox.killed for sandbox in created)) + + def test_pool_waits_for_synchronized_start_after_warmup(self): + with ( + patch.object(benchmark.time, "time", return_value=97.5), + patch.object(benchmark.time, "sleep") as sleep, + ): + result = benchmark.benchmark_pool( + "invoke", + FakeSandbox, + operations=4, + concurrency=2, + sandbox_count=2, + command_timeout=5, + start_at_epoch=100, + ) + + sleep.assert_called_once_with(2.5) + self.assertEqual(result.success, 4) + + def test_pool_can_leave_sandboxes_for_external_cleanup(self): + created = [] + + def factory(): + sandbox = FakeSandbox() + created.append(sandbox) + return sandbox + + result = benchmark.benchmark_pool( + "invoke", + factory, + operations=4, + concurrency=2, + sandbox_count=2, + command_timeout=5, + keep_sandboxes=True, + ) + + self.assertEqual(result.success, 4) + self.assertTrue(all(not sandbox.killed for sandbox in created)) + benchmark._KEPT_SANDBOXES.clear() + +if __name__ == "__main__": + unittest.main() diff --git a/src/sandboxd b/src/sandboxd index 5970730..418ef5c 160000 --- a/src/sandboxd +++ b/src/sandboxd @@ -1 +1 @@ -Subproject commit 5970730acc01c25314d7db8322518f6ac4795d0a +Subproject commit 418ef5cc095c8deff23ddf3bdb423f78ec2b788d diff --git a/src/yuanrong b/src/yuanrong index 353e955..f0ea658 160000 --- a/src/yuanrong +++ b/src/yuanrong @@ -1 +1 @@ -Subproject commit 353e955ff6aea6c175a453e1b38c895269758f62 +Subproject commit f0ea65808fb54e499166d8e7341d9c94a6dc0144