fix: site/ structure (no Procfile)

2026-07-21 17:50:22 +04:00
parent 3e1505f6b1
commit 6b63433077
3 changed files with 3 additions and 4 deletions
@@ -0,0 +1,149 @@
+"""
+IoT Producer — Flask app that generates fake IoT sensor events
+and sends them to Kafka (TLS + SASL).
+
+Env vars (set by Terraform):
+  KAFKA_BROKERS     — bootstrap server (host:port)
+  KAFKA_TOPIC       — topic name
+  KAFKA_USERNAME    — SASL username
+  KAFKA_PASSWORD    — SASL password
+  KAFKA_CA_CRT      — CA certificate (PEM)
+  KAFKA_USER_CRT    — user certificate (PEM)
+  KAFKA_USER_KEY    — user private key (PEM)
+  PRODUCE_INTERVAL  — seconds between events (default 3)
+"""
+
+import json
+import os
+import random
+import ssl
+import tempfile
+import time
+import threading
+import uuid
+from datetime import datetime, timezone
+
+from flask import Flask, jsonify
+from kafka import KafkaProducer
+
+app = Flask(__name__)
+
+# --- Config from env ---
+KAFKA_BROKERS = os.environ["KAFKA_BROKERS"]
+KAFKA_TOPIC = os.environ["KAFKA_TOPIC"]
+KAFKA_USERNAME = os.environ["KAFKA_USERNAME"]
+KAFKA_PASSWORD = os.environ["KAFKA_PASSWORD"]
+KAFKA_CA_CRT = os.environ.get("KAFKA_CA_CRT", "")
+KAFKA_USER_CRT = os.environ.get("KAFKA_USER_CRT", "")
+KAFKA_USER_KEY = os.environ.get("KAFKA_USER_KEY", "")
+PRODUCE_INTERVAL = float(os.environ.get("PRODUCE_INTERVAL", 3))
+
+# --- TLS certs → temp files (kafka-python needs file paths) ---
+_cert_dir = tempfile.mkdtemp(prefix="kafka-certs-")
+
+def _write_cert(name, content):
+    if not content:
+        return None
+    path = os.path.join(_cert_dir, name)
+    with open(path, "w") as f:
+        f.write(content)
+    return path
+
+_ca_file = _write_cert("ca.crt", KAFKA_CA_CRT)
+_cert_file = _write_cert("user.crt", KAFKA_USER_CRT)
+_key_file = _write_cert("user.key", KAFKA_USER_KEY)
+
+# --- IoT simulation config ---
+SENSORS = [
+    {"id": "temp_living",   "type": "temperature", "location": "living_room", "unit": "°C",   "min": 18.0, "max": 28.0},
+    {"id": "temp_bedroom",  "type": "temperature", "location": "bedroom",     "unit": "°C",   "min": 16.0, "max": 26.0},
+    {"id": "temp_office",   "type": "temperature", "location": "office",      "unit": "°C",   "min": 19.0, "max": 27.0},
+    {"id": "humidity_living","type": "humidity",    "location": "living_room", "unit": "%",    "min": 30.0, "max": 70.0},
+    {"id": "humidity_bed",  "type": "humidity",    "location": "bedroom",     "unit": "%",    "min": 35.0, "max": 65.0},
+    {"id": "power_kitchen", "type": "power_meter", "location": "kitchen",     "unit": "kW",   "min": 0.1,  "max": 3.5},
+    {"id": "power_living",  "type": "power_meter", "location": "living_room", "unit": "kW",   "min": 0.2,  "max": 2.0},
+    {"id": "power_office",  "type": "power_meter", "location": "office",      "unit": "kW",   "min": 0.3,  "max": 4.0},
+]
+
+producer = None
+events_sent = 0
+errors_count = 0
+
+
+def get_producer():
+    global producer
+    if producer is None:
+        kwargs = {
+            "bootstrap_servers": KAFKA_BROKERS,
+            "security_protocol": "SASL_SSL",
+            "sasl_mechanism": "PLAIN",
+            "sasl_plain_username": KAFKA_USERNAME,
+            "sasl_plain_password": KAFKA_PASSWORD,
+            "value_serializer": lambda v: json.dumps(v).encode("utf-8"),
+            "key_serializer": lambda k: k.encode("utf-8") if k else None,
+        }
+        if _ca_file:
+            kwargs["ssl_cafile"] = _ca_file
+        if _cert_file:
+            kwargs["ssl_certfile"] = _cert_file
+        if _key_file:
+            kwargs["ssl_keyfile"] = _key_file
+        # Internal cluster communication — skip hostname check
+        kwargs["ssl_check_hostname"] = False
+
+        producer = KafkaProducer(**kwargs)
+    return producer
+
+
+def generate_event():
+    sensor = random.choice(SENSORS)
+    value = round(random.uniform(sensor["min"], sensor["max"]), 2)
+    return {
+        "event_id": str(uuid.uuid4()),
+        "sensor_id": sensor["id"],
+        "device_type": sensor["type"],
+        "location": sensor["location"],
+        "value": value,
+        "unit": sensor["unit"],
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    }
+
+
+def produce_loop():
+    global events_sent, errors_count
+    while True:
+        try:
+            event = generate_event()
+            p = get_producer()
+            future = p.send(KAFKA_TOPIC, key=event["sensor_id"], value=event)
+            future.get(timeout=5)
+            events_sent += 1
+            print(f"[PRODUCER] Sent: {event['sensor_id']} = {event['value']}{event['unit']} ({event['location']})")
+        except Exception as e:
+            errors_count += 1
+            print(f"[PRODUCER] Error: {e}")
+        time.sleep(PRODUCE_INTERVAL)
+
+
+@app.route("/")
+def index():
+    return jsonify({
+        "service": "iot-producer",
+        "status": "running",
+        "kafka_topic": KAFKA_TOPIC,
+        "events_sent": events_sent,
+        "errors": errors_count,
+        "interval": PRODUCE_INTERVAL,
+        "sensors": len(SENSORS),
+    })
+
+
+@app.route("/health")
+def health():
+    return jsonify({"status": "ok", "events_sent": events_sent})
+
+
+if __name__ == "__main__":
+    threading.Thread(target=produce_loop, daemon=True).start()
+    port = int(os.environ.get("PORT", 5000))
+    app.run(host="0.0.0.0", port=port)