anjia0532/discovery-syncer-python repository overview

⁠多端注册中心网关同步工具

支持从nacos(已实现)，eureka(已实现)等注册中心同步到apisix(已实现)和kong(已实现)等网关， 后续将支持自定义插件，支持用户自己用 python 实现支持类似携程阿波罗注册中心，etcd注册中心，consul注册中心等插件，以及spring gateway等网关插件的高扩展性

⁠快速开始

⁠通过docker运行

docker run anjia0532/discovery-syncer-python:v3.0.0

特别的，-c 支持配置远端http[s]的地址，比如读取静态资源的，比如读取nacos的 -c http://xxxxx/nacos/v1/cs/configs?tenant=public&group=DEFAULT_GROUP&dataId=discovery-syncer.yaml ,便于管理

运行时依赖（已简化）

自 v3 起已移除 funboost 与 Redis：任务执行改为进程内 ThreadPoolExecutor （app/runtime/taskbus.py），定时调度改为本地 APScheduler + MemoryJobStore （app/runtime/scheduler.py）。不再需要任何外部中间件，一个容器即可运行。

历史说明：v2.7.0 曾引入 Redis 解决内存增长与任务阻塞问题（v2.6.13 及以前无 Redis）。 现在通过「订阅事件去抖 + 5 分钟全量对账兜底」从根本上降低了队列压力， 消息队列与分布式锁都不再必要。

⁠配置文件

点击查看 config.yaml⁠

运行的时候不再需要配置 REDIS_* 环境变量，也无需启动 Redis。 安装依赖即可：pip install -r requirements.txt（显式依赖 APScheduler、nb_log）。

⁠Api接口

注意:

请勿将此服务暴露到公网，否则对于引发的一切安全事故概不负责。安全起见，所有接口统一都增加 SYNCER-API-KEY header 头校验，需要在配置文件 common.syncer-api-key 修改默认值， 长度最低为32位，需要同时包含大小写字母，数字，和特殊字符

⁠GET /show-memory?num=20 查看内存使用情况

num 是显示的条数，默认20条

⁠GET /-/reload 重新加载配置文件

加载成功返回OK

主要是 cicd 场景或者 k8s 的 configmap reload 场景使用

⁠GET /health 判断服务是否健康，可以配合k8s等容器服务的健康检查使用

返回值

{
  // 一共有几个enabled的同步任务(targets)
  "total": 2,
  // 正常在跑的有几个
  "running": 2,
  // 有几个超过 配置文件定义的maximum-interval-sec的检测时间没有运行的，失联的。
  "lost": 0,
  // 都在跑，状态是OK（http状态码是200），有在跑的，有失联的，状态是WARN（http状态码是200），全部失联，状态是DOWN(http状态码500)
  "status": "OK",
  // 哪些成功，哪些失败
  "details": [
    "syncer:a_task,is ok",
    "syncer:b-api,is ok"
  ],
  // 运行时长
  "uptime": "1m6s"
}

⁠PUT /discovery/{discovery-name}?alive_num=1 主动下线上线注册中心的服务,配合CI/CD发版业务用

discovery-name 是注册中心的名字，如果不存在，则返回 Not Found http status code 是404

alive_num 是执行上下线操作后，断言存活实例数量，小于等于0则不判断，不传默认1，不满足则报错，http status code 是500, content 类似 ('最少存活实例数1不满足，总实例数(含之前已下线数量)2，要下线实例数2，剩余在线实例数0',)

body入参

{
  // 检索哪个服务下的实例
  "serviceName": "",
  // 基于注册中心元数据还是基于实例ip来查找
  "type": "METADATA/IP",
  // 匹配的查询条件，支持正则
  "regexpStr": "",
  // 匹配的元数据key，如果是ip则不用填
  "metadataKey": "",
  // 匹配到的将状态改成上线还是下线
  "status": "UP/DOWN",
  // 其他没匹配的，状态是上线还是下线，ORIGIN保持不变
  "otherStatus": "UP/DOWN/ORIGIN",
  // 扩展参数
  "extData": {
  }
}

⁠GET /gateway-api-to-file/{gateway-name}?file=/tmp/apisix.yaml 读取网关admin api转换成文件用于备份或者db-less模式

gateway-name 是网关的名字，如果不存在，则返回 Not Found，http status code是404

如果服务报错，resp body 会返回空字符串，header 中的 syncer-err-msg 会返回具体原因，http status code 是500，参数 file 是可选的，如果不传，则默认/tmp/文件名 比如 /tmp/apisix.yaml，并返回文件路径

如果正常，resp body 会返回转换后的文本内容，syncer-file-location 会返回syncer服务端的路径( 一般是系统临时目录+文件名，例如/tmp/apisix.yaml)， http status code是200

注意

精力有限，目前仅实现了 apisix 的 admin api 转 yaml，kong 的未实现，有需要的，欢迎提PR贡献代码或者提issues 来反馈

⁠POST /migrate/{origin_gateway_name}/to/{target_gateway_name} 相同网关不同实例间数据迁移

origin_gateway_name 是源网关的名字，target_gateway_name 是目标网关的名字，如果不存在，则返回 Not Found http status code 是404

注意

精力有限，目前仅实现了 apisix 实例间数据迁移 （支持 apisix 2x->3x , apisix 3x->2x，apisix 2x->2x，apisix 3x->3x），kong 的未实现，有需要的，欢迎提PR贡献代码或者提issues 来反馈

⁠PUT /restore/{gateway-name} 将 db-less 文件还原到网关

gateway-name 是网关的名字，如果不存在，则返回 Not Found http status code 是404

body入参（以apisix为例）

# Auto generate by https://github.com/anjia0532/discovery-syncer-python, Don't Modify

# apisix 2.x modify conf/config.yaml https://apisix.apache.org/docs/apisix/2.15/stand-alone/
# apisix:
#  enable_admin: false
#  config_center: yaml

# apisix 3.x modify conf/config.yaml https://apisix.apache.org/docs/apisix/3.2/deployment-modes/#standalone
# deployment:
#  role: data_plane
#  role_data_plane:
#    config_provider: yaml

# save as conf/apisix.yaml

routes: [ ]
services: [ ]
upstreams: [ ]
plugins: [ ]

#END

注意

仅限同版本还原，不支持跨版本还原，如apisix 2.x 还原到 apisix 3.x，apisix 3.x 还原到 apisix 2.x。有需要跨版本的 精力有限，目前仅实现了 apisix 的 yaml 还原 apisix ，kong 的未实现，有需要的，欢迎提PR贡献代码或者提issues 来反馈

⁠定时备份和还原网关数据

⁠备份

job.json 格式（多集群多 kubeconfig）：

{
  "clusters": {
    "a": {
      "kubeconfig": "/root/.kube/config-a",
      "context": "ctx-a",
      "namespace": "syncer",
      "service": "syncer",
      "remote_port": 9797
    },
    "b": {
      "kubeconfig": "/root/.kube/config-b",
      "context": "ctx-b",
      "namespace": "syncer",
      "service": "syncer"
    }
  },
  "jobs": [
    {
      "name": "gateway-backup",
      "clusters": ["a", "b", "c", "d", "e"],
      "gateways": ["apisix1", "apisix2"],
      "base_name": "apisix.yaml",
      "syncer_api_key": "xxx",
      "timeout": 10,
      "ssl_verify": false
    }
  ]
}

多集群备份：顶层 clusters 定义每个集群的 kubeconfig / context / namespace / service， jobs 通过 clusters 列表引用集群名，一个 job 即可批量备份多个集群。 备份路径为 base_dir/<job>/<cluster>/<gateway>/<base_name>，例如：

syncer/gateway-backup/a/apisix1/apisix.yaml
syncer/gateway-backup/b/apisix1/apisix.yaml
...

单集群直连（兼容旧格式，无需 kubeconfig）：

{
  "apisix1": {
    "syncer": "http://localhost:9797",
    "gateways": ["apisix1"],
    "syncer_api_key": "xxx"
  }
}

完整多集群示例见 tools/backup-multi-cluster.json.example。

如果用 docker 的话

echo '{}' > ./job.json
docker run -e SYNC_JOB_JSON=job.json \
        -e BRANCH=main \
        -e REMOTE=origin \
        -e GIT_REPO='[email protected]:anjia0532/discovery-syncer-python-demo.git' \
        -e USER_EMAIL='[email protected]'  \
        -e BASE_DIR='syncer'  \
        -v $(pwd)/job.json:/opt/discovery-syncer/job.json  \
        -v $HOME/.kube:/root/.kube:ro \
        anjia0532/discovery-syncer-python-backup:v2.4.5

如果用命令行的话

wget https://raw.githubusercontent.com/anjia0532/discovery-syncer-python/master/tools/backup.py
pip install -i https://pypi.tuna.tsinghua.edu.cn/simple --no-cache-dir --upgrade GitPython httpx

# 安装 git 和 python3.11
echo '{}' > ./job.json
export SYNC_JOB_JSON=job.json
python3 backup.py

也支持命令行参数直接指定配置：

python3 backup.py --config job.json --git-repo [email protected]:xxx/backup.git --branch main

⁠还原

vi restore.py

python3 restore.py

⁠新增服务发现或者网关插件

⁠服务发现

修改 app.model.config.DiscoveryType⁠ 新增类型,比如

class DiscoveryType(Enum):
    NACOS = "nacos"
    EUREKA = "eureka"
    # 新增 redis
    REDIS = "redis"

创建 app/service/discovery/redis.py 文件

from typing import List

from app.model.syncer_model import Service, Instance, Registration
from app.service.discovery.discovery import Discovery
from core.lib.logger import for_service

logger = for_service(__name__)


class Redis(Discovery):

    def __init__(self, config):
        super().__init__(config)

    def get_all_service(self, config: dict, enabled_only: bool = True) -> List[Service]:
        pass

    def get_service_all_instances(self, service_name: str, ext_data: dict, enabled_only: bool = True) -> tuple[
        List[Instance], int]:
        pass

    def modify_registration(self, registration: Registration, instances: List[Instance]):
        pass

修改 config.yaml

discovery-servers:
  redis1:
    type: redis
    weight: 100
    prefix: 2
    host: "redis://localhost:6379"
    config:
      demo: demo
gateway-servers:
  apisix1:
    type: apisix
    admin-url: http://apisix-server:9080
    prefix: /apisix/admin/
    config:
      X-API-KEY: xxxxxxxx-xxxx-xxxxxxxxxxxx
      version: v3
targets:
  - discovery: redis
    gateway: apisix1
    enabled: false
    # .. 忽略其他部分

⁠网关

修改 app.model.config.GatewayType⁠ 新增类型,比如

class GatewayType(Enum):
    KONG = "kong"
    APISIX = "apisix"
    SPRING_GATEWAY = "spring_gateway"

创建 app/service/gateway/spring_gateway.py 文件

from typing import Tuple, List

from app.model.syncer_model import Instance
from app.service.gateway.gateway import Gateway


class SpringGateway(Gateway):
    def __init__(self, config):
        super().__init__(config)

    def get_service_all_instances(self, target: dict, upstream_name: str = None) -> List[Instance]:
        pass

    def sync_instances(self, target: dict, upstream_name: str, diff_ins: list, instances: list):
        pass

    def fetch_admin_api_to_file(self, file_name: str) -> Tuple[str, str]:
        pass

    async def migrate_to(self, target_gateway: 'Gateway'):
        pass

修改 config.yaml

discovery-servers:
  nacos1:
    type: nacos
    weight: 100
    prefix: /nacos/v1/
    host: "http://nacos-server:8858"
gateway-servers:
  spring_gateway1:
    type: spring_gateway
    admin-url: http://gateway-server:9080
    prefix: /
    config:
      demo: xxxxxxxx-xxxx-xxxxxxxxxxxx
targets:
  - discovery: nacos1
    gateway: spring_gateway1
    enabled: false
    # .. 忽略其他部分

⁠订阅机制与多版本适配

⁠从轮询到订阅

订阅层可插拔，按上游版本能力自动选择：

• Nacos 1.x — 原生 UDP 推送：本地起 UDP 接收端，周期性重发带 udpPort 的 GET /nacos/v1/ns/instance/list 完成订阅续约与新服务发现。
• Nacos 2.x — 官方订阅是 gRPC(9848)。服务端保留 UDP 兼容层，因此优先尝试 UDP； 若 grpc_stream 注入了真实双向流则走 gRPC；否则自动降级为快照 diff 增量轮询。
• Nacos 3.x — 不再兼容 v1/v2 HTTP OpenAPI，改用 /nacos/v3/admin/*（范围查询） 与 /nacos/v3/client/*（单服务），默认命名空间由 "" 变为 public， 且客户端 API 不支持「查询全部服务列表」。gRPC 是唯一订阅传输， 未注入 gRPC 流时自动降级为增量轮询 + 入站 webhook。
• Eureka — 无服务端 push，统一走快照 diff 增量轮询。
• 任意版本 — 可叠加 POST /-/notify 入站回调，由 CI/CD 或外部监听脚本主动推送。

⁠版本能力矩阵

版本判定顺序：显式配置优先（subscribe.version / discovery-servers.<n>.version / gateway-servers.<n>.version，旧的 config.version 继续兼容）→ 自动探测 （Nacos v3/v1 console/server/state + 9848 端口；APISIX 响应形态；Kong 根路径 version）。 探测失败一律回落到最保守的已知版本，不会让旧配置启动失败。

⁠配置兼容

旧 config.yaml 一行不改即可升级：所有新增字段都是可选的并带默认值。 fetch-interval / maximum-interval-sec / healthcheck / config.template / exclude-service / upstream-prefix 语义保留；订阅模式下 fetch-interval 转作 健康检查与增量轮询周期，全量对账另由 subscribe.reconcile-interval 控制（默认 5 分钟）。

需要完全灰度对比时可设置 subscribe.enabled: false 或 subscribe.mode: off， 行为与旧版定时全量轮询一致。完整可选项见 config-example.yaml 末尾注释块。

⁠本地运行时（去 funboost / Redis）

去掉的能力与补偿：

• 任务持久化（进程重启丢消息）：由默认 @every 5m 的全量对账兜底补齐。
• 死信可观测性：GET /-/subscriptions 的 _runtime.taskbus 暴露 submitted/succeeded/failed/retries；且同步失败时不刷新心跳， /health 可通过 maximum-interval-sec 感知故障。
• function_timeout 强杀：Python 线程不可中断，改由各 HTTP 客户端的 10s timeout 保障。
• 分布式控频 / 多副本互斥：单副本运行，无需该能力。

/-/reload 现在显式向本地任务总线提交 reload；Settings.load_config() 只负责 解析配置，不再隐式触发重载，消除了原先的重复触发与潜在重入。

⁠新增运维接口

所有接口仍由 SYNCER-API-KEY 中间件保护；subscribe.notify-token 可作为回调额外令牌。

⁠多版本镜像验证

tests/integration/docker-compose.matrix.yml 提供完整版本矩阵： Nacos 1.4.6 / 2.5.x / 3.x、APISIX 2.15 / 3.x、Kong 2.8 / 3.7。

# 在 WSL / Linux 且可运行 docker 的环境执行
cd tests/integration
./run_matrix.sh up        # 拉起矩阵并等待健康检查
./run_matrix.sh probe     # 打印版本探测结果与 /-/versions 对照
./run_matrix.sh test      # 逐个版本跑注册实例 -> 变更 -> 校验网关 upstream
./run_matrix.sh down      # 清理

验证项包括：版本探测与镜像大版本一致、APISIX 2.x node.nodes[] vs 3.x list/total 的响应归一化、upstream PUT + PATCH /nodes 实际生效，以及 Kong 3.x 重复 POST /targets 返回 409 而 PUT 幂等成功（2.x 则 POST 可连续更新）。

无 Docker 环境的等价验证（本地协议级 mock，覆盖完全相同的检查项）：

python3 tests/integration/run_local_matrix.py     # 需 pip install httpx

⁠发版流程

⁠工作流总览

⁠发版步骤

1. 确保工作区在 master 分支且已合并所有待发布改动。
2. 在 GitHub Actions 页面选择 Release → Run workflow，输入版本号（如 3.0.0）。
3. 工作流自动执行：
   • test 作业：单测（48 项）+ 应用冒烟（含内存压力）+ Docker 多版本矩阵契约验证。
   • 测试通过后进入 release 作业：生成 CHANGELOG → 打 tag v{version} → 创建 GitHub Release → 推送镜像到 gcr.io 与 Docker Hub。
4. 镜像标签：{version}、latest、{sha} 三个。

⁠GCP OIDC 配置（gcr.io 推送）

release.yml 通过 OIDC workload identity 免密钥推送到 gcr.io。首次使用需在 GCP 完成一次性配置：

1. 在 GCP 创建 workload identity pool 与 provider（参考 GCP 官方文档⁠）。

2. 将 GitHub 仓库加入 provider 的 attribute mapping：

   https://token.actions.githubusercontent.com|repo:anjia0532/discovery-syncer-python
   

   Copy

3. 在 GitHub 仓库 Settings → Secrets and variables → Actions → Variables 添加：

   变量名	值
   GCP_PROJECT_ID	gcr.io 对应的 GCP 项目 ID
   GCP_WI_POOL_ID	workload identity pool ID
   GCP_WI_PROVIDER_ID	workload identity provider ID

4. Docker Hub 推送使用已有的 HUB_NAME / HUB_PWD secrets，无需额外配置。

⁠CHANGELOG 生成

scripts/gen_changelog.py 从上一个 v* tag 到 HEAD 的提交生成 Keep a Changelog 格式条目， 按 conventional commit 类型（feat / fix / refactor / perf / test / docs / build / chore）自动分组。 release.yml 在打 tag 前自动调用，也可本地手动运行：

python3 scripts/gen_changelog.py --version 3.0.0

⁠待优化点

1. 已解决 目前的同步任务是串行的，如果待同步的量比较大，或者同步时间窗口设置的特别小的情况下，会导致挤压

2. 已解决 不支持自定义同步插件，不利于自行扩展

3. 已解决 ~~同步机制目前是基于定时轮询，效率比较低，有待优化，比如增加缓存开关，上游注册中心与缓存比对没有差异的情况下，不去拉取/变更下