基于 Python 的分布式配置中心客户端设计与实现

"""
分布式配置中心客户端
====================

模块覆盖：
1. 配置模型与多格式解析（JSON/YAML/Properties）
2. HTTP 长轮询变更监听
3. 本地文件缓存与原子写入
4. 命名空间隔离与版本管理
5. 热更新回调机制
6. 故障降级与重试策略
7. 配置快照与回滚
8. 监控统计
"""

import os
import sys
import json
import time
import hashlib
import logging
import threading
import traceback
from enum import Enum
from typing import Any, Callable, Dict, List, Optional, Set
from pathlib import Path
from dataclasses import dataclass, field
from urllib.parse import urljoin
from concurrent.futures import ThreadPoolExecutor

# ==================== 日志配置 ====================

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    handlers=[logging.StreamHandler(sys.stdout)],
)
logger = logging.getLogger("config_client")


# ==================== 常量 ====================

DEFAULT_SERVER_URL = "http://localhost:8888"
DEFAULT_NAMESPACE = "default"
DEFAULT_GROUP = "DEFAULT_GROUP"
DEFAULT_DATA_ID = "application"
DEFAULT_CACHE_DIR = "./config_cache"
DEFAULT_POLL_TIMEOUT = 30
DEFAULT_POLL_INTERVAL = 1
DEFAULT_MAX_RETRIES = 3
DEFAULT_RETRY_BACKOFF = 2.0
DEFAULT_CONNECT_TIMEOUT = 5
DEFAULT_READ_TIMEOUT = 35

CONFIG_FORMAT_JSON = "json"
CONFIG_FORMAT_YAML = "yaml"
CONFIG_FORMAT_PROPERTIES = "properties"
CONFIG_FORMAT_TEXT = "text"


# ==================== 配置格式解析器 ====================

class ConfigParser: 16044.m.kuaisou.com 
    """多格式配置解析器"""

    @staticmethod
    def parse(content: str, fmt: str) -> Dict[str, Any]: 16046.m.kuaisou.com 
        if not content or not content.strip(): 16045.m.kuaisou.com 
            return {}
        try:
            if fmt == CONFIG_FORMAT_JSON: 16048.m.kuaisou.com 
                return json.loads(content)
            elif fmt == CONFIG_FORMAT_YAML: 16049.m.kuaisou.com 
                return ConfigParser._parse_yaml(content)
            elif fmt == CONFIG_FORMAT_PROPERTIES: 16050.m.kuaisou.com 
                return ConfigParser._parse_properties(content)
            else: 16051.m.kuaisou.com 
                return {"_raw": content}
        except Exception as e: 16052.m.kuaisou.com 
            logger.error(f"Failed to parse config ({fmt}): {e}")
            return {"_raw": content, "_parse_error": str(e)}

    @staticmethod
    def _parse_yaml(content: str) -> Dict[str, Any]: 16053.m.kuaisou.com 
        """简易 YAML 解析（仅支持扁平 key: value）"""
        result = {}
        for line in content.splitlines(): 16054.m.kuaisou.com 
            line = line.strip()
            if not line or line.startswith("#"): 16055.m.kuaisou.com 
                continue
            if ":" in line: 16056.m.kuaisou.com 
                key, _, value = line.partition(":")
                key = key.strip()
                value = value.strip(16057.m.kuaisou.com)
                # 尝试类型转换
                if value.lower() in ("true", "false"): 16059.m.kuaisou.com 
                    result[key] = value.lower() == "true"
                elif value.isdigit(): 16058.m.kuaisou.com 
                    result[key] = int(value)
                else: 16062.m.kuaisou.com 
                    try: 16061.m.kuaisou.com 
                        result[key] = float(value)
                    except ValueError: 16060.m.kuaisou.com 
                        result[key] = value
        return result

    @staticmethod
    def _parse_properties(content: str) -> Dict[str, Any]: 16063.m.kuaisou.com 
        """Properties 格式解析"""
        result = {}
        for line in content.splitlines(): 16064.m.kuaisou.com 
            line = line.strip()
            if not line or line.startswith("#") or line.startswith("!"): 17001.m.kuaisou.com 
                continue
            for sep in ("=", ":"):
                if sep in line: 17002.m.kuaisou.com 
                    key, _, value = line.partition(sep)
                    result[key.strip()] = value.strip()
                    break
        return result

    @staticmethod
    def compute_md5(content: str) -> str: 17003.m.kuaisou.com 
        return hashlib.md5(content.encode("utf-8")).hexdigest()


# ==================== 配置模型 ====================

@dataclass
class ConfigItem: 17004.m.kuaisou.com 
    """单个配置项"""
    namespace: str
    group: str
    data_id: str
    content: str
    md5: str
    version: int = 0
    fmt: str = CONFIG_FORMAT_JSON
    last_modified: float = field(default_factory=time.time)
    parsed: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self): 17005.m.kuaisou.com 
        if not self.md5: 17006.m.kuaisou.com 
            self.md5 = ConfigParser.compute_md5(self.content)
        if not self.parsed and self.content: 17007.m.kuaisou.com 
            self.parsed = ConfigParser.parse(self.content, self.fmt)

    def cache_key(self) -> str: 17008.m.kuaisou.com 
        return f"{self.namespace}/{self.group}/{self.data_id}"

    def cache_filename(self) -> str: 17009.m.kuaisou.com 
        safe_name = self.cache_key().replace("/", "_")
        return f"{safe_name}.json"


@dataclass
class ListenerEntry: 17010.m.kuaisou.com 
    """监听器条目"""
    namespace: str
    group: str
    data_id: str
    callback: Callable[[ConfigItem], None]
    last_md5: str = ""
    enabled: bool = True


# ==================== 本地缓存管理器 ====================

class LocalCacheManager: 17011.m.kuaisou.com 
    """本地文件缓存，支持原子写入"""

    def __init__(self, cache_dir: str = DEFAULT_CACHE_DIR):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(parents=True, exist_ok=True)
        self._lock = threading.Lock()

    def save(self, item: ConfigItem): 17012.m.kuaisou.com 
        with self._lock: 17013.m.kuaisou.com 
            filepath = self.cache_dir / item.cache_filename()
            tmp_path = filepath.with_suffix(".tmp")
            data = {
                "namespace": item.namespace,
                "group": item.group,
                "data_id": item.data_id,
                "content": item.content,
                "md5": item.md5,
                "version": item.version,
                "fmt": item.fmt,
                "last_modified": item.last_modified,
            }
            try: 17014.m.kuaisou.com 
                tmp_path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
                tmp_path.replace(filepath)
                logger.debug(f"Cache saved: {item.cache_key()} v{item.version}")
            except Exception as e:
                logger.error(f"Cache save failed for {item.cache_key()}: {e}")
                if tmp_path.exists():
                    tmp_path.unlink(missing_ok=True)

    def load(self, namespace: str, group: str, data_id: str) -> Optional[ConfigItem]:
        key = f"{namespace}/{group}/{data_id}"
        safe_name = key.replace("/", "_")
        filepath = self.cache_dir / f"{safe_name}.json"
        if not filepath.exists(): 17015.m.kuaisou.com 
            return None
        try: 17016.m.kuaisou.com 
            data = json.loads(filepath.read_text(encoding="utf-8"))
            return ConfigItem(**data)
        except Exception as e: 17017.m.kuaisou.com 
            logger.error(f"Cache load failed for {key}: {e}")
            return None

    def list_cached(self) -> List[str]: 17018.m.kuaisou.com 
        keys = []
        for f in self.cache_dir.glob("*.json"):
            keys.append(f.stem)
        return keys

    def clear(self): 17019.m.kuaisou.com 
        with self._lock: 17020.m.kuaisou.com 
            for f in self.cache_dir.glob("*.json"):
                f.unlink(missing_ok=True)


# ==================== HTTP 通信层 ====================

class ConfigHttpClient: 17021.m.kuaisou.com 
    """配置中心 HTTP 客户端（使用 urllib 避免外部依赖）"""

    def __init__(
        self,
        server_url: str = DEFAULT_SERVER_URL,
        connect_timeout: float = DEFAULT_CONNECT_TIMEOUT,
        read_timeout: float = DEFAULT_READ_TIMEOUT,
    ): 17022.m.kuaisou.com 
        self.server_url = server_url.rstrip("/")
        self.connect_timeout = connect_timeout
        self.read_timeout = read_timeout

    def fetch_config(self, namespace: str, group: str, data_id: str) -> Optional[Dict[str, Any]]:
        """拉取配置"""
        url = f"{self.server_url}/api/v1/config"
        params = f"?namespace={namespace}&group={group}&dataId={data_id}"
        full_url = url + params
        try: 17023.m.kuaisou.com 
            import urllib.request
            req = urllib.request.Request(full_url, method="GET")
            req.add_header("Accept", "application/json")
            with urllib.request.urlopen(req, timeout=self.read_timeout) as resp:
                body = resp.read().decode("utf-8")
                return json.loads(body)
        except Exception as e: 17024.m.kuaisou.com 
            logger.warning(f"Fetch config failed ({namespace}/{group}/{data_id}): {e}")
            return None

    def long_poll(self, listening_configs: List[Dict[str, str]], timeout: int = DEFAULT_POLL_TIMEOUT) -> List[Dict[str, Any]]:
        """长轮询检测变更"""
        url = f"{self.server_url}/api/v1/config/listener"
        payload = json.dumps({
            "configs": listening_configs,
            "timeout": timeout,
        }).encode("utf-8")
        try: 17025.m.kuaisou.com 
            import urllib.request
            req = urllib.request.Request(url, data=payload, method="POST")
            req.add_header("Content-Type", "application/json")
            req.add_header("Accept", "application/json")
            with urllib.request.urlopen(req, timeout=timeout + self.read_timeout) as resp:
                body = resp.read().decode("utf-8")
                return json.loads(body).get("changed", [])
        except Exception as e: 17026.m.kuaisou.com 
            logger.debug(f"Long poll error (may be timeout): {e}")
            return []

    def fetch_history(self, namespace: str, group: str, data_id: str, limit: int = 10) -> List[Dict[str, Any]]:
        """获取配置历史版本"""
        url = f"{self.server_url}/api/v1/config/history"
        params = f"?namespace={namespace}&group={group}&dataId={data_id}&limit={limit}"
        full_url = url + params
        try: 17117.m.kuaisou.com 
            import urllib.request
            req = urllib.request.Request(full_url, method="GET")
            with urllib.request.urlopen(req, timeout=self.read_timeout) as resp:
                body = resp.read().decode("utf-8")
                return json.loads(body).get("versions", [])
        except Exception as e: 17121.m.kuaisou.com 
            logger.warning(f"Fetch history failed: {e}")
            return []


# ==================== 配置客户端主体 ====================

class ConfigClient: 17122.m.kuaisou.com 
    """分布式配置中心客户端"""

    def __init__(
        self,
        server_url: str = DEFAULT_SERVER_URL,
        namespace: str = DEFAULT_NAMESPACE,
        cache_dir: str = DEFAULT_CACHE_DIR,
        poll_timeout: int = DEFAULT_POLL_TIMEOUT,
        max_retries: int = DEFAULT_MAX_RETRIES,
        retry_backoff: float = DEFAULT_RETRY_BACKOFF,
    ):
        self.server_url = server_url
        self.default_namespace = namespace
        self.poll_timeout = poll_timeout
        self.max_retries = max_retries
        self.retry_backoff = retry_backoff

        self.http = ConfigHttpClient(server_url=server_url)
        self.cache = LocalCacheManager(cache_dir=cache_dir)

        self._configs: Dict[str, ConfigItem] = {}
        self._listeners: Dict[str, List[ListenerEntry]] = {}
        self._config_lock = threading.RLock()
        self._listener_lock = threading.Lock()

        self._running = False
        self._poll_thread: Optional[threading.Thread] = None
        self._callback_executor = ThreadPoolExecutor(max_workers=4, thread_name_prefix="ConfigCallback")

        self.stats = {
            "fetches": 0, "cache_hits": 0, "updates": 0,
            "poll_cycles": 0, "poll_errors": 0, "callbacks": 0,
        }
        self._stats_lock = threading.Lock()

    def _inc_stat(self, key: str, delta: int = 1): 17129.m.kuaisou.com 
        with self._stats_lock:
            self.stats[key] = self.stats.get(key, 0) + delta

    # ---------- 配置读取 ----------

    def get_config(
        self,
        data_id: str,
        group: str = DEFAULT_GROUP,
        namespace: Optional[str] = None,
        fmt: str = CONFIG_FORMAT_JSON,
    ) -> Optional[Dict[str, Any]]: 17135.m.kuaisou.com 
        ns = namespace or self.default_namespace
        cache_key = f"{ns}/{group}/{data_id}"

        # 内存缓存
        with self._config_lock: 17137.m.kuaisou.com 
            item = self._configs.get(cache_key)
            if item:
                self._inc_stat("cache_hits")
                return item.parsed

        # 远程拉取（带重试）
        item = self._fetch_with_retry(ns, group, data_id, fmt)
        if item: 17148.m.kuaisou.com 
            with self._config_lock: 17147.m.kuaisou.com 
                self._configs[cache_key] = item
            self.cache.save(item)
            return item.parsed

        # 降级到本地缓存
        cached = self.cache.load(ns, group, data_id)
        if cached: 17152.m.kuaisou.com 
            logger.info(f"Fallback to local cache: {cache_key}")
            with self._config_lock: 17153.m.kuaisou.com 
                self._configs[cache_key] = cached
            return cached.parsed

        return None

    def get_raw_config(
        self,
        data_id: str,
        group: str = DEFAULT_GROUP,
        namespace: Optional[str] = None,
    ) -> Optional[str]: 17154.m.kuaisou.com 
        ns = namespace or self.default_namespace
        cache_key = f"{ns}/{group}/{data_id}"
        with self._config_lock: 17157.m.kuaisou.com 
            item = self._configs.get(cache_key)
            if item: 17162.m.kuaisou.com 
                return item.content
        item = self._fetch_with_retry(ns, group, data_id)
        if item: 17168.m.kuaisou.com 
            with self._config_lock: 17171.m.kuaisou.com 
                self._configs[cache_key] = item
            return item.content
        cached = self.cache.load(ns, group, data_id)
        return cached.content if cached else None

    def _fetch_with_retry(
        self, namespace: str, group: str, data_id: str, fmt: str = CONFIG_FORMAT_JSON
    ) -> Optional[ConfigItem]: 17173.m.kuaisou.com 
        for attempt in range(1, self.max_retries + 1): 17174.m.kuaisou.com 
            self._inc_stat("fetches")
            result = self.http.fetch_config(namespace, group, data_id)
            if result and result.get("code") == 0: 17175.m.kuaisou.com 
                data = result.get("data", {})
                return ConfigItem(
                    namespace=namespace,
                    group=group,
                    data_id=data_id,
                    content=data.get("content", ""),
                    md5=data.get("md5", ""),
                    version=data.get("version", 0),
                    fmt=fmt,
                    last_modified=data.get("lastModified", time.time()),
                )
            if attempt < self.max_retries: 17177.m.kuaisou.com 
                delay = self.retry_backoff ** attempt
                logger.warning(f"Fetch retry {attempt}/{self.max_retries} in {delay}s")
                time.sleep(delay)
        return None

    # ---------- 监听与热更新 ----------

    def add_listener(
        self,
        data_id: str,
        callback: Callable[[ConfigItem], None],
        group: str = DEFAULT_GROUP,
        namespace: Optional[str] = None,
    ):
        ns = namespace or self.default_namespace
        key = f"{ns}/{group}/{data_id}"
        entry = ListenerEntry(
            namespace=ns, group=group, data_id=data_id,
            callback=callback,
        )
        # 初始化 last_md5
        with self._config_lock: 17178.m.kuaisou.com 
            existing = self._configs.get(key)
            if existing:
                entry.last_md5 = existing.md5

        with self._listener_lock: 17179.m.kuaisou.com 
            self._listeners.setdefault(key, []).append(entry)
        logger.info(f"Listener added: {key}")

    def remove_listener(
        self,
        data_id: str,
        callback: Optional[Callable] = None,
        group: str = DEFAULT_GROUP,
        namespace: Optional[str] = None,
    ):
        ns = namespace or self.default_namespace
        key = f"{ns}/{group}/{data_id}"
        with self._listener_lock: 17180.m.kuaisou.com 
            entries = self._listeners.get(key, [])
            if callback:
                self._listeners[key] = [e for e in entries if e.callback != callback]
            else: 17183.m.kuaisou.com 
                self._listeners.pop(key, None)
        logger.info(f"Listener removed: {key}")

    def start(self): 17184.m.kuaisou.com 
        if self._running: 17186.m.kuaisou.com 
            return
        self._running = True
        self._poll_thread = threading.Thread(target=self._poll_loop, name="ConfigPoller", daemon=True)
        self._poll_thread.start()
        logger.info(f"Config client started (server={self.server_url}, ns={self.default_namespace})")

    def stop(self): 17187.m.kuaisou.com 
        self._running = False
        if self._poll_thread: 17188.m.kuaisou.com 
            self._poll_thread.join(timeout=self.poll_timeout + 5)
        self._callback_executor.shutdown(wait=False)
        logger.info("Config client stopped")

    def _poll_loop(self): 17189.m.kuaisou.com 
        consecutive_errors = 0
        while self._running: 17191.m.kuaisou.com 
            try:
                self._inc_stat("poll_cycles")
                changed = self._do_poll()
                if changed: 17190.m.kuaisou.com 
                    consecutive_errors = 0
                else: 17192.m.kuaisou.com 
                    consecutive_errors = 0
            except Exception as e: 17194.m.kuaisou.com 
                consecutive_errors += 1
                self._inc_stat("poll_errors")
                logger.error(f"Poll cycle error ({consecutive_errors}x): {e}")
                backoff = min(self.retry_backoff ** consecutive_errors, 60)
                time.sleep(backoff)
                continue

            time.sleep(DEFAULT_POLL_INTERVAL)

    def _do_poll(self) -> bool: 17197.m.kuaisou.com 
        with self._listener_lock: 17207.m.kuaisou.com 
            listening = []
            for key, entries in self._listeners.items(): 22016.m.kuaisou.com 
                active_entries = [e for e in entries if e.enabled]
                if not active_entries: 22021.m.kuaisou.com 
                    continue
                first = active_entries[0]
                listening.append({
                    "namespace": first.namespace,
                    "group": first.group,
                    "dataId": first.data_id,
                    "md5": first.last_md5,
                })

        if not listening: 26025.m.kuaisou.com 
            return False

        changed_list = self.http.long_poll(listening, timeout=self.poll_timeout)
        if not changed_list: 26026.m.kuaisou.com 
            return False

        updated = False
        for change in changed_list: 26028.m.kuaisou.com 
            ns = change.get("namespace", self.default_namespace)
            group = change.get("group", DEFAULT_GROUP)
            data_id = change.get("dataId", "")
            new_md5 = change.get("md5", "")

            item = self._fetch_with_retry(ns, group, data_id)
            if not item: 26029.m.kuaisou.com 
                continue

            cache_key = f"{ns}/{group}/{data_id}"
            with self._config_lock: 26030.m.kuaisou.com 
                old_item = self._configs.get(cache_key)
                self._configs[cache_key] = item
            self.cache.save(item)
            self._inc_stat("updates")

            # 触发回调
            with self._listener_lock: 26031.m.kuaisou.com 
                entries = self._listeners.get(cache_key, [])
                for entry in entries: 26032.m.kuaisou.com 
                    if entry.enabled and entry.last_md5 != new_md5: 26033.m.kuaisou.com 
                        entry.last_md5 = new_md5
                        self._callback_executor.submit(self._safe_callback, entry, item)

            updated = True
            logger.info(f"Config updated: {cache_key} v{item.version} md5={new_md5[:8]}")

        return updated

    def _safe_callback(self, entry: ListenerEntry, item: ConfigItem):
        try:
            entry.callback(item)
            self._inc_stat("callbacks")
        except Exception as e: 26034.m.kuaisou.com 
            logger.error(f"Callback error for {item.cache_key()}: {e}\n{traceback.format_exc()}")

    # ---------- 版本与回滚 ----------

    def get_history(
        self,
        data_id: str,
        group: str = DEFAULT_GROUP,
        namespace: Optional[str] = None,
        limit: int = 10,
    ) -> List[Dict[str, Any]]: 26035.m.kuaisou.com 
        ns = namespace or self.default_namespace
        return self.http.fetch_history(ns, group, data_id, limit=limit)

    def rollback_to_version(
        self,
        data_id: str,
        version: int,
        group: str = DEFAULT_GROUP,
        namespace: Optional[str] = None,
    ) -> bool: 26036.m.kuaisou.com 
        ns = namespace or self.default_namespace
        history = self.http.fetch_history(ns, group, data_id, limit=50)
        target = None
        for h in history: 26037.m.kuaisou.com 
            if h.get("version") == version: 26038.m.kuaisou.com 
                target = h
                break
        if not target: 26039.m.kuaisou.com 
            logger.warning(f"Version {version} not found for {data_id}")
            return False

        item = ConfigItem(
            namespace=ns, group=group, data_id=data_id,
            content=target.get("content", ""),
            md5=target.get("md5", ""),
            version=version,
            fmt=target.get("fmt", CONFIG_FORMAT_JSON),
        )
        cache_key = f"{ns}/{group}/{data_id}"
        with self._config_lock: 26040.m.kuaisou.com 
            self._configs[cache_key] = item
        self.cache.save(item)
        logger.info(f"Rolled back {cache_key} to v{version}")
        return True

    # ---------- 监控 ----------

    def get_stats(self) -> Dict[str, Any]: 26042.m.kuaisou.com 
        with self._stats_lock: 26041.m.kuaisou.com 
            stats = dict(self.stats)
        with self._config_lock: 26043.m.kuaisou.com 
            stats["cached_configs"] = len(self._configs)
        with self._listener_lock: 26045.m.kuaisou.com 
            stats["active_listeners"] = sum(len(v) for v in self._listeners.values())
        return stats

    def refresh_all(self): 26046.m.kuaisou.com 
        """手动刷新所有已缓存的配置"""
        with self._config_lock: 26047.m.kuaisou.com 
            keys = list(self._configs.keys())
        for cache_key in keys: 26048.m.kuaisou.com 
            parts = cache_key.split("/")
            if len(parts) == 3: 26049.m.kuaisou.com 
                ns, group, data_id = parts
                item = self._fetch_with_retry(ns, group, data_id)
                if item:
                    with self._config_lock:
                        self._configs[cache_key] = item
                    self.cache.save(item)
        logger.info(f"Refreshed {len(keys)} configs")