3d8ee1c2ed
- config: GatewayConfig.certs_from_env() reads only CERT_BUCKET/CERT_CACHE_DIR; IngestorManager uses it, so a malformed write/gRPC env var (GRPC_PORT, WRITE_MIN_PAUSE_S, IDLE_CONN_MAX_S) can no longer stop the reads. - certs: CertProvider raises CertificateNotConfiguredError with an actionable message when no platform_certificates document exists; downloads once under an asyncio.Lock, caches the paths, writes atomically (tmp + os.replace); invalidate() forces a fresh download. - pool: server_id normalized to str (1 and '1' share a session); warns when a get() asks for other url/server_uri/use_security than the pooled client was opened with (client reused, the slot config wins); invalidates the cert cache when a secured connect fails. - SubscriptionEngine: tracks nodes per subscription and forgets them on unsubscribe (no more cycle alarms / tag count for removed tags), keeping addresses still monitored by another slot. - ServerReader.shutdown(evict) / IngestorManager.remove_server(evict): a server that only left this pod's slots keeps the shared connection for NATS writes/gRPC; lost servers and config changes still evict. - IngestorManager warns once per server when a slot still sets server_cert_path (server certificate pinning is no longer applied). - slots: CONNECTION_FIELDS exported and reused by ServerReader. Accepted, not changed: read metrics label server_name with the server_id (same as the write path, which only knows the id); a malformed slot is notified each lease attempt (alerts are deduplicated by notification_id). Tests: 236 unit tests pass (+18); ruff clean; local e2e against an asyncua server incl. slot moved (connection kept and reused by the write path) and lost-server eviction/re-initialization. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
48 lines
2.0 KiB
Python
48 lines
2.0 KiB
Python
"""Parse the slot written by the Orchestrator to Redis into the slot v2 structure (contract C1).
|
|
|
|
Both shapes are accepted so the gateway and the Orchestrator can be deployed in
|
|
any order (SIENTIAPDE-1974):
|
|
|
|
- v2: each tag carries `names` (one entry per collection / tag name / model);
|
|
- v1 (legacy): each tag carries `tag_name` + `topics`, and the server carries the
|
|
file-based `cert_path`. v1 is converted to v2: one `names` entry per topic, and
|
|
`use_security` is true when the server had a certificate path (the certificate
|
|
itself now always comes from MinIO via CertProvider, never from that path).
|
|
"""
|
|
|
|
SERVER_FIELDS = ('server_id', 'url', 'server_uri', 'use_security', 'subscription_period_ms')
|
|
# What a connection/subscription needs (the pool keys clients by server_id itself).
|
|
CONNECTION_FIELDS = tuple(f for f in SERVER_FIELDS if f != 'server_id')
|
|
|
|
|
|
def _parse_names(address: str, tag: dict) -> list[dict]:
|
|
names = tag.get('names')
|
|
if names:
|
|
return names
|
|
topics = tag.get('topics')
|
|
if topics and tag.get('tag_name'):
|
|
return [{'collection': topic, 'tag_name': tag['tag_name']} for topic in topics]
|
|
raise ValueError(f'tag {address!r} has no names')
|
|
|
|
|
|
def _parse_tag(address: str, tag: dict) -> dict:
|
|
return {
|
|
'frequency': tag['frequency'],
|
|
'names': _parse_names(address, tag),
|
|
'active_read': tag.get('active_read', {'enabled': False}),
|
|
}
|
|
|
|
|
|
def _parse_server(name: str, server: dict) -> dict:
|
|
parsed = {field: server[field] for field in SERVER_FIELDS if field in server}
|
|
parsed['name'] = server.get('name', name)
|
|
if 'use_security' not in parsed:
|
|
parsed['use_security'] = bool(server.get('cert_path'))
|
|
parsed['tags'] = {addr: _parse_tag(addr, tag) for addr, tag in server['tags'].items()}
|
|
return parsed
|
|
|
|
|
|
def parse_slot(raw: dict) -> dict:
|
|
"""Normalize a raw slot (v1 or v2) into `{server_name: {..., tags: {addr: tag}}}`."""
|
|
return {name: _parse_server(name, server) for name, server in raw.items()}
|