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>
59 lines
2.4 KiB
Python
59 lines
2.4 KiB
Python
"""Gateway-specific env config (NATS/gRPC/certs/write flags).
|
|
|
|
Separate from ingestor.py's inline os.getenv (Redis/Mongo/TTLs) — this module
|
|
covers only the new gateway wiring consumed by app.py in Task 18.
|
|
"""
|
|
|
|
from dataclasses import dataclass
|
|
from os import getenv
|
|
|
|
|
|
def _env_bool(name: str, default: bool) -> bool:
|
|
value = getenv(name)
|
|
if value is None:
|
|
return default
|
|
return value.strip().lower() in ('true', '1', 'yes')
|
|
|
|
|
|
@dataclass
|
|
class GatewayConfig:
|
|
nats_url: str = 'nats://localhost:4222'
|
|
nats_write_subject: str = 'opc.write'
|
|
nats_queue: str = 'gateway-writers'
|
|
grpc_enabled: bool = False
|
|
grpc_port: int = 50051
|
|
# Legacy frontend bucket the platform already provisions (SIENTIAPDE-1974 D7);
|
|
# the design's `sientia-dataops` bucket does not exist in the platform MinIO.
|
|
cert_bucket: str = 'streamlit-connectors'
|
|
cert_cache_dir: str = '/tmp/opc-certs' # noqa: S108 # nosec B108 - default is spec'd, overridable via env
|
|
write_timestamp: bool = True
|
|
write_quality: bool = True
|
|
write_min_pause_s: float = 0.0
|
|
idle_conn_max_s: float = 300.0
|
|
|
|
@classmethod
|
|
def certs_from_env(cls) -> 'GatewayConfig':
|
|
"""Only the certificate settings — what the read path needs. Plain strings,
|
|
no number parsing, so a malformed write/gRPC env var (GRPC_PORT,
|
|
WRITE_MIN_PAUSE_S, ...) can only disable the runtime, never the reads."""
|
|
return cls(
|
|
cert_bucket=getenv('CERT_BUCKET', cls.cert_bucket),
|
|
cert_cache_dir=getenv('CERT_CACHE_DIR', cls.cert_cache_dir),
|
|
)
|
|
|
|
@classmethod
|
|
def from_env(cls) -> 'GatewayConfig':
|
|
return cls(
|
|
nats_url=getenv('NATS_URL', cls.nats_url),
|
|
nats_write_subject=getenv('NATS_WRITE_SUBJECT', cls.nats_write_subject),
|
|
nats_queue=getenv('NATS_QUEUE', cls.nats_queue),
|
|
grpc_enabled=_env_bool('GRPC_ENABLED', cls.grpc_enabled),
|
|
grpc_port=int(getenv('GRPC_PORT', cls.grpc_port)),
|
|
cert_bucket=getenv('CERT_BUCKET', cls.cert_bucket),
|
|
cert_cache_dir=getenv('CERT_CACHE_DIR', cls.cert_cache_dir),
|
|
write_timestamp=_env_bool('WRITE_TIMESTAMP', cls.write_timestamp),
|
|
write_quality=_env_bool('WRITE_QUALITY', cls.write_quality),
|
|
write_min_pause_s=float(getenv('WRITE_MIN_PAUSE_S', cls.write_min_pause_s)),
|
|
idle_conn_max_s=float(getenv('IDLE_CONN_MAX_S', cls.idle_conn_max_s)),
|
|
)
|