diff --git a/conf/app.py b/conf/app.py index 9cb5493..f700c5e 100644 --- a/conf/app.py +++ b/conf/app.py @@ -60,7 +60,7 @@ js = None # Configuration NATS_URL = os.getenv("NATS_URL", "nats://nats-server:4222") NATS_USER = os.getenv("NATS_USER", "admin") -NATS_PASSWORD = os.getenv("NATS_PASSWORD", "package@321#") +NATS_PASSWORD = os.getenv("NATS_PASSWORD", "") # Endpoint to NATS subject mapping ENDPOINT_ROUTES = { @@ -172,7 +172,7 @@ async def publish_request_to_nats(endpoint: str, data: Dict[str, Any], request_m subject = ENDPOINT_ROUTES.get(endpoint, "api.unknown") if not js: - raise HTTPException(status_code=503, detail="NATS by connected") + raise HTTPException(status_code=503, detail="NATS not connected") try: # Create a unique inbox for the reply diff --git a/conf/nginx-jupiter.conf b/conf/nginx-jupiter.conf index 5641d4a..e18ca15 100644 --- a/conf/nginx-jupiter.conf +++ b/conf/nginx-jupiter.conf @@ -12,7 +12,15 @@ http { # Proxy to Kubernetes NodePort (30822) # NodePorts are bound to 0.0.0.0 and are more reliable to access from host.docker.internal proxy_pass http://jupiter_k8s; - + + # Strip any CORS headers the backend may set itself, so we don't + # end up sending duplicate Access-Control-* headers (browsers + # reject a response that has more than one value for these). + proxy_hide_header 'Access-Control-Allow-Origin'; + proxy_hide_header 'Access-Control-Allow-Methods'; + proxy_hide_header 'Access-Control-Allow-Headers'; + proxy_hide_header 'Access-Control-Max-Age'; + proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; diff --git a/conf/worker.py b/conf/worker.py index 542f7a7..f03e075 100644 --- a/conf/worker.py +++ b/conf/worker.py @@ -12,6 +12,7 @@ import asyncio import aiohttp import nats from nats.errors import TimeoutError as NatsTimeoutError +from nats.js.api import ConsumerConfig from prometheus_client import Counter, Histogram, Gauge, start_http_server import signal import sys @@ -25,7 +26,7 @@ messages_in_flight = Gauge('worker_messages_in_flight', 'Messages currently bein # Configuration NATS_URL = os.getenv("NATS_URL", "nats://nats-server:4222") NATS_USER = os.getenv("NATS_USER", "admin") -NATS_PASSWORD = os.getenv("NATS_PASSWORD", "package@321#") +NATS_PASSWORD = os.getenv("NATS_PASSWORD", "") # Domain Configuration (Env Vars from Deployment) NATS_STREAM = os.getenv("NATS_STREAM", "ORDERS") @@ -35,6 +36,10 @@ FILTER_SUBJECTS = os.getenv("FILTER_SUBJECT", "api.v1.mob.orders.*").split(",") WORKER_CONCURRENCY = int(os.getenv("WORKER_CONCURRENCY", "10")) BASE_URL = os.getenv("EXTERNAL_BASE_URL", "https://jupiter.nearle.app") +# Must stay comfortably above the outbound HTTP timeout below (30s), otherwise +# JetStream can redeliver a message to another worker while this one is still +# waiting on a slow-but-legitimate response, causing a duplicate forward. +ACK_WAIT_SECONDS = int(os.getenv("ACK_WAIT_SECONDS", "60")) # Endpoint mapping is still useful for constructing the target URL ENDPOINT_MAPPING = { @@ -56,6 +61,7 @@ js = None session: aiohttp.ClientSession = None running = True semaphore = None # Initialized in main +active_tasks: set = set() # In-flight process_message() tasks, drained on shutdown def signal_handler(sig, frame): global running @@ -80,7 +86,10 @@ async def forward_to_external(endpoint: str, payload: dict, api_key: str = None, if api_key: headers["Authorization"] = f"Bearer {api_key}" - data_to_forward = payload.get("data", payload) + if "data" not in payload: + print(f"?? Message for {endpoint} missing 'data' field, dropping.") + return "DROP", None + data_to_forward = payload["data"] # Payload Normalization for logs if endpoint == "/live/api/v2/deliveries/createdeliverylog": @@ -166,7 +175,8 @@ async def process_message(msg): await msg.term() except Exception as e: print(f"? Critical Worker Error: {e}") - await msg.nak() + # Delay avoids a tight redelivery loop if this is a persistent bug + await msg.nak(delay=5) finally: messages_in_flight.dec() duration = asyncio.get_event_loop().time() - start_time @@ -210,8 +220,9 @@ async def main(): try: sub = await js.pull_subscribe( subject, - durable=durable_name, - stream=NATS_STREAM + durable=durable_name, + stream=NATS_STREAM, + config=ConsumerConfig(ack_wait=ACK_WAIT_SECONDS) ) subs.append(sub) except Exception as e: @@ -231,7 +242,9 @@ async def main(): # Batch size 10, short timeout to keep loop responsive msgs = await sub.fetch(10, timeout=0.5) for m in msgs: - asyncio.create_task(process_message(m)) + task = asyncio.create_task(process_message(m)) + active_tasks.add(task) + task.add_done_callback(active_tasks.discard) except NatsTimeoutError: pass except Exception as e: @@ -244,6 +257,9 @@ async def main(): except Exception as e: print(f"?? Fatal Error: {e}") finally: + if active_tasks: + print(f"?? Draining {len(active_tasks)} in-flight message(s) before shutdown...") + await asyncio.wait(active_tasks, timeout=25) if session: await session.close() if nc: diff --git a/docker-compose.yml b/docker-compose.yml index a13c105..988f5a4 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -106,6 +106,27 @@ services: - "traefik.http.routers.atlantis-api.tls.certresolver=letsencrypt" - "traefik.http.services.atlantis-api.loadbalancer.server.port=8205" + # Doormile API Proxy (doormile.com -> K8s NodePort 30830) + doormile-proxy: + image: nginx:alpine + container_name: doormile-proxy + restart: unless-stopped + ports: + - "8206:8206" + volumes: + - ./conf/nginx-doormile.conf:/etc/nginx/nginx.conf:ro + extra_hosts: + - "host.docker.internal:host-gateway" + networks: + - web + labels: + - "traefik.enable=true" + - "traefik.docker.network=web" + - "traefik.http.routers.doormile-api.rule=Host(`doormile.com`)" + - "traefik.http.routers.doormile-api.tls=true" + - "traefik.http.routers.doormile-api.tls.certresolver=letsencrypt" + - "traefik.http.services.doormile-api.loadbalancer.server.port=8206" + networks: web: external: true diff --git a/manifests/core/worker-script.yaml b/manifests/core/worker-script.yaml index 3d6eef2..bfd1dfb 100644 --- a/manifests/core/worker-script.yaml +++ b/manifests/core/worker-script.yaml @@ -19,6 +19,7 @@ data: import aiohttp import nats from nats.errors import TimeoutError as NatsTimeoutError + from nats.js.api import ConsumerConfig from prometheus_client import Counter, Histogram, Gauge, start_http_server import signal import sys @@ -32,17 +33,21 @@ data: # Configuration NATS_URL = os.getenv("NATS_URL", "nats://nats-server:4222") NATS_USER = os.getenv("NATS_USER", "admin") - NATS_PASSWORD = os.getenv("NATS_PASSWORD", "package@321#") + NATS_PASSWORD = os.getenv("NATS_PASSWORD", "") # Domain Configuration (Env Vars from Deployment) NATS_STREAM = os.getenv("NATS_STREAM", "ORDERS") NATS_CONSUMER = os.getenv("NATS_CONSUMER", "orders-worker") # Supports comma-separated subjects: "api.v1.orders.*,api.v2.orders.*" FILTER_SUBJECTS = os.getenv("FILTER_SUBJECT", "api.v1.mob.orders.*").split(",") - + WORKER_CONCURRENCY = int(os.getenv("WORKER_CONCURRENCY", "10")) BASE_URL = os.getenv("EXTERNAL_BASE_URL", "https://jupiter.nearle.app") - + # Must stay comfortably above the outbound HTTP timeout below (30s), otherwise + # JetStream can redeliver a message to another worker while this one is still + # waiting on a slow-but-legitimate response, causing a duplicate forward. + ACK_WAIT_SECONDS = int(os.getenv("ACK_WAIT_SECONDS", "60")) + # Endpoint mapping is still useful for constructing the target URL ENDPOINT_MAPPING = { "/live/api/v1/deliveries/createdeliveries": f"{BASE_URL}/live/api/v1/deliveries/createdeliveries", @@ -63,6 +68,7 @@ data: session: aiohttp.ClientSession = None running = True semaphore = None # Initialized in main + active_tasks: set = set() # In-flight process_message() tasks, drained on shutdown def signal_handler(sig, frame): global running @@ -87,7 +93,10 @@ data: if api_key: headers["Authorization"] = f"Bearer {api_key}" - data_to_forward = payload.get("data", payload) + if "data" not in payload: + print(f"?? Message for {endpoint} missing 'data' field, dropping.") + return "DROP", None + data_to_forward = payload["data"] # Payload Normalization for logs if endpoint == "/live/api/v2/deliveries/createdeliverylog": @@ -97,7 +106,7 @@ data: data_to_forward = [data_to_forward] http_method = method.upper() if method else "POST" - + try: # We use the global 'session' here async with session.request( @@ -108,7 +117,7 @@ data: headers=headers, timeout=aiohttp.ClientTimeout(total=30) ) as response: - + body = None try: body = await response.json() @@ -118,7 +127,7 @@ data: if 200 <= response.status < 300: return "OK", body elif response.status >= 500: - print(f"?? Server Error {response.status} from {endpoint}: {body}") + print(f"?? Server Error {response.status} from {endpoint}") return "RETRY", body else: print(f"?? Client Error {response.status} from {endpoint}: {body}") @@ -136,7 +145,7 @@ data: messages_in_flight.inc() start_time = asyncio.get_event_loop().time() endpoint = "unknown" - + try: payload = json.loads(msg.data.decode()) endpoint = payload.get("endpoint", payload.get("original_path", "unknown")) @@ -145,14 +154,14 @@ data: print(f"?? Processing {endpoint}") api_key = os.getenv("EXTERNAL_ENDPOINT_API_KEY", "") - + # Single attempt status, response_data = await forward_to_external(endpoint, payload, api_key, http_method) - + if status == "OK": await msg.ack() messages_processed.labels(status="success", endpoint=endpoint).inc() - + # Request-Reply Logic if msg.reply: reply_payload = json.dumps(response_data) if isinstance(response_data, (dict, list)) else str(response_data) @@ -162,7 +171,7 @@ data: # Unrecoverable error or unknown endpoint await msg.term() # Terminate stops redelivery messages_processed.labels(status="dropped", endpoint=endpoint).inc() - + else: # RETRY # Let JetStream handle backoff await msg.nak(delay=2) # Custom delay before redelivery if desired, or just nak() @@ -173,7 +182,8 @@ data: await msg.term() except Exception as e: print(f"? Critical Worker Error: {e}") - await msg.nak() + # Delay avoids a tight redelivery loop if this is a persistent bug + await msg.nak(delay=5) finally: messages_in_flight.dec() duration = asyncio.get_event_loop().time() - start_time @@ -181,7 +191,7 @@ data: async def main(): global nc, js, session, semaphore - + print(f"?? Starting Worker for Domain: {NATS_CONSUMER}") print(f"?? Stream: {NATS_STREAM}, Subjects: {FILTER_SUBJECTS}") print(f"?? Concurrency: {WORKER_CONCURRENCY}") @@ -202,12 +212,12 @@ data: # Create Pull Subscriptions for each filter subject # All sharing the same Consumer Name ensuring load balancing if multiple pods run this - + subs = [] for subject in FILTER_SUBJECTS: subject = subject.strip() if not subject: continue - + # FIX: On WorkQueue streams, we cannot reuse the same durable name for different filters # We append a sanitized version of the subject to ensure uniqueness per filter clean_subject_suffix = subject.replace(".", "_").replace("*", "all").replace(">", "all") @@ -217,8 +227,9 @@ data: try: sub = await js.pull_subscribe( subject, - durable=durable_name, - stream=NATS_STREAM + durable=durable_name, + stream=NATS_STREAM, + config=ConsumerConfig(ack_wait=ACK_WAIT_SECONDS) ) subs.append(sub) except Exception as e: @@ -238,19 +249,24 @@ data: # Batch size 10, short timeout to keep loop responsive msgs = await sub.fetch(10, timeout=0.5) for m in msgs: - asyncio.create_task(process_message(m)) + task = asyncio.create_task(process_message(m)) + active_tasks.add(task) + task.add_done_callback(active_tasks.discard) except NatsTimeoutError: pass except Exception as e: print(f"?? Fetch Error: {e}") await asyncio.sleep(1) - + # Small sleep to prevent tight loop if no messages # await asyncio.sleep(0.01) except Exception as e: print(f"?? Fatal Error: {e}") finally: + if active_tasks: + print(f"?? Draining {len(active_tasks)} in-flight message(s) before shutdown...") + await asyncio.wait(active_tasks, timeout=25) if session: await session.close() if nc: diff --git a/manifests/nearle/fiesta-gateway.yaml b/manifests/nearle/fiesta-gateway.yaml index 3ebd7cf..9befdc6 100644 --- a/manifests/nearle/fiesta-gateway.yaml +++ b/manifests/nearle/fiesta-gateway.yaml @@ -67,7 +67,7 @@ data: # Configuration NATS_URL = os.getenv("NATS_URL", "nats://nats-server:4222") NATS_USER = os.getenv("NATS_USER", "admin") - NATS_PASSWORD = os.getenv("NATS_PASSWORD", "package@321#") + NATS_PASSWORD = os.getenv("NATS_PASSWORD", "") # Endpoint to NATS subject mapping ENDPOINT_ROUTES = { @@ -179,7 +179,7 @@ data: subject = ENDPOINT_ROUTES.get(endpoint, "api.unknown") if not js: - raise HTTPException(status_code=503, detail="NATS by connected") + raise HTTPException(status_code=503, detail="NATS not connected") try: # Create a unique inbox for the reply diff --git a/manifests/nearle/jupiter-cors-proxy.yaml b/manifests/nearle/jupiter-cors-proxy.yaml index 568b023..b32126e 100644 --- a/manifests/nearle/jupiter-cors-proxy.yaml +++ b/manifests/nearle/jupiter-cors-proxy.yaml @@ -16,6 +16,15 @@ data: location / { proxy_pass http://jupiter_backend; + + # Strip any CORS headers the backend may set itself, so we don't + # end up sending duplicate Access-Control-* headers (browsers + # reject a response that has more than one value for these). + proxy_hide_header 'Access-Control-Allow-Origin'; + proxy_hide_header 'Access-Control-Allow-Methods'; + proxy_hide_header 'Access-Control-Allow-Headers'; + proxy_hide_header 'Access-Control-Max-Age'; + proxy_set_header Host $host; proxy_set_header X-Real-IP $remote_addr; proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;