团队的微服务数量突破50个后，情况开始失控。一次边缘服务的超时，通过我们谁也无法完整画出的调用链，最终导致了核心交易链路的雪崩。复盘会上，我们在白板上花了两个小时，依赖各个核心开发的回忆，才勉强拼凑出故障发生时的服务依赖拓扑。这个过程低效且极易出错。依赖人工维护的架构图早已过时，我们迫切需要一个能反映真实部署情况的、自动化的服务拓扑可视化系统。
初步构想很简单：一个定时任务，扫描我们的基础设施，识别出所有服务实例及其依赖关系，将这些关系存入一个专门的数据库，最后通过一个Web界面展示出来。真正的挑战在于技术选型。我们需要一个能准确反映基础设施现状的数据源，一个能高效存储和查询拓扑关系的模型，一个能清晰展示复杂网络的渲染引擎，以及一个轻量级的前端框架来呈现交互。
经过一番评估，最终的技术栈确定为：
- 数据源: DigitalOcean API。我们所有的服务都部署在DigitalOcean的Droplets上，通过Tags来管理服务身份和依赖关系。这是我们最直接、最准确的“事实来源”。
- 图存储: Neo4j。服务拓扑本质上是一个图（Graph），节点是服务，边是依赖关系。使用图数据库来建模远比关系型数据库或文档数据库来得自然和高效。Cypher查询语言在处理多级依赖、环路检测等场景下表现极其出色。
- 后端渲染: Python + Matplotlib + NetworkX。这是一个非主流但经过深思熟虑的选择。为什么不在前端用D3.js或vis.js？原因有二：第一，我们不希望将复杂的图布局计算（这在服务数量多时非常耗费CPU）下放到用户的浏览器中，这会严重影响体验。第二，团队内的Python技术栈非常成熟，利用NetworkX强大的图算法库进行布局计算，再由Matplotlib生成高质量的静态拓扑图，将其作为一张图片传给前端，是一种成本极低的“后端渲染”方案。前端只负责展示和简单的交互。
- 前端交互: Solid.js。我们只需要一个轻量、高性能的UI库来展示服务列表、渲染后端生成的拓扑图，并在用户点击图上节点时，拉取并展示该服务的详细信息。Solid.js的细粒度响应式模型和无虚拟DOM的设计，非常适合这种IO密集型而非重度状态管理的场景。
整个系统的架构因此变得清晰：
graph TD
subgraph "Python CronJob: Collector"
A[Scheduler] --> B{Collect Metadata};
B -- Droplet Tags --> C[DigitalOcean API];
B --> D{Parse Dependencies};
D --> E[Format Graph Data];
E -- Cypher Queries --> F[(Neo4j Database)];
end
subgraph "Python API Server: Flask"
G[Frontend UI] -- HTTP GET /api/topology/image --> H{Renderer};
H -- Read Graph --> F;
H -- NetworkX & Matplotlib --> I[Generate PNG Image];
I --> G;
G -- HTTP GET /api/service/details --> J{API Endpoint};
J -- Read Node Details --> F;
J -- JSON Response --> G;
end
subgraph "Browser"
K[Solid.js App] --> G;
end
第一步：依赖采集与图谱构建 (Collector Service)
Collector是整个流水线的起点，它是一个定时的Python脚本，负责连接DigitalOcean API，拉取所有Droplets的信息，并解析预先定义好的Tags来构建依赖关系。
我们约定了Tag规范：
-
service-name:<service-name>: 标记Droplet属于哪个服务。 -
service-deps:<dep1>,<dep2>: 标记该服务依赖于哪些其他服务。
这是Collector核心逻辑的实现。它需要处理认证、分页、错误，并将解析结果写入Neo4j。
collector/main.py:
import os
import time
import logging
from typing import Dict, List, Set
import digitalocean
from neo4j import GraphDatabase, exceptions
from dotenv import load_dotenv
# --- 配置与日志 ---
load_dotenv()
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
DO_TOKEN = os.getenv('DO_API_TOKEN')
NEO4J_URI = os.getenv('NEO4J_URI')
NEO4J_USER = os.getenv('NEO4J_USER')
NEO4J_PASSWORD = os.getenv('NEO4J_PASSWORD')
# --- 核心逻辑 ---
class TopologyCollector:
def __init__(self, do_token: str, neo4j_uri: str, neo4j_user: str, neo4j_password: str):
if not all([do_token, neo4j_uri, neo4j_user, neo4j_password]):
raise ValueError("Missing required environment variables for DigitalOcean or Neo4j.")
self.manager = digitalocean.Manager(token=do_token)
self.neo4j_driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password))
logging.info("Successfully connected to DigitalOcean and Neo4j.")
def close(self):
self.neo4j_driver.close()
logging.info("Neo4j connection closed.")
def fetch_services_from_droplets(self) -> Dict[str, List[str]]:
"""
从DigitalOcean API获取所有Droplets，并根据tags解析出服务及其依赖。
真实项目中，这里需要处理分页。为简化示例，我们假设Droplet数量不多。
"""
services = {}
droplets = self.manager.get_all_droplets()
for droplet in droplets:
service_name = None
dependencies = []
# 这里的tag解析逻辑是关键，必须健壮
for tag in droplet.tags:
if tag.startswith('service-name:'):
service_name = tag.split(':', 1)[1].strip()
elif tag.startswith('service-deps:'):
deps_str = tag.split(':', 1)[1].strip()
if deps_str:
dependencies = [d.strip() for d in deps_str.split(',')]
if service_name:
if service_name not in services:
services[service_name] = []
# 合并所有同名服务实例的依赖项，并去重
services[service_name].extend(dependencies)
# 去重依赖
for service, deps in services.items():
services[service] = sorted(list(set(deps)))
logging.info(f"Fetched and parsed {len(services)} unique services from Droplets.")
return services
def update_neo4j_graph(self, services: Dict[str, List[str]]):
"""
将服务拓扑数据写入Neo4j。使用MERGE确保幂等性。
"""
all_service_names: Set[str] = set(services.keys())
for deps in services.values():
all_service_names.update(deps)
with self.neo4j_driver.session() as session:
# 1. 标记所有节点为'stale'，准备进行清理
session.write_transaction(self._mark_all_nodes_stale)
# 2. 批量创建或更新所有服务节点，并标记为'active'
session.write_transaction(self._create_or_update_nodes, list(all_service_names))
# 3. 批量创建关系
for service_name, dependencies in services.items():
if dependencies:
session.write_transaction(self._create_relationships, service_name, dependencies)
# 4. 删除所有仍然是'stale'状态的节点和它们的关系，这些是已经下线的服务
deleted_count = session.write_transaction(self._cleanup_stale_nodes)
logging.info(f"Cleaned up {deleted_count} stale services from the graph.")
@staticmethod
def _mark_all_nodes_stale(tx):
logging.info("Marking all existing nodes as stale...")
query = "MATCH (s:Service) SET s.status = 'stale'"
tx.run(query)
@staticmethod
def _create_or_update_nodes(tx, service_names: List[str]):
logging.info(f"Updating {len(service_names)} service nodes...")
# 使用UNWIND来批量处理，性能远高于单条执行
query = """
UNWIND $services AS service_name
MERGE (s:Service {name: service_name})
ON CREATE SET s.createdAt = timestamp(), s.status = 'active'
ON MATCH SET s.updatedAt = timestamp(), s.status = 'active'
"""
tx.run(query, services=service_names)
@staticmethod
def _create_relationships(tx, service_name: str, dependencies: List[str]):
logging.info(f"Updating relationships for service: {service_name}")
query = """
MATCH (source:Service {name: $source_name})
UNWIND $deps AS dep_name
MATCH (target:Service {name: dep_name})
MERGE (source)-[:DEPENDS_ON]->(target)
"""
tx.run(query, source_name=service_name, deps=dependencies)
@staticmethod
def _cleanup_stale_nodes(tx) -> int:
logging.info("Cleaning up stale nodes...")
query = """
MATCH (s:Service {status: 'stale'})
DETACH DELETE s
RETURN count(s)
"""
result = tx.run(query)
count = result.single()[0]
return count
if __name__ == "__main__":
collector = TopologyCollector(DO_TOKEN, NEO4J_URI, NEO4J_USER, NEO4J_PASSWORD)
try:
# 实际生产中，这将由cron或类似的调度器触发
logging.info("Starting topology collection cycle...")
service_data = collector.fetch_services_from_droplets()
if service_data:
collector.update_neo4j_graph(service_data)
logging.info("Topology collection cycle finished.")
except Exception as e:
logging.error(f"An error occurred during collection: {e}", exc_info=True)
finally:
collector.close()这里的坑在于，必须处理服务的生命周期。当一个服务下线后，它对应的Droplet会被销毁，下一次采集时就不会再出现。因此，更新Neo4j的逻辑不能只是简单地添加，而应该是一个“同步”过程。我采用了一个“标记-清扫”策略：先将数据库中所有节点标记为stale（陈旧），然后在本次采集中出现的服务，将其标记更新为active（活跃），最后删除所有仍然是stale状态的节点。这确保了图谱的准确性。
第二步：后端渲染与API服务 (Renderer & API)
这个服务使用Flask构建，提供两个核心接口：
-
/api/topology/image: 从Neo4j拉取全量拓扑数据，使用NetworkX和Matplotlib动态生成一张PNG图片并返回。 -
/api/service/details/{service_name}: 查询指定服务的详细信息（如依赖它的服务，以及它依赖的服务）。
api/server.py:
import io
import logging
from flask import Flask, Response, jsonify, request
from neo4j import GraphDatabase
import networkx as nx
import matplotlib
matplotlib.use('Agg') # 必须在pyplot导入前设置，用于非GUI环境
import matplotlib.pyplot as plt
import os
from dotenv import load_dotenv
load_dotenv()
# --- 配置 ---
NEO4J_URI = os.getenv('NEO4J_URI')
NEO4J_USER = os.getenv('NEO4J_USER')
NEO4J_PASSWORD = os.getenv('NEO4J_PASSWORD')
app = Flask(__name__)
driver = GraphDatabase.driver(NEO4J_URI, auth=(NEO4J_USER, NEO4J_PASSWORD))
logging.basicConfig(level=logging.INFO)
# --- 核心路由 ---
@app.route('/api/topology/image')
def get_topology_image():
"""
生成并返回拓扑图的PNG图像以及节点坐标。
"""
try:
with driver.session() as session:
# 1. 从Neo4j获取节点和边
nodes, edges = session.read_transaction(fetch_graph_data)
if not nodes:
return jsonify({"error": "No service data found"}), 404
# 2. 使用NetworkX创建图并计算布局
G = nx.DiGraph()
G.add_nodes_from(nodes)
G.add_edges_from(edges)
# spring_layout是常用的力导向布局算法，k值可以调整节点间距
pos = nx.spring_layout(G, k=0.9, iterations=50, seed=42)
# 3. 使用Matplotlib绘制
plt.figure(figsize=(20, 20), dpi=100)
nx.draw_networkx_nodes(G, pos, node_size=3000, node_color='#4E84FF')
nx.draw_networkx_edges(G, pos, node_size=3000, arrowstyle='->', arrowsize=20, edge_color='gray', width=1.5)
nx.draw_networkx_labels(G, pos, font_size=10, font_color='white', font_weight='bold')
plt.axis('off')
# 4. 将图像保存到内存中的BytesIO对象
img_io = io.BytesIO()
plt.savefig(img_io, format='png', bbox_inches='tight', pad_inches=0.1)
img_io.seek(0)
plt.close() # 必须关闭，否则会内存泄漏
# 返回PNG图像
return Response(img_io.getvalue(), mimetype='image/png')
except Exception as e:
logging.error(f"Failed to generate topology image: {e}", exc_info=True)
return jsonify({"error": "Internal server error"}), 500
@app.route('/api/services')
def get_service_list():
try:
with driver.session() as session:
result = session.run("MATCH (s:Service) RETURN s.name AS name ORDER BY name")
service_names = [record["name"] for record in result]
return jsonify(service_names)
except Exception as e:
logging.error(f"Failed to fetch service list: {e}", exc_info=True)
return jsonify({"error": "Failed to connect to database"}), 500
@app.route('/api/service/details/<service_name>')
def get_service_details(service_name: str):
"""
获取单个服务的详细依赖信息。
"""
try:
with driver.session() as session:
details = session.read_transaction(fetch_service_details, service_name)
if not details:
return jsonify({"error": "Service not found"}), 404
return jsonify(details)
except Exception as e:
logging.error(f"Failed to fetch details for {service_name}: {e}", exc_info=True)
return jsonify({"error": "Internal server error"}), 500
# --- Neo4j事务函数 ---
def fetch_graph_data(tx):
nodes_result = tx.run("MATCH (s:Service) RETURN s.name AS name")
nodes = [record["name"] for record in nodes_result]
edges_result = tx.run("MATCH (source:Service)-[:DEPENDS_ON]->(target:Service) RETURN source.name AS source, target.name AS target")
edges = [(record["source"], record["target"]) for record in edges_result]
return nodes, edges
def fetch_service_details(tx, service_name):
query = """
MATCH (s:Service {name: $name})
OPTIONAL MATCH (upstream)-[:DEPENDS_ON]->(s)
OPTIONAL MATCH (s)-[:DEPENDS_ON]->(downstream)
RETURN s.name as name,
collect(DISTINCT upstream.name) as consumed_by,
collect(DISTINCT downstream.name) as depends_on
"""
result = tx.run(query, name=service_name).single()
return dict(result) if result and result['name'] else None
if __name__ == '__main__':
# 在生产环境中，应使用Gunicorn或uWSGI等WSGI服务器
app.run(host='0.0.0.0', port=5001, debug=False)一个常见的错误是在Flask这样的Web服务器中直接使用matplotlib.pyplot.show()，这会试图打开一个GUI窗口并导致程序崩溃。正确的方式是使用matplotlib.use('Agg')后端，它允许我们将图像渲染到一个内存缓冲区，然后通过HTTP响应发送。同时，每次绘图后必须调用plt.close()，否则Matplotlib会持续占用内存，最终导致服务OOM。
第三步：前端交互界面 (Solid.js)
前端的任务相对纯粹：
- 从
/api/services获取所有服务的列表并展示。 - 将
/api/topology/image的URL作为<img>标签的src，展示拓扑图。 - 当用户点击服务列表中的某一项时，调用
/api/service/details/{service_name}，并在侧边栏展示其依赖关系。
frontend/src/App.jsx:
import { createSignal, createResource, For, Show } from 'solid-js';
import styles from './App.module.css';
// API服务的基地址，在生产环境中应通过环境变量配置
const API_BASE_URL = 'http://localhost:5001';
// --- API请求函数 ---
const fetchServices = async () => (await fetch(`${API_BASE_URL}/api/services`)).json();
const fetchServiceDetails = async (serviceName) => {
if (!serviceName) return null;
const response = await fetch(`${API_BASE_URL}/api/service/details/${serviceName}`);
if (!response.ok) throw new Error(`Failed to fetch details for ${serviceName}`);
return response.json();
};
function App() {
// --- 响应式状态 ---
const [services] = createResource(fetchServices);
const [selectedService, setSelectedService] = createSignal(null);
// 当selectedService变化时，自动重新请求服务详情
const [details] = createResource(selectedService, fetchServiceDetails);
return (
<div class={styles.App}>
<aside class={styles.sidebar}>
<h1>Service Topology</h1>
<Show when={!services.loading} fallback={<div>Loading services...</div>}>
<ul>
<For each={services()}>{(service, i) =>
<li
class={selectedService() === service ? styles.selected : ''}
onClick={() => setSelectedService(service)}
>
{service}
</li>
}</For>
</ul>
</Show>
<Show when={services.error}>
<div class={styles.error}>Failed to load services. Is the API server running?</div>
</Show>
</aside>
<main class={styles.mainContent}>
<div class={styles.graphContainer}>
{/* 这里的key={Date.now()}是一个技巧，用于在需要时强制刷新图片 */}
<img src={`${API_BASE_URL}/api/topology/image?t=${Date.now()}`} alt="Service Topology Graph" />
</div>
<Show when={selectedService()}>
<div class={styles.detailsPanel}>
<h2>Details: {selectedService()}</h2>
<Show when={!details.loading} fallback={<div>Loading details...</div>}>
<div class={styles.dependencySection}>
<h3>Consumed By ({details()?.consumed_by?.length || 0})</h3>
<ul>
<For each={details()?.consumed_by || []} fallback={<li>None</li>}>
{(dep) => <li>{dep}</li>}
</For>
</ul>
</div>
<div class={styles.dependencySection}>
<h3>Depends On ({details()?.depends_on?.length || 0})</h3>
<ul>
<For each={details()?.depends_on || []} fallback={<li>None</li>}>
{(dep) => <li>{dep}</li>}
</For>
</ul>
</div>
</Show>
<Show when={details.error}>
<div class={styles.error}>Could not load details.</div>
</Show>
</div>
</Show>
</main>
</div>
);
}
export default App;Solid.js的createResource在这里非常方便，它封装了异步数据获取的加载、错误和成功状态，让UI代码非常声明式。当selectedService这个Signal变化时，依赖它的details Resource会自动触发新的数据请求，UI随之更新，整个流程清晰且高效。
一个需要注意的细节是在图片URL后面加上一个时间戳查询参数 ?t=${Date.now()}。这可以防止浏览器缓存旧的拓扑图，确保每次刷新页面（或在未来增加刷新按钮时）都能获取到最新的后端渲染结果。
局限性与未来迭代
这套系统有效地解决了我们最初的痛点，但它远非完美。在真实项目中，当前的实现有几个明显的局限性：
- 依赖发现的脆弱性: 当前方案强依赖于开发者在DigitalOcean上正确设置Tags。任何人为的疏忽或错误都将导致拓扑图失真。这是一种“约定优于配置”的弱保证。
- 静态可视化: 虽然交互性有了，但拓扑图本身是静态图片。我们无法在前端实现拖拽、缩放、高亮调用链路等高级交互。这是一个为了简化实现而做出的权衡。
- 缺乏运行时信息: 系统只展示了服务间的静态依赖关系，但无法反映真实的流量情况、调用频率、延迟或错误率。它告诉我们“谁可以调用谁”，而不是“谁正在调用谁”。
未来的迭代路径也因此清晰起来：
- 增强依赖发现: 更可靠的方式是集成服务网格（如Istio, Linkerd）或APM工具（如OpenTelemetry）的遥测数据。通过分析真实的运行时流量来自动发现服务依赖，可以获得100%准确的拓扑，并摆脱对人工标记的依赖。
- 转向前端渲染: 对于更丰富的交互需求，最终还是需要将图数据（节点和边）直接发送给前端，由D3.js, vis.js或react-flow这类库进行客户端渲染。这需要对后端API进行改造，并可能需要引入WebSocket来实现拓扑的实时更新。
- 融合运行时指标: 将从Prometheus等监控系统收集的指标（QPS, 延迟, 错误率）附加到Neo4j的节点和关系上。这样，我们不仅能看到拓扑，还能在图上直观地看到每个服务和每次调用的健康状况，例如用边的颜色或粗细来表示流量大小或延迟高低，这将使它从一个架构文档工具演进为一个真正的线上诊断平台。
