Nearle/kubernetes
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix deployment tooling: shell scripts and Terraform

Browse Source 
Terraform (validated with the real terraform CLI - was never actually
run against this cluster, no state file existed):
- Delete main.tf: it declared a duplicate kubernetes_namespace.core
  (also in namespaces.tf) and a duplicate provider "kubernetes" block
  (also in providers.tf), both hard errors that would fail
  `terraform plan` immediately.
- Fix workloads.tf references to 6 files deleted in the manifest
  cleanup (jupiter-sts/svc, atlantis-sts/svc, fiesta-sts/svc) - now
  points at the canonical nearle-jupiter/atlantis/fiesta.yaml.
- Fix every kubernetes_manifest resource: they fed multi-document
  YAML (multiple '---'-separated docs per file) straight into
  yamldecode(), which only parses a single document. Rewrote using a
  split-on-'---' + for_each pattern, confirmed safe first by checking
  separator counts exactly match document counts for every affected
  file (no embedded '---' inside any script/config content).
- Add the doormile namespace; rename kubernetes_namespace to
  kubernetes_namespace_v1 (fixes a deprecation warning).
- `terraform validate` now passes clean.

Shell scripts:
- deploy-nearle-stack.sh only applied 4 of the ~13 files in
  manifests/nearle/ - missing the ConfigMap/Secrets fiesta/jupiter/
  titan/ariane need via envFrom, the fiesta gateway script ConfigMap,
  atlantis entirely, and the Gateway/ReferenceGrant/jupiter-cors-proxy
  resources. Now applies every file (verified by diffing the
  directory listing against the script).
- Added deploy-doormile.sh and deploy-ingress.sh - nothing previously
  applied ingress-unified.yaml or traefik-middlewares.yaml at all.
- Rewrote deploy.sh as an orchestrator calling all of the above in
  order (previously referenced a manifests/namespace.yaml layout that
  hasn't existed since before this repo's initial commit).
- Rewrote check-k8s-status.sh to check the real namespaces
  (core/nearle/alaska/doormile/kubernetes-dashboard) instead of a
  'nats-backend' namespace that never existed in this repo.
- Fixed a `cd` bug in setup-jetstream.sh that made it change into
  shfiles/ and then look for scripts/setup_jetstream.py there (a
  child directory that doesn't exist) - it could never have found its
  own target file. Now pulls NATS credentials from the live
  nats-credentials Secret instead of a third hardcoded copy.

Python scripts:
- sync_manifests.py had hardcoded Windows paths (e:\nats\kubernetes\...)
  - replaced with paths relative to the script's own location so it
  actually runs here (or anywhere). Verified by running it.
- setup_jetstream.py created durable consumers under different names
  than worker.py computes at runtime ({NATS_CONSUMER}_{subject}), so
  its max_deliver/ack_wait settings never actually reached the
  consumers workers bind to. Naming now derived with the same logic
  worker.py uses - verified all 10 derived names match workers.yaml
  exactly.
- purge-old-messages.py had hardcoded NATS credentials with no env
  var override at all - fixed to match the pattern used everywhere
  else.
This commit is contained in:
Suriya
2026-07-18 16:08:15 +05:30
parent 91dd240431
commit 0a8c3b0374
14 changed files with 303 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
scripts/purge-old-messages.py
View File

@@ -7,9 +7,9 @@ import nats
import os
async def purge_messages():
nats_url = "nats://nats.workolik.com:4222"
nats_user = "admin"
nats_password = "package@321#"
nats_url = os.getenv("NATS_URL", "nats://nats.workolik.com:4222")
nats_user = os.getenv("NATS_USER", "admin")
nats_password = os.getenv("NATS_PASSWORD", "")
try:
print(f"Connecting to NATS at {nats_url}...")

141
scripts/setup_jetstream.py
View File

@@ -1,77 +1,54 @@
#!/usr/bin/env python3
"""
Setup JetStream stream and consumer for NATS
Setup JetStream streams and consumers for NATS
Run this after NATS is deployed and running
Durable consumer names are derived with the exact same logic worker.py uses
at runtime (NATS_CONSUMER + sanitized subject). WORKER_DOMAINS below must be
kept in sync with the NATS_STREAM / NATS_CONSUMER / FILTER_SUBJECT env vars
in manifests/core/workers.yaml - if they drift apart, the consumers created
here (with their max_deliver/ack_wait settings) will never be the ones the
workers actually bind to, and this script's config becomes a no-op.
"""
import asyncio
import nats
import os
import sys
WORKER_DOMAINS = [
{"stream": "ORDERS", "consumer": "orders-worker", "subjects": [
"api.v1.mob.orders.createorder",
]},
{"stream": "DELIVERIES", "consumer": "deliveries-worker", "subjects": [
"api.v1.deliveries.createdeliveries",
"api.v1.deliveries.updatedelivery",
"api.v2.deliveries.createdeliverylog",
]},
{"stream": "CUSTOMERS", "consumer": "customers-worker", "retention": "work", "subjects": [
"api.v1.mob.customers.login",
"api.v1.mob.customers.create",
]},
{"stream": "RIDER", "consumer": "rider-logs-worker", "subjects": [
"api.v2.partners.createriderlog",
"api.v2.partners.createbreaklog",
"api.v2.partners.updatebreaklog",
]},
{"stream": "PRODUCTS", "consumer": "products-worker", "subjects": [
"api.v1.web.products.create",
]},
]
def durable_name(consumer: str, subject: str) -> str:
"""Mirrors worker.py's durable-name derivation exactly."""
suffix = subject.replace(".", "_").replace("*", "all").replace(">", "all")
return f"{consumer}_{suffix}"
async def setup_jetstream():
"""Configure JetStream with two streams (DELIVERIES, RIDER) and per-subject consumers."""
nats_url = os.getenv("NATS_URL", "nats://nats.workolik.com:4222")
nats_user = os.getenv("NATS_USER", "admin")
nats_password = os.getenv("NATS_PASSWORD", "package@321#")
# Stream definitions
streams = {
"DELIVERIES": {
"subjects": [
"api.v1.deliveries.createdeliveries",
"api.v1.deliveries.updatedelivery",
"api.v2.deliveries.createdeliverylog",
],
},
"RIDER": {
"subjects": [
"api.v2.partners.createriderlog",
"api.v2.partners.createbreaklog",
"api.v2.partners.updatebreaklog",
],
},
"ORDERS": {
"subjects": [
"api.v1.mob.orders.createorder",
],
},
"PRODUCTS": {
"subjects": [
"api.v1.web.products.create",
],
},
"CUSTOMERS": {
"subjects": [
"api.v1.mob.customers.login",
"api.v1.mob.customers.create",
],
"retention": "work" # Special handling for Login queue: delete immediately after ack
},
}
# Per-subject durable consumers
consumers = {
"DELIVERIES": {
"api.v1.deliveries.createdeliveries": "deliveries_createdeliveries",
"api.v1.deliveries.updatedelivery": "deliveries_updatedelivery",
"api.v2.deliveries.createdeliverylog": "deliveries_createdeliverylog",
},
"RIDER": {
"api.v2.partners.createriderlog": "rider_createriderlog",
"api.v2.partners.createbreaklog": "rider_createbreaklog",
"api.v2.partners.updatebreaklog": "rider_updatebreaklog",
},
"ORDERS": {
"api.v1.mob.orders.createorder": "orders_createorder",
},
"PRODUCTS": {
"api.v1.web.products.create": "products_create",
},
"CUSTOMERS": {
"api.v1.mob.customers.login": "customers_login",
"api.v1.mob.customers.create": "customers_create",
},
}
nats_password = os.getenv("NATS_PASSWORD", "")
try:
print(f"Connecting to NATS at {nats_url}...")
@@ -85,17 +62,16 @@ async def setup_jetstream():
js = nc.jetstream()
# Create / recreate streams
for stream_name, cfg in streams.items():
for domain in WORKER_DOMAINS:
stream_name = domain["stream"]
subjects = domain["subjects"]
retention_policy = domain.get("retention", "limits")
try:
info = await js.stream_info(stream_name)
print(f"⚠️ Stream '{stream_name}' already exists with subjects={info.config.subjects}, updating...")
# Determine retention policy
retention_policy = cfg.get("retention", "limits")
await js.update_stream(
name=stream_name,
subjects=cfg["subjects"],
subjects=subjects,
storage="memory",
retention=retention_policy,
max_age=24 * 60 * 60,
@@ -106,13 +82,9 @@ async def setup_jetstream():
except Exception as e:
if "not found" in str(e).lower() or "404" in str(e).lower():
print(f"Creating stream '{stream_name}'...")
# Determine retention policy (use 'limits' by default, 'work' for queues)
retention_policy = cfg.get("retention", "limits")
await js.add_stream(
name=stream_name,
subjects=cfg["subjects"],
subjects=subjects,
storage="memory",
retention=retention_policy,
max_age=24 * 60 * 60,
@@ -123,10 +95,14 @@ async def setup_jetstream():
else:
print(f"⚠️ Could not inspect stream '{stream_name}': {e}")
# Create durable consumers per subject
# Create durable consumers per subject - names match what worker.py
# computes at runtime, so max_deliver/ack_wait here actually apply.
print("\nConfiguring consumers...")
for stream_name, subject_map in consumers.items():
for subject, durable in subject_map.items():
for domain in WORKER_DOMAINS:
stream_name = domain["stream"]
consumer = domain["consumer"]
for subject in domain["subjects"]:
durable = durable_name(consumer, subject)
try:
print(f"Creating consumer '{durable}' on stream '{stream_name}' for subject '{subject}'...")
await js.add_consumer(
@@ -136,7 +112,7 @@ async def setup_jetstream():
ack_policy="explicit",
deliver_policy="all",
max_deliver=5,
ack_wait=30,
ack_wait=60,
)
print(f"✅ Consumer '{durable}' created")
except Exception as e:
@@ -147,12 +123,12 @@ async def setup_jetstream():
print("\n✅ JetStream setup complete!")
print(" Streams:")
for name, cfg in streams.items():
print(f" - {name}: {', '.join(cfg['subjects'])}")
for domain in WORKER_DOMAINS:
print(f" - {domain['stream']}: {', '.join(domain['subjects'])}")
print(" Consumers:")
for stream_name, subject_map in consumers.items():
for subject, durable in subject_map.items():
print(f" - {durable}: stream={stream_name}, subject={subject}")
for domain in WORKER_DOMAINS:
for subject in domain["subjects"]:
print(f" - {durable_name(domain['consumer'], subject)}: stream={domain['stream']}, subject={subject}")
await nc.close()
sys.exit(0)
@@ -163,4 +139,3 @@ async def setup_jetstream():
if __name__ == "__main__":
asyncio.run(setup_jetstream())

10
scripts/sync_manifests.py
View File

@@ -51,16 +51,18 @@ def update_yaml_with_script(yaml_path, script_path, key_line_start):
print(f"Successfully updated {yaml_path}")
REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# Update Fiesta Gateway
update_yaml_with_script(
r'e:\nats\kubernetes\manifests\nearle\fiesta-gateway.yaml',
r'e:\nats\kubernetes\conf\app.py',
os.path.join(REPO_ROOT, 'manifests', 'nearle', 'fiesta-gateway.yaml'),
os.path.join(REPO_ROOT, 'conf', 'app.py'),
' app.py: |'
)
# Update Worker Script
update_yaml_with_script(
r'e:\nats\kubernetes\manifests\core\worker-script.yaml',
r'e:\nats\kubernetes\conf\worker.py',
os.path.join(REPO_ROOT, 'manifests', 'core', 'worker-script.yaml'),
os.path.join(REPO_ROOT, 'conf', 'worker.py'),
' worker.py: |'
)