构建以Memcached为核心特征缓存的可观测模型部署架构

架构设计

文章字数: 3.4k

阅读时长: 15 分

模型推理服务的P99延迟，是悬在每个算法工程师和SRE头上的达摩克利斯之剑。在一个典型的在线推理场景中，延迟的主要来源往往不是模型计算本身，而是前置的特征工程——从多个数据源拉取、转换、组合成模型所需的特征向量。当QPS攀升，对底层特征存储（如HBase、MySQL）的重复查询会迅速压垮整个系统，导致SLO（服务等级目标）彻底失效。

显而易见的解决方案是引入缓存。但一个未经深思熟虑的缓存层，在真实项目中往往会演变成一个难以诊断的“黑盒”。当延迟再次抖动时，问题会变得更加棘手：是缓存穿透了？是缓存节点网络延迟？还是缓存实例本身达到了性能瓶颈？缺乏足够的可观测性，任何优化都无异于盲人摸象。

定义问题：在缓存与可观测性之间抉择

我们面对的挑战是为一个高并发的推荐模型推理服务设计一套特征缓存架构。该服务对P9Š9延迟要求在50ms以内。特征获取是主要瓶颈，涉及对用户画像、物品属性等多个数据源的查询。

方案A：简单、直接的“黑盒”缓存

最快的实现方式是在推理服务中直接集成一个标准的Memcached客户端。当请求到来时，服务首先根据请求ID或用户ID生成一个缓存键（Cache Key），查询Memcached。

命中（Hit）: 直接返回特征，进入模型推理阶段。
未命中（Miss）: 回源到特征存储获取数据，成功后将结果异步写入Memcached，然后返回给推理阶段。

# 方案A的伪代码示意 - 一个天真的实现
import time
from pymemcache.client.base import Client

# 假设的特征获取函数
def fetch_features_from_source(user_id: str) -> dict:
    # 模拟耗时的I/O操作
    time.sleep(0.1) 
    return {"feature1": 0.5, "feature2": 0.8}

MEMCACHED_CLIENT = Client(('memcached-server', 11211))

def get_user_features_simple(user_id: str) -> dict:
    cache_key = f"user_features:{user_id}"
    
    # 1. 查询缓存
    cached_result = MEMCACHED_CLIENT.get(cache_key)
    
    if cached_result:
        # 反序列化
        return json.loads(cached_result.decode('utf-8'))
    else:
        # 2. 缓存未命中，回源
        features = fetch_features_from_source(user_id)
        
        # 3. 异步写入缓存，设置1小时过期
        # 这里的错误处理非常粗糙
        try:
            MEMCACHED_CLIENT.set(cache_key, json.dumps(features).encode('utf-8'), expire=3600)
        except Exception as e:
            # 在生产环境中，这里应该有详细的日志记录
            print(f"Failed to set cache: {e}")
            
        return features

方案A的优劣分析:

优点: 实现简单，立竿见影。对于高频访问的相同用户，可以显著降低平均延迟。
缺点 (致命的):
1. 缺乏可观测性: 我们无法回答以下关键问题：缓存命中率是多少？缓存查询本身的延迟是多少？Memcached实例的内存使用率、驱逐（eviction）率、连接数是否健康？
2. 诊断困难: 当P99延迟再次超标时，我们无法区分是缓存命中率下降（例如，大量新用户涌入），还是Memcached服务本身出了问题。
3. 脆弱的错误处理: 上述代码中的try-except块过于简单。如果Memcached持续不可用，会导致每次请求都尝试连接，增加不必要的开销和延迟，甚至可能拖垮整个推理服务。

在真实项目中，一个不可观测的组件就是一颗定时炸弹。方案A因为其“黑盒”属性，在架构评审阶段被直接否决。

方案B：可观测性驱动的缓存架构

这个方案的核心思想是：将缓存层视为一等公民，其内部状态和性能指标必须完全透明，并纳入统一的监控体系。这要求我们做两件事：

构建一个“智能”的、自带埋点的缓存客户端，在应用层暴露丰富的性能指标。
部署一个针对Memcached的定制化Exporter，将Memcached底层的运行状态暴露给Prometheus。

最终，我们将应用层指标（如业务逻辑层面的缓存命中率）和基础设施层指标（如Memcached的内存使用）结合起来，形成一个完整的监控视图。

graph TD
    subgraph "Kubernetes Cluster"
        subgraph "Inference Service Pod"
            A[Inference App] --> B{Smart Cache Client};
            B -- TCP --> C[Memcached Instance];
            D[App Metrics Exporter] -- Scrape --> E[Prometheus];
            A -- Exposes Metrics --> D;
        end
        subgraph "Monitoring Pod"
             F[Custom Memcached Exporter] -- "STATS command" --> C;
             F -- Scrape --> E;
        end
    end
    
    G[Grafana] -- Queries --> E;
    H[Alertmanager] -- Alerts --> E;

方案B的优劣分析:

优点:
1. 深度可观测性: 提供了从应用到基础设施的全链路监控。我们可以精确度量缓存带来的性能提升，并对缓存失效、性能瓶颈等问题设置精确告警。
2. 数据驱动决策: 基于历史数据，可以分析缓存命中率与业务活动的关系，指导缓存容量规划、过期策略调整。例如，我们可以判断是否需要为不同的特征设置不同的TTL。
3. 强大的故障排查能力: 当问题发生时，可以通过关联应用指标和Memcached指标快速定位根源。
缺点:
1. 实现复杂度更高: 需要开发和维护智能客户端以及自定义的Exporter。
2. 轻微性能开销: 指标的收集和暴露会带来微不足道的CPU和网络开销，但在一个健壮的系统中，这是完全可以接受的成本。

最终决策:

我们毫不犹豫地选择了方案B。对于一个核心在线服务，长期的可维护性和稳定性远比短期的开发速度重要。一次因为“黑盒”问题导致的线上故障，其损失足以覆盖构建一套完善监控体系所需成本的数倍。

核心实现概览

我们将分三部分实现方案B：智能客户端、自定义Exporter和Prometheus配置。

1. Python智能缓存客户端 (Smart Cache Client)

我们将封装pymemcache，并使用官方的prometheus-client库来暴露指标。

# smart_cache_client.py
import json
import logging
import time
from contextlib import contextmanager
from typing import Optional, Dict, Any

from prometheus_client import Counter, Histogram
from pymemcache.client.base import Client
from pymemcache.exceptions import MemcacheError

# --- Prometheus Metrics Definition ---
# 使用 namespace 和 subsystem 来组织指标
NAMESPACE = "model_serving"
SUBSYSTEM = "feature_cache"

CACHE_REQUESTS_TOTAL = Counter(
    "cache_requests_total",
    "Total number of cache requests.",
    ["cache_name", "operation"], # operation: get, set, delete
    namespace=NAMESPACE,
    subsystem=SUBSYSTEM,
)

CACHE_HITS_TOTAL = Counter(
    "cache_hits_total",
    "Total number of cache hits.",
    ["cache_name"],
    namespace=NAMESPACE,
    subsystem=SUBSYSTEM,
)

CACHE_LATENCY_SECONDS = Histogram(
    "cache_latency_seconds",
    "Latency of cache operations in seconds.",
    ["cache_name", "operation"],
    namespace=NAMESPACE,
    subsystem=SUBSYSTEM,
    # buckets可以根据实际延迟分布进行调整
    buckets=[0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0]
)

CACHE_OPERATION_ERRORS_TOTAL = Counter(
    "cache_operation_errors_total",
    "Total number of cache operation errors.",
    ["cache_name", "operation", "reason"], # reason: connection_error, serialization_error, etc.
    namespace=NAMESPACE,
    subsystem=SUBSYSTEM,
)

# --- Logger Setup ---
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

class SmartMemcachedClient:
    """
    An observable Memcached client wrapper.
    """
    def __init__(self, server: tuple, cache_name: str, connect_timeout=5, timeout=5):
        self.client = Client(server, connect_timeout=connect_timeout, timeout=timeout)
        self.cache_name = cache_name
        logger.info(f"SmartMemcachedClient for '{cache_name}' initialized for server {server}")

    @contextmanager
    def _instrument(self, operation: str):
        """A context manager to handle instrumentation for an operation."""
        start_time = time.perf_counter()
        CACHE_REQUESTS_TOTAL.labels(cache_name=self.cache_name, operation=operation).inc()
        try:
            yield
        finally:
            latency = time.perf_counter() - start_time
            CACHE_LATENCY_SECONDS.labels(cache_name=self.cache_name, operation=operation).observe(latency)
    
    def get(self, key: str) -> Optional[Dict[str, Any]]:
        with self._instrument("get"):
            try:
                value = self.client.get(key.encode('utf-8'))
                if value is not None:
                    CACHE_HITS_TOTAL.labels(cache_name=self.cache_name).inc()
                    # 这里的反序列化失败也应该被视为一种错误
                    try:
                        return json.loads(value.decode('utf-8'))
                    except (json.JSONDecodeError, UnicodeDecodeError) as e:
                        logger.error(f"Deserialization error for key '{key}': {e}")
                        CACHE_OPERATION_ERRORS_TOTAL.labels(
                            cache_name=self.cache_name, 
                            operation="get", 
                            reason="deserialization_error"
                        ).inc()
                        return None
                else:
                    # Miss is not an error, just a cache miss.
                    return None
            except MemcacheError as e:
                logger.error(f"Memcached 'get' operation failed for key '{key}': {e}")
                CACHE_OPERATION_ERRORS_TOTAL.labels(
                    cache_name=self.cache_name, 
                    operation="get", 
                    reason="memcache_error"
                ).inc()
                return None

    def set(self, key: str, value: Dict[str, Any], expire: int = 3600):
        with self._instrument("set"):
            try:
                # 序列化失败是应用层问题，必须捕获
                serialized_value = json.dumps(value).encode('utf-8')
                self.client.set(key.encode('utf-8'), serialized_value, expire=expire)
            except TypeError as e:
                logger.error(f"Serialization error for key '{key}': {e}")
                CACHE_OPERATION_ERRORS_TOTAL.labels(
                    cache_name=self.cache_name, 
                    operation="set", 
                    reason="serialization_error"
                ).inc()
            except MemcacheError as e:
                logger.error(f"Memcached 'set' operation failed for key '{key}': {e}")
                CACHE_OPERATION_ERRORS_TOTAL.labels(
                    cache_name=self.cache_name, 
                    operation="set", 
                    reason="memcache_error"
                ).inc()

# --- Example Usage in a FastAPI/Flask application ---
# from prometheus_client import make_wsgi_app
# 
# # In your app setup:
# feature_cache_client = SmartMemcachedClient(
#     server=('memcached.default.svc.cluster.local', 11211),
#     cache_name="user_features_v1"
# )
#
# # In your inference endpoint:
# def get_features(user_id: str):
#     cache_key = f"user_features:{user_id}"
#     features = feature_cache_client.get(cache_key)
#     if features:
#         return features
#     
#     # Miss, go to source
#     features_from_source = fetch_features_from_source(user_id)
#     feature_cache_client.set(cache_key, features_from_source)
#     return features_from_source
#
# # Expose metrics endpoint (e.g., /metrics)
# # For Flask: app.add_url_rule('/metrics', 'metrics', make_wsgi_app())

这个客户端的关键在于_instrument上下文管理器，它确保了每次操作的耗时、计数和错误都被精确记录。标签（cache_name, operation）的使用至关重要，它允许我们在Grafana中对数据进行灵活的切分和聚合。

2. 自定义Memcached Exporter

虽然社区有现成的memcached_exporter，但为了演示其原理并确保能获取我们最关心的指标，我们在此处实现一个简化版的Python Exporter。它通过telnet或socket连接到Memcached，执行stats命令，并解析其输出。

# memcached_exporter.py
import socket
import time
import re
from prometheus_client import start_http_server, Gauge, Counter
from typing import Dict, Tuple

# --- Exporter Configuration ---
MEMCACHED_HOST = 'localhost'
MEMCACHED_PORT = 11211
EXPORTER_PORT = 9151

# --- Prometheus Metric Definitions ---
# 使用Gauge因为这些值是瞬时的状态
MEMCACHED_UP = Gauge('memcached_up', 'Value is 1 if Memcached is reachable, 0 otherwise.')
MEMCACHED_GET_HITS = Counter('memcached_get_hits_total', 'Total number of get hits.')
MEMCACHED_GET_MISSES = Counter('memcached_get_misses_total', 'Total number of get misses.')
MEMCACHED_EVICTIONS = Counter('memcached_evictions_total', 'Total number of evictions.')
MEMCACHED_BYTES = Gauge('memcached_bytes', 'Current number of bytes used by this server to store items.')
MEMCACHED_LIMIT_MAXBYTES = Gauge('memcached_limit_maxbytes', 'Number of bytes this server is allowed to use for storage.')
MEMCACHED_CURR_ITEMS = Gauge('memcached_curr_items', 'Current number of items stored by the server.')
MEMCACHED_CURR_CONNECTIONS = Gauge('memcached_curr_connections', 'Current number of open connections.')

# 我们关心的统计项及其对应的Prometheus Metric对象
# 'stat_name': (metric_object, metric_type)
STATS_MAPPING = {
    'get_hits': (MEMCACHED_GET_HITS, 'counter'),
    'get_misses': (MEMCACHED_GET_MISSES, 'counter'),
    'evictions': (MEMCACHED_EVICTIONS, 'counter'),
    'bytes': (MEMCACHED_BYTES, 'gauge'),
    'limit_maxbytes': (MEMCACHED_LIMIT_MAXBYTES, 'gauge'),
    'curr_items': (MEMCACHED_CURR_ITEMS, 'gauge'),
    'curr_connections': (MEMCACHED_CURR_CONNECTIONS, 'gauge'),
}

def get_memcached_stats(host: str, port: int) -> Dict[str, int]:
    """Connects to Memcached and gets stats."""
    try:
        with socket.create_connection((host, port), timeout=2) as s:
            s.sendall(b'stats\r\n')
            buffer = b''
            while True:
                chunk = s.recv(4096)
                if not chunk:
                    break
                buffer += chunk
                if b'END\r\n' in buffer:
                    break
            
            stats_data = buffer.decode('utf-8')
            MEMCACHED_UP.set(1)
            
            parsed_stats = {}
            # STAT pid 1
            # STAT uptime 3433
            # ...
            # END
            for line in stats_data.splitlines():
                if line.startswith('STAT'):
                    parts = line.split()
                    if len(parts) == 3:
                        _, key, value = parts
                        try:
                            parsed_stats[key] = int(value)
                        except ValueError:
                            # Ignore non-integer stats
                            pass
            return parsed_stats
            
    except (socket.error, socket.timeout) as e:
        print(f"Error connecting to Memcached: {e}")
        MEMCACHED_UP.set(0)
        return {}

def process_metrics():
    """Periodically fetches stats and updates Prometheus metrics."""
    # 用于Counter类型指标的本地状态，因为Prometheus Counter只能增不能减
    # Memcached的stats是自启动以来的累计值，正好适合
    print("Exporter starting...")
    while True:
        stats = get_memcached_stats(MEMCACHED_HOST, MEMCACHED_PORT)
        if stats:
            for stat_name, (metric, metric_type) in STATS_MAPPING.items():
                value = stats.get(stat_name)
                if value is not None:
                    if metric_type == 'gauge':
                        metric.set(value)
                    elif metric_type == 'counter':
                        # The _total attribute is managed internally by the client library
                        # We need to set the total value each time.
                        # This is a bit of a hack for Counter; a better way might be to track delta.
                        # But since memcached stats are cumulative, we can use a gauge with a _total suffix
                        # to signal its nature. For true Counter semantics with `inc`, we'd need to store the last value.
                        # For simplicity, we directly update the underlying value.
                        metric._value.set(value) # Directly set the counter's internal value

        time.sleep(15) # Scrape interval

if __name__ == '__main__':
    start_http_server(EXPORTER_PORT)
    print(f"Memcached exporter listening on port {EXPORTER_PORT}")
    process_metrics()

这个Exporter会定期从Memcached拉取统计信息，并将其转换为Prometheus格式。一个关键的指标是evictions（驱逐次数）。如果这个值持续增长，说明缓存空间不足，最老的数据正在被强制删除以为新数据腾出空间，这通常是缓存命中率下降的直接原因。

3. Prometheus与告警配置

最后一步是让Prometheus来抓取这些新暴露的指标。

# prometheus.yml

# ... global config ...

scrape_configs:
  - job_name: 'model-serving-app'
    # 使用Kubernetes服务发现来自动找到所有推理服务的Pod
    kubernetes_sd_configs:
      - role: pod
    relabel_configs:
      # 只选择带有 app=model-serving 标签的Pod
      - source_labels: [__meta_kubernetes_pod_label_app]
        action: keep
        regex: model-serving
      # 选择容器端口名为 'http-metrics' 的
      - source_labels: [__meta_kubernetes_pod_container_port_name]
        action: keep
        regex: http-metrics
        
  - job_name: 'memcached-exporter'
    static_configs:
      # 假设Exporter部署为一个单独的服务
      - targets: ['memcached-exporter.monitoring.svc.cluster.local:9151']

# alert.rules.yml
groups:
- name: FeatureCacheAlerts
  rules:
  - alert: HighCacheMissRate
    expr: |
      sum(rate(model_serving_feature_cache_requests_total{cache_name="user_features_v1", operation="get"}[5m]))
      -
      sum(rate(model_serving_feature_cache_hits_total{cache_name="user_features_v1"}[5m]))
      /
      sum(rate(model_serving_feature_cache_requests_total{cache_name="user_features_v1", operation="get"}[5m]))
      > 0.3
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "High cache miss rate for user_features_v1"
      description: "Cache hit rate dropped below 70% for the last 5 minutes. Current value is {{$value | humanizePercentage}}."

  - alert: HighMemcachedEvictions
    expr: rate(memcached_evictions_total[5m]) > 10
    for: 10m
    labels:
      severity: critical
    annotations:
      summary: "Memcached is evicting items"
      description: "Memcached eviction rate is over 10 per second, indicating insufficient memory. Investigate increasing cache size."

  - alert: HighP99CacheLatency
    expr: |
      histogram_quantile(0.99, sum(rate(model_serving_feature_cache_latency_seconds_bucket{cache_name="user_features_v1"}[5m])) by (le))
      > 0.1
    for: 2m
    labels:
      severity: critical
    annotations:
      summary: "High P99 latency on feature cache"
      description: "The 99th percentile latency for feature cache operations has exceeded 100ms."

通过这些配置，我们不仅能在一个Grafana仪表盘上同时观察到应用层的缓存命中率和底层的内存驱逐率，还能设置精准的告警，在问题萌芽阶段就介入处理。

架构的扩展性与局限性

这套以可观测性为核心的缓存架构模式，其价值远超Memcached本身。封装客户端并提供定制化Exporter的思路，可以平滑地迁移到Redis或其他任何数据存储中间件上。它倡导了一种文化：任何引入架构的组件都不能是“黑盒”，其成本必须包含实现其可观测性的成本。

当然，当前方案也存在局限性。首先，它只解决了“读”的缓存。对于需要“写”后立即读到的强一致性场景，需要更复杂的缓存同步或失效策略。其次，我们实现的Exporter较为简单，一个生产级的Exporter还需要处理更复杂的网络异常和Memcached多实例的聚合问题。最后，对于缓存未命中导致的“惊群效应”（Thundering Herd），即大量请求同时回源同一个资源，本架构并未解决，可能需要引入请求合并（Request Coalescing）等更高级的模式作为下一阶段的迭代。

架构设计 Memcached Prometheus 模型部署

eBPF 结合 MapReduce 实现大规模主机集群的离线行为审计架构

2023-10-27 架构设计

eBPF Puppet MapReduce 可观测性大数据

构建从Zustand到Ktor的全栈可观测性整合Sentry与Pulumi实现分布式追踪

2023-10-27 可观测性

Sentry Pulumi Ktor Zustand Turbopack