128 lines
5.9 KiB
Python
128 lines
5.9 KiB
Python
"""Строит граф межсервисных вызовов прикладного слоя.
|
||
|
||
py scan_calls.py <имя-кластера> <выход.json>
|
||
|
||
Три источника, по убыванию надёжности:
|
||
svc — внутрикластерный адрес сервиса в манифестах вызывающего;
|
||
env — переменная окружения вида <APP>_URL / _HOST / _BASE_URL в манифестах;
|
||
doc — та же переменная, но найденная только в *ENDPOINTS.md / *CONFIGURATION.md.
|
||
|
||
Третий источник нужен потому, что значительная часть адресов приезжает из
|
||
k8s-секретов и в манифестах отсутствует — на технологическом слое такой же
|
||
приём вскрыл четверть связей.
|
||
"""
|
||
import re, json, pathlib, sys, io, collections
|
||
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
|
||
if len(sys.argv) != 3:
|
||
sys.exit(__doc__)
|
||
CLUSTER, OUT = sys.argv[1], pathlib.Path(sys.argv[2])
|
||
ROOT = pathlib.Path(__file__).resolve().parents[2]
|
||
|
||
# Считаем развёрнутыми все приложения репозитория — см. комментарий в scan_contour.py
|
||
ALL = sorted(d.name for d in (ROOT / "apps").iterdir() if d.is_dir())
|
||
apps = ALL
|
||
|
||
# namespace -> приложение. Совпадают везде, кроме алиасов других контуров.
|
||
ALIASES = {
|
||
"proc": "processing", "workflow": "processing", "sarex-processing": "processing",
|
||
"orchestrator": "cde", "rfi-prod": "rfi",
|
||
"transmittals": "transmittal", "transmittal-api-stage": "transmittal",
|
||
"sarex": "django", "sarex-django": "django",
|
||
"sarex-documentations": "documentations",
|
||
"sarex-faas": "faas", "sarex-workspaces": "workspaces",
|
||
"sarex-attachments": "attachments", "planning": "pm",
|
||
}
|
||
# namespace технологических компонентов — они уже отражены на технологическом слое
|
||
INFRA_NS = {"postgresql", "rabbitmq", "minio", "kafka", "signoz", "redis",
|
||
"vault", "zitadel", "camunda", "istio-system", "insert-db-documentation"}
|
||
|
||
# В адресах встречаются суффиксы контура: documentations-prod, bim-api, eav-stage
|
||
NS_SUFFIX = ("-prod", "-stage", "-preprod", "-test", "-api")
|
||
|
||
def resolve(ns):
|
||
if ns in INFRA_NS:
|
||
return None
|
||
if ns in ALL:
|
||
return ns
|
||
if ns in ALIASES:
|
||
return ALIASES[ns]
|
||
for suf in NS_SUFFIX:
|
||
if ns.endswith(suf) and ns[:-len(suf)] in ALL:
|
||
return ns[:-len(suf)]
|
||
return None
|
||
|
||
# Адрес внутри кластера: <сервис>.<namespace> с необязательным .svc.cluster.local
|
||
# и портом. Короткая форма (backend-svc.django:8000) встречается наравне с полной.
|
||
HOST = re.compile(r"https?://([a-z0-9][a-z0-9-]*)\.([a-z0-9][a-z0-9-]*)"
|
||
r"(?:\.svc(?:\.cluster\.local)?)?(?::\d+)?(?![a-z0-9.-])")
|
||
TLD = {"ru", "com", "tech", "io", "lan", "net", "org", "dev", "local", "svc"}
|
||
|
||
def envpat(app):
|
||
u = app.upper().replace("-", "_")
|
||
return re.compile(rf"\b{u}(?:_[A-Z0-9]+)*_(?:URL|HOST|BASE_URL|INTERNAL_URL|ENDPOINT)\b")
|
||
|
||
PATS = {a: envpat(a) for a in ALL}
|
||
|
||
# DJANGO_POSTGRES_HOST, ISSUES_DB_HOST — адрес базы данных сервиса, а не самого
|
||
# сервиса. Такая переменная означает доступ в чужую БД, а не вызов по HTTP.
|
||
INFRA_SEG = re.compile(r"_(?:DB|POSTGRES|PG|REDIS|KAFKA|RABBITMQ|AMQP|S3|SMTP"
|
||
r"|MONGO|CLICKHOUSE|VAULT|BROKER|CACHE)_")
|
||
|
||
def env_hit(pat, blob):
|
||
return any(not INFRA_SEG.search(m.group(0)) for m in pat.finditer(blob))
|
||
|
||
def read(paths):
|
||
return "\n".join(p.read_text(encoding="utf-8", errors="replace") for p in paths)
|
||
|
||
edges, unresolved = {}, collections.Counter()
|
||
for caller in apps:
|
||
d = ROOT / "apps" / caller
|
||
yaml_blob = read([f for sub in ("base", CLUSTER) if (d / sub).is_dir()
|
||
for f in (d / sub).rglob("*.yaml")])
|
||
doc_blob = read(sorted(d.glob("*ENDPOINTS*.md")) + sorted(d.glob("*CONFIGURATION*.md")))
|
||
|
||
for svc, ns in set(HOST.findall(yaml_blob)):
|
||
if ns in TLD:
|
||
continue
|
||
callee = resolve(ns)
|
||
if callee is None:
|
||
if ns not in INFRA_NS:
|
||
unresolved[f"{svc}.{ns}"] += 1
|
||
continue
|
||
if callee != caller:
|
||
edges[(caller, callee)] = "svc"
|
||
|
||
for callee, pat in PATS.items():
|
||
if callee == caller or (caller, callee) in edges:
|
||
continue
|
||
if env_hit(pat, yaml_blob):
|
||
edges[(caller, callee)] = "env"
|
||
elif env_hit(pat, doc_blob):
|
||
edges[(caller, callee)] = "doc"
|
||
|
||
inbound = collections.Counter(c for _, c in edges)
|
||
outbound = collections.Counter(c for c, _ in edges)
|
||
by_src = collections.Counter(edges.values())
|
||
|
||
print(f"приложений в контуре: {len(apps)} связей: {len(edges)}")
|
||
print("по источнику:", dict(by_src))
|
||
print()
|
||
print("самые вызываемые (входящие связи):")
|
||
for a, n in inbound.most_common(12):
|
||
print(f" {a:18s} {n:2d} исходящих: {outbound[a]}")
|
||
print()
|
||
print("самые зависимые (исходящие связи):")
|
||
for a, n in outbound.most_common(8):
|
||
print(f" {a:18s} {n:2d}")
|
||
isolated = [a for a in apps if not inbound[a] and not outbound[a]]
|
||
print(f"\nбез связей: {', '.join(isolated) if isolated else 'нет'}")
|
||
if unresolved:
|
||
print(f"\nне сопоставлены с приложением: {dict(unresolved.most_common(8))}")
|
||
|
||
OUT.write_text(json.dumps({
|
||
"apps": apps,
|
||
"edges": [{"from": c, "to": t, "src": s} for (c, t), s in sorted(edges.items())],
|
||
}, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
print(f"\nзаписано: {OUT}")
|