当前位置：首页 > news >正文

Volga按需计算层：面向AI特征工程的实时函数计算平台

news 2026/7/19 9:28:20

1. 项目概述：为什么我们需要 Volga 的“按需计算”这一层？

你有没有遇到过这种场景：模型服务接口一调用，后端就开始疯狂查 Redis、调第三方 API、做向量相似度计算，整个链路卡在特征生成这一步，P99 延迟直接飙到 800ms？我去年在一家做实时推荐的公司做架构支持，就亲眼见过一个“用户兴趣向量实时融合”功能，因为所有特征都硬塞进 Flink 流水线里预计算，结果每次用户刷新首页，系统就得等 3 秒——不是模型推理慢，是特征根本没准备好。后来我们拆出来单独跑一个轻量服务，延迟立刻压到 120ms 以内。这件事让我彻底意识到：流式引擎再强，也解决不了“请求时刻才出现的数据”这个根本矛盾。

Volga 的 On-Demand Compute Layer（按需计算层）就是为这个矛盾而生的。它不试图把所有事情都塞进一个流里，而是坦然承认：有些数据天生就是“请求驱动”的——比如用户此刻的 GPS 坐标、当前会话的点击序列、临时上传的一张图片、或者需要调用另一个微服务才能拿到的风控结果。这些数据在事件发生前根本不存在，更不可能被提前写入 Kafka 或 Flink 的状态中。强行用流式处理去模拟，只会让架构变得臃肿、延迟不可控、运维成本翻倍。

Volga 的核心设计哲学很朴素：Push 和 Pull 不是对立，而是互补。Push 部分（Streaming Engine）负责“事件时间”下的确定性计算——用户点了什么、买了什么、页面停留多久，这些事一旦发生，就该立刻、稳定、可重放地算出来，存进共享存储；Pull 部分（On-Demand Layer）则专注“处理时间”下的灵活性——当一个请求打进来，它需要什么，就现场拉取、现场计算、现场返回。两者通过统一的存储抽象（Storage Abstraction）和统一的特征定义语言（Feature DSL）无缝衔接。这不是拼凑，而是把两种计算范式真正拧成一股绳。

你可能会问：那用 Flask/FastAPI 自己写个服务不就行了？确实可以，但 Volga 解决的是规模化落地时的“隐性成本”。比如，如何让一个新写的user_gps_distance_to_store特征，既能被线上模型服务毫秒级调用，又能被离线训练任务批量拉取？如何保证 50 个不同团队写的特征函数，不会因为某一个依赖了慢接口就把整个集群拖垮？如何让特征之间的依赖关系（A 依赖 B，B 又依赖 C 的最新值）自动拓扑排序、并行执行、错误隔离？这些都不是“写个 API”能搞定的，而是 Volga 在框架层就帮你兜底的事。它不是替代你的业务代码，而是让你的业务代码能在一个健壮、可观测、可伸缩的底盘上跑得更稳。

所以，Volga 的按需计算层，本质上是一个“面向 AI/ML 场景深度定制的、带状态管理与依赖调度能力的实时函数计算平台”。它不追求通用 Serverless 的极致弹性，而是聚焦在“特征计算”这个垂直切口上，把从请求解析、依赖拉取、DAG 执行、结果组装到错误熔断的整条链路，都打磨成开箱即用的工业级能力。如果你正在构建或维护一个实时推荐、搜索排序、风控决策、个性化广告投放系统，那么 Volga 的这一层，很可能就是你架构图里缺失的那块关键拼图。

2. 核心架构拆解：四个组件如何协同完成一次毫秒级特征计算？

Volga 的按需计算层不是单体服务，而是一个由四个核心角色紧密协作的分布式系统。理解它们各自的职责和交互方式，是掌握其威力的前提。我把它比作一家高效运转的“特征计算工厂”：有总调度室（Coordinator）、有流水线工人（Server）、有原料仓库（Storage）、还有外部接单窗口（Load Balancer）。下面我们就以一次真实的search_personalization_score请求为例，逐层拆解这个工厂是如何在 100ms 内完成全部工作的。

2.1 OnDemandCoordinator：不只是调度器，更是“特征-工人”的智能匹配官

OnDemandCoordinator是整个按需计算层的“大脑”和“人事经理”。它的存在，直接决定了系统能否应对业务的快速迭代。很多团队自己搭的特征服务，上线新特征就得手动改配置、重启服务，而 Volga 的 Coordinator 让这一切变成动态的。

它的核心工作有三项，每一项都直击痛点：

第一，逻辑隔离与资源分配。假设你有 20 个特征函数，其中 5 个是 CPU 密集型（比如做图像 embedding），10 个是 IO 密集型（频繁调第三方 API），另外 5 个是内存敏感型（需要大缓存）。Coordinator 不会把它们全塞进同一个 Worker 进程。它会根据你在注册特征时声明的resource_requirements（比如{"cpu": 2, "memory": 4096}），结合每个OnDemandServer实例启动时上报的资源画像，进行智能分组。一个只跑 CPU 型特征的 Worker，就不会被分配到任何需要调用外部 HTTP 接口的特征——这从根本上避免了“一个慢接口拖垮所有特征”的雪崩效应。我实测过，当一个特征因网络抖动超时，其他同 Worker 的特征完全不受影响，响应时间曲线依然平滑。

第二，弹性扩缩容的决策者。Coordinator 持续监听所有OnDemandServer的健康心跳和指标（CPU、内存、队列积压数）。当它发现某个节点上的平均请求排队时间超过 50ms，且持续 30 秒，它就会触发扩容：向 Ray 集群申请一个新的OnDemandServerActor，并将一部分负载较轻的特征（比如那些低频、低耗的）迁移过去。反之，当流量低谷期，它会主动下线空闲 Worker。这个过程对上游 Load Balancer 完全透明，无需任何 DNS 刷新或配置变更。我们在线上压测时，从 100 QPS 突然拉升到 5000 QPS，系统在 8 秒内就完成了从 4 个 Worker 到 16 个 Worker 的自动扩容，P99 延迟始终控制在 150ms 以内。

第三，特征注册与生命周期管家。当你执行coordinator.register_features.remote(...)时，Coordinator 干的远不止是存个函数指针。它会：

解析simple_feature函数的 AST，提取出所有@on_demand装饰器里的dependencies参数；
递归检查这些依赖是否已注册，构建出一张完整的“特征依赖图谱”；
为每个特征生成唯一的feature_id，并将其与物理 Worker 的映射关系写入一个轻量级的分布式状态存储（默认是 Ray 的NamedActor）；
启动一个后台任务，定期扫描所有已注册特征的代码哈希值，一旦检测到更新（比如你 push 了新版本的 feature repo），就触发热重载——整个过程无需重启任何进程。

提示：Coordinator 的所有操作都是异步且幂等的。这意味着你可以放心地在生产环境反复调用register_features，它只会更新变化的部分，不会造成服务中断。这是我们在灰度发布新特征时最依赖的机制。

2.2 OnDemandServer：一个 Starlette 进程，如何扛住千级并发？

OnDemandServer是真正的“一线工人”，它是一个 Python 进程，内部运行着一个 Starlette ASGI 应用。很多人看到这里会疑惑：Starlette 是 Web 框架，怎么和“计算”扯上关系？答案在于 Volga 对它的深度改造。

首先，它不是一个传统的“一个请求一个线程”的模型。每个OnDemandServer进程启动时，会创建一个asyncio事件循环，并在这个循环上注册多个协程任务。当一个 HTTP 请求（比如/v1/feature）到达时，Starlette 的路由层会根据target_features参数，快速定位到要执行的特征函数列表，然后启动一个“特征执行 DAG”。

这个 DAG 的执行是高度优化的：

并行拉取：如果simple_feature依赖test_feature和user_profile_feature两个 pipeline 特征，OnDemandServer会同时发起两个异步fetch_latest调用，而不是串行等待；
结果缓存：对于同一请求中多次用到的相同 key（比如{'id': 'user-123'}），OnDemandServer会在本次请求生命周期内缓存其拉取结果，避免重复 IO；
超时熔断：每个依赖拉取都有独立的timeout_s参数（默认 2s），一旦超时，会立即返回一个预设的 fallback 值（比如0.0或None），并记录告警，绝不会让一个慢依赖阻塞整个 DAG。

其次，它的部署模式非常巧妙。每个OnDemandServer实例绑定一个固定端口（如8001），而操作系统（Linux）的SO_REUSEPORT机制会自动将 Load Balancer 分发过来的连接，以轮询（Round-Robin）方式分发给本机上所有监听该端口的进程。这意味着，你可以在一台 32 核的机器上轻松启动 16 个OnDemandServer，它们共享同一个端口，却能完美实现 CPU 核心级别的负载均衡。我们做过对比测试：用 Nginx 做反向代理，QPS 上限卡在 12000；而直接用SO_REUSEPORT，同一台机器轻松突破 25000 QPS，且 CPU 利用率分布极其均匀。

注意：OnDemandServer的启动参数num_servers_per_node=2并非随意指定。它需要与你的机器 CPU 核心数、内存大小以及特征的计算复杂度做精细匹配。我们的经验是：对于纯 CPU 计算型特征，num_servers_per_node最好等于CPU 核心数 / 2；对于 IO 密集型，则可以设为CPU 核心数 * 1.5。这个值必须通过压测来确定，没有银弹。

2.3 Storage：共享存储不是“数据库”，而是“特征事实的交换站”

Volga 的存储抽象，是整个架构中最容易被误解，也最关键的一环。很多人第一反应是：“哦，就是用 Redis 存 pipeline 特征结果”。这没错，但只说对了一半。Volga 的 Storage，本质是一个语义化的数据契约，它定义了“谁写、谁读、怎么读、读什么”。

写入方（Push Side）：是 Volga 的 Streaming Engine。它运行的每一个pipeline job，其输出目标就是一个PipelineDataConnector。这个 Connector 就像一个“翻译官”，它把 Flink 或 Spark 计算出来的原始数据（可能是 Avro、Parquet、JSON），按照 Volga 定义的 Schema（比如TestEntity的结构），序列化并写入后端存储。关键点在于：Streaming Engine 从不关心下游怎么读，它只负责把“事实”准确、及时、按分区规则地写进去。我们线上用 ScyllaDB，它会自动按feature_name + entity_key做分区，确保单 key 查询的极致性能。
读取方（Pull Side）：是OnDemandServer。但它从不直接操作底层数据库驱动。它只和OnDemandDataConnector打交道。这个 Connector 就像一个“查询策略中心”，它把“我要最新的test_feature值”这个业务意图，翻译成具体的数据库操作。比如InMemoryActorOnDemandDataConnector的fetch_latest方法，最终调用的是cache_actor.get_latest.remote()，而这个cache_actor可能是一个基于 Redis 的 LRU 缓存，也可能是一个基于 RocksDB 的本地磁盘缓存，甚至是一个混合缓存（先查内存，再查 Redis，最后查 Scylla）。业务代码（simple_feature函数）完全不知道也不需要知道这些细节。

这种分离带来了巨大的好处。举个真实案例：我们有一个特征user_recent_clicks，最初用 Redis 存储，但随着用户量增长，Redis 内存暴涨。我们只需修改OnDemandDataConnector的实现，换成一个基于 ScyllaDB 的RangeQueryConnector，并调整query_dict中的'range'映射，所有依赖它的 17 个on_demand特征函数，一行代码都不用改，就能无缝切换到按时间窗口查询的模式。这就是 Volga “存储无关性”设计带来的巨大运维红利。

2.4 Load Balancer：为什么它“不属于 Volga”，却是成败关键？

文档里明确写着：“The Load Balancer is not a part of Volga”。这句话初看有点奇怪，但细想非常精妙。Volga 的设计者清楚地认识到：负载均衡是基础设施层的责任，不是应用框架该管的事。强行把 LB 嵌入框架，只会让 Volga 变得笨重、难以调试、与云厂商绑定。

所以，Volga 只定义了与 LB 的“契约”：它要求 LB 必须能将 HTTP 请求，以最小连接粒度（Connection-level），分发到后端任意一个OnDemandServer的 IP:Port 上。至于 LB 是 AWS ALB、GCP CLB、Nginx 还是 MetalLB，Volga 一概不管。它只提供一个标准的/healthz探针接口，供 LB 做健康检查。

这个“契约”看似简单，实则暗藏玄机。我们线上踩过一个大坑：初期用 Nginx，配置了keepalive 32，结果在高并发下，Nginx 的连接池被占满，大量请求在 LB 层就超时了。后来我们改成 AWS ALB，并启用了connection_draining（连接排空）和slow_start（慢启动）特性，问题迎刃而解。ALB 会自动识别OnDemandServer的启动和退出，并在 Worker 下线前，优雅地将现有连接处理完，再切断。这让我们在做滚动升级时，实现了真正的零感知。

提示：务必在 LB 层配置HTTP 2xx作为健康检查的成功状态码，并将/healthz的超时时间设为1s以内。OnDemandServer的/healthz接口会检查其内部的 asyncio 事件循环是否卡死、以及与 Coordinator 的心跳是否正常，这是一个非常可靠的探针。

3. 实操全流程：从写第一个特征到线上压测，手把手带你走通

光看架构图是不够的，真正的理解来自于亲手敲下每一行代码，并看着它在生产环境里稳定运行。下面我将以一个最典型的“实时搜索个性化得分”特征为例，完整复现从开发、测试、部署到压测的全过程。所有命令和配置，都来自我们线上真实环境的简化版，你可以直接复制粘贴使用。

3.1 第一步：定义你的第一个 Pipeline 特征（事件时间）

一切始于数据源。我们假设有一个 Kafka Topicuser_search_events，里面是用户搜索行为的原始日志。我们要做的，是用 Volga 的 Streaming Engine，实时计算出每个用户的“最近 3 次搜索关键词的 TF-IDF 向量”，并存入共享存储。

# features/pipeline/search_tfidf.py from volga.api.source import source from volga.api.pipeline import pipeline from volga.api.entity import Entity from typing import List, Dict, Any import numpy as np from sklearn.feature_extraction.text import TfidfVectorizer # 定义实体：这是 Volga 的“数据契约” class SearchVectorEntity(Entity): user_id: str vector: List[float] # 128维的TF-IDF向量 timestamp: float # 数据源：从Kafka读取原始事件 @source(SearchVectorEntity) def search_events_source() -> KafkaConnector: return KafkaConnector( bootstrap_servers="kafka-broker:9092", topic="user_search_events", group_id="volga-search-tfidf", value_deserializer=lambda x: json.loads(x.decode('utf-8')) ) # Pipeline特征：核心计算逻辑 @pipeline(input_sources=[search_events_source], output_entity=SearchVectorEntity) def recent_search_tfidf() -> SearchVectorEntity: # 1. 使用Flink State存储每个用户的最近10次搜索词 # （实际代码会用Flink的ListState，此处为示意） user_search_history = get_user_search_history() # 返回 [str, str, str] # 2. 构建TF-IDF向量（简化版，实际会用更复杂的模型） vectorizer = TfidfVectorizer(max_features=128, stop_words='english') tfidf_matrix = vectorizer.fit_transform(user_search_history) # 取最后一行（最新搜索）的向量 latest_vector = tfidf_matrix[-1].toarray()[0].tolist() return SearchVectorEntity( user_id=get_current_user_id(), vector=latest_vector, timestamp=time.time() )

这段代码的关键，在于@pipeline装饰器。它告诉 Volga：这是一个需要由 Streaming Engine 在事件时间（Event Time）下持续运行的作业。recent_search_tfidf函数本身不包含任何“如何启动 Flink”的逻辑，它只描述“我要做什么”。Volga 的编译器会自动将它转换成 Flink JobGraph，并提交到集群。我们线上用的是 Flink 1.17，这个作业会 24/7 运行，将结果源源不断地写入 ScyllaDB 的search_tfidf表中。

3.2 第二步：编写你的第一个 On-Demand 特征（请求时间）

现在，当用户在搜索框输入“手机”，并点击搜索时，后端服务需要实时获取这个用户的search_tfidf向量，并与“手机”这个词的向量做余弦相似度，得到一个个性化得分。这个计算，必须在请求到来的那一刻发生。

# features/on_demand/search_personalization.py from volga.api.on_demand import on_demand from volga.api.entity import Entity from typing import List, Dict, Any import numpy as np from scipy.spatial.distance import cosine # 定义输入实体（来自Pipeline） class SearchVectorEntity(Entity): user_id: str vector: List[float] timestamp: float # 定义输出实体（本次请求的计算结果） class PersonalizationScoreEntity(Entity): user_id: str score: float query: str timestamp: float # On-Demand特征：依赖Pipeline特征 @on_demand( dependencies=[ ('recent_search_tfidf', 'latest') # 依赖recent_search_tfidf的最新值 ], resource_requirements={"cpu": 1, "memory": 2048} # 声明资源需求 ) def search_personalization_score( dep: SearchVectorEntity, # Pipeline特征的输入 query: str = "default" # 请求时传入的参数 ) -> PersonalizationScoreEntity: """ 计算用户搜索历史与当前查询词的相似度得分 """ # 1. 加载预训练的query向量模型（只在首次调用时加载，后续复用） if not hasattr(search_personalization_score, 'query_model'): search_personalization_score.query_model = load_query_embedding_model() # 2. 获取当前查询词的向量 query_vector = search_personalization_score.query_model.encode([query])[0] # 3. 计算余弦相似度 similarity = 1 - cosine(dep.vector, query_vector) return PersonalizationScoreEntity( user_id=dep.user_id, score=float(similarity), query=query, timestamp=time.time() )

注意几个关键点：

dependencies参数中的'latest'，对应OnDemandDataConnector中query_dict定义的fetch_latest方法；
resource_requirements会被 Coordinator 用来做 Worker 的智能分组；
query: str = "default"这个参数，会自动从 HTTP 请求的 JSON body 中提取，无需手动解析。

3.3 第三步：启动 Coordinator 和 Server，完成注册

现在，我们有了特征代码，接下来是启动框架。

# 1. 启动Ray集群（假设已在本地或K8s上运行） ray start --head --port=6379 # 2. 启动Coordinator（在Python脚本中） # coordinator_launcher.py from volga.core.on_demand.coordinator import create_on_demand_coordinator from volga.core.on_demand.config import OnDemandConfig from volga.core.on_demand.connector import MockOnDemandDataConnector config = OnDemandConfig( num_servers_per_node=4, server_port=8001, data_connector=OnDemandDataConnectorConfig( connector_class=MockOnDemandDataConnector, connector_args={} ) ) coordinator = create_on_demand_coordinator(config) ray.get(coordinator.start.remote()) # 注册所有特征（会自动解析依赖） from volga.core.feature.repository import FeatureRepository features = FeatureRepository.get_features_with_deps(['search_personalization_score']) ray.get(coordinator.register_features.remote(features)) print("✅ Coordinator started and features registered!")

# 3. 启动OnDemandServer（在另一台机器或容器中） # server_launcher.py from volga.core.on_demand.server import create_on_demand_server from volga.core.on_demand.config import OnDemandServerConfig server_config = OnDemandServerConfig( host="0.0.0.0", port=8001, coordinator_address="ray://localhost:10001", # Coordinator的Ray地址 features_to_handle=['search_personalization_score'] # 指定本Worker处理哪些特征 ) server = create_on_demand_server(server_config) server.run() # 启动Starlette服务

启动后，你可以用curl直接测试：

curl -X POST http://localhost:8001/v1/feature \ -H "Content-Type: application/json" \ -d '{ "target_features": ["search_personalization_score"], "feature_keys": { "search_personalization_score": [{"user_id": "user-123"}] }, "udf_args": { "search_personalization_score": {"query": "手机"} } }'

你会看到一个包含score字段的 JSON 响应。恭喜，你的第一个按需特征已经跑通！

3.4 第四步：线上压测与性能调优（这才是真功夫）

压测不是为了刷数字，而是为了暴露瓶颈。我们用locust工具，模拟 1000 个并发用户，持续请求search_personalization_score。

# locustfile.py from locust import HttpUser, task, between import json class FeatureUser(HttpUser): wait_time = between(0.5, 2.0) @task def get_personalization_score(self): self.client.post("/v1/feature", json={ "target_features": ["search_personalization_score"], "feature_keys": {"search_personalization_score": [{"user_id": f"user-{self.random.randint(1, 10000)}"}]}, "udf_args": {"search_personalization_score": {"query": "手机"}} })

压测结果出来后，我们重点关注三个指标：

P99 延迟：目标 < 200ms；
错误率：目标 0%；
CPU/内存利用率：各 Worker 是否均衡。

第一次压测，P99 达到了 320ms。我们用py-spy record抓取了OnDemandServer的火焰图，发现 65% 的时间花在了load_query_embedding_model()这个函数上——它在每次请求时都被重复加载！这就是典型的“冷启动”问题。

解决方案：利用OnDemandServer的进程生命周期，在__init__方法中做一次性的模型加载。

# 修改后的特征函数（关键改动） class SearchPersonalizationFeature: _model = None # 类变量，所有实例共享 def __init__(self): if SearchPersonalizationFeature._model is None: SearchPersonalizationFeature._model = load_query_embedding_model() @on_demand(dependencies=[('recent_search_tfidf', 'latest')]) def search_personalization_score(self, dep: SearchVectorEntity, query: str) -> PersonalizationScoreEntity: # ... 其余代码不变，直接使用 self._model ...

再次压测，P99 降到 145ms。接着，我们发现当并发从 1000 升到 2000 时，错误率开始上升。查看日志，是fetch_latest调用 ScyllaDB 超时了。于是我们调整OnDemandDataConnector的fetch_latest方法，增加重试和降级逻辑：

async def fetch_latest(self, feature_name: str, keys: List[Dict[str, Any]]) -> List[List[Any]]: try: # 首次尝试，带超时 return await asyncio.wait_for( self.cache_actor.get_latest.remote(feature_name, keys), timeout=0.5 ) except asyncio.TimeoutError: # 降级：返回一个预设的默认向量 default_vector = [0.0] * 128 return [[default_vector]]

经过这两轮调优，系统在 2000 QPS 下，P99 稳定在 160ms，错误率为 0%。这证明了 Volga 的架构是经得起实战考验的。

4. 常见问题与独家避坑指南：那些文档里不会写的血泪教训

在把 Volga 推上生产环境的半年里，我和团队踩过的坑，比读过的文档还多。下面这些，全是来自真实战场的第一手经验，每一条都附带了“为什么”和“怎么做”，希望能帮你绕开那些我们已经趟过的雷。

4.1 问题：特征函数里调用`time.sleep(1)`，整个 Worker 进程就卡死了！

现象：一个开发同学为了模拟第三方 API 的延迟，在on_demand函数里加了time.sleep(1)，结果导致该 Worker 处理的所有请求都卡住，P99 延迟瞬间飙升到 5s 以上。

原因：OnDemandServer是基于asyncio的单线程事件循环。time.sleep()是一个同步阻塞调用，它会让整个事件循环停下来 1 秒，期间无法处理任何其他请求、无法响应心跳、无法执行任何协程。这就像在一条单行道上，一辆车突然熄火，后面所有车都得等着。

正确做法：永远使用await asyncio.sleep(1)。它是异步的，会把控制权交还给事件循环，让其他协程得以运行。对于真正的 IO 操作（如 HTTP 请求），必须使用aiohttp、httpx等异步库，而不是requests。

提示：在 CI/CD 流水线中，加入一个静态代码检查规则，禁止在on_demand函数中出现time.sleep、requests.get、open()等同步调用。我们用pylint的banned-api规则实现了这一点。

4.2 问题：`OnDemandCoordinator`频繁重启，特征注册失败，日志里全是`RayActorError`

现象：Coordinator 的 Pod 在 K8s 里反复 CrashLoopBackOff，kubectl logs显示RayActorError: Failed to connect to GCS。

原因：Coordinator 是一个 Ray Actor，它需要与 Ray 的全局控制服务（GCS）保持长连接。当网络抖动、GCS 重启、或 Coordinator 自身内存泄漏时，这个连接就会断开。而 Volga 的默认重连策略比较激进，一旦断开就直接抛异常退出。

解决方案：在启动 Coordinator 时，显式配置更稳健的重连参数：

config = OnDemandConfig( # ... 其他配置 gcs_retry_policy={ "max_retries": 10, "initial_backoff_s": 1.0, "max_backoff_s": 30.0, "jitter": True } )

更重要的是，给 Coordinator 的容器分配足够的内存。我们线上给 Coordinator 分配了 4GB 内存，因为它的内部状态（特征注册表、Worker 映射关系）会随着特征数量线性增长。一个只有 512MB 的 Coordinator，在注册 200+ 特征后，OOM Killer 就会把它干掉。

4.3 问题：`OnDemandServer`的`/healthz`探针一直返回 503，LB 把它踢出了服务池

现象：Nginx 的 upstream 显示某个OnDemandServer的健康检查失败，但curl http://ip:port/healthz却能成功返回 200。

原因：/healthz接口的实现，除了检查自身进程状态，还会检查与OnDemandCoordinator的心跳连接。如果 Coordinator 的网络延迟很高（比如跨 AZ），或者 Coordinator 本身负载过高，OnDemandServer就会认为“协调失联”，从而返回 503。

排查步骤：

在OnDemandServer容器内，执行telnet coordinator-service 10001，确认网络连通性；
查看OnDemandServer日志，搜索coordinator heartbeat关键字，看是否有timeout或connection refused；
检查 Coordinator 的 CPU 和内存指标，确认是否过载。

终极方案：在 K8s 的livenessProbe中，不要直接用/healthz，而是用一个更轻量的探针：

livenessProbe: httpGet: path: /ping port: 8001 initialDelaySeconds: 30 periodSeconds: 10

/ping接口只检查进程是否存活，不依赖 Coordinator。而/healthz仅用于 LB 的readinessProbe，确保只有健康的、能连上 Coordinator 的 Worker 才接收流量。

4.4 问题：`OnDemandDataConnector`的`fetch_range`查询，ScyllaDB 返回了 10 万条记录，`OnDemandServer`内存爆了

现象：一个按时间窗口查询的特征，在高峰期导致OnDemandServer的内存使用率瞬间冲到 95%，触发 OOM。

原因：fetch_range的语义是“拉取某个时间范围内的所有数据”，但如果这个范围太大（比如start=0, end=time.time()），而底层存储又没有做分区裁剪，就可能把全表数据都扫出来。Volga 的OnDemandServer默认会把所有结果加载到内存中再处理，这就成了内存炸弹。

安全实践：

强制分页：在OnDemandDataConnector的fetch_range方法中，加入limit参数，并在调用方（特征函数）中显式声明limit=100。
服务端过滤：确保你的PipelineDataConnector在写入时，就按feature_name + entity_key + timestamp做了复合主键。这样fetch_range就能利用 ScyllaDB 的分区键进行高效过滤，而不是全表扫描。
客户端熔断：在OnDemandServer的配置中，设置max_fetch_result_size_mb=10，一旦单次拉取结果超过 10MB，就直接报错，拒绝执行。

我们线上所有fetch_range类型的 Connector，都强制要求在 SQL 或 CQL 查询中带上LIMIT子句，并在单元测试中覆盖limit=1和limit=1000的边界 case。

4.5 问题：特征之间循环依赖，`OnDemandCoordinator`启动时报`Cycle detected in feature graph`

现象：coordinator.register_features.remote(...)报错，提示Cycle detected in feature graph: A -> B -> C -> A。

原因：Volga 的 DAG 执行器要求特征依赖图必须是有向无环图（DAG）。循环依赖意味着 A 需要 B 的结果，B 需要 C 的结果，C 又需要 A 的结果，这在逻辑上是死锁的，无法执行。

常见场景：

特征 A 依赖user_profile的最新值，而user_profile这个 pipeline 特征，其计算逻辑里又调用了feature A的某个中间结果（比如一个统计值）；
两个on_demand特征互相调用，形成 A->B->A 的闭环。

解决思路：

重构依赖：这是最根本的。问问自己：这个循环真的是业务必需的吗？很多时候，是设计时为了“方便”而引入的耦合。把公共逻辑抽成一个独立的、无依赖的utility函数，让 A 和 B 都去调用它。
引入缓存层：如果 A 确实需要 B 的结果，而 B 的计算又很慢，可以考虑让 B 的结果写入一个专门的cache_table，然后 A 通过fetch_latest去读这个cache_table，而不是直接依赖 B。这打破了直接的函数调用依赖。
使用@on_demand的fallback参数：在定义 A 时，给dependencies加上fallback，当 B 不可用时，A 可以用一个默认值继续执行，避免整个链路卡死。