From c2b8cf2f8f066702c1ed13c3bff088654a132378 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E2=80=9CNaeel=E2=80=9D?= <tazet@narod.ru>
Date: Tue, 21 Jul 2026 16:32:18 +0400
Subject: [PATCH] =?UTF-8?q?Initial=20commit=20=E2=80=94=20IoT=20Kafka=20co?=
 =?UTF-8?q?nsumer?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 Procfile         |   1 +
 app.py           | 155 +++++++++++++++++++++++++++++++++++++++++++++++
 requirements.txt |   4 ++
 3 files changed, 160 insertions(+)
 create mode 100644 Procfile
 create mode 100644 app.py
 create mode 100644 requirements.txt

diff --git a/Procfile b/Procfile
new file mode 100644
index 0000000..c79a8f9
--- /dev/null
+++ b/Procfile
@@ -0,0 +1 @@
+web: gunicorn -b 0.0.0.0:$PORT app:app
diff --git a/app.py b/app.py
new file mode 100644
index 0000000..8a2fc02
--- /dev/null
+++ b/app.py
@@ -0,0 +1,155 @@
+"""
+IoT Consumer — Flask app that reads IoT events from Kafka
+and inserts them into ClickHouse.
+
+Env vars (set by Terraform):
+  KAFKA_BROKERS     — Kafka bootstrap servers
+  KAFKA_TOPIC       — topic name
+  KAFKA_USERNAME    — SASL username
+  KAFKA_PASSWORD    — SASL password
+  KAFKA_GROUP_ID    — consumer group id
+  CH_HOST           — ClickHouse host
+  CH_PORT           — ClickHouse port (default 8123)
+  CH_USER           — ClickHouse user
+  CH_PASSWORD       — ClickHouse password
+  CH_DATABASE       — ClickHouse database name
+"""
+
+import json
+import os
+import threading
+
+from flask import Flask, jsonify
+from kafka import KafkaConsumer
+import clickhouse_connect
+
+app = Flask(__name__)
+
+# --- Config ---
+KAFKA_BROKERS = os.environ["KAFKA_BROKERS"].split(",")
+KAFKA_TOPIC = os.environ["KAFKA_TOPIC"]
+KAFKA_USERNAME = os.environ["KAFKA_USERNAME"]
+KAFKA_PASSWORD = os.environ["KAFKA_PASSWORD"]
+KAFKA_GROUP_ID = os.environ.get("KAFKA_GROUP_ID", "iot-consumer-group")
+
+CH_HOST = os.environ["CH_HOST"]
+CH_PORT = int(os.environ.get("CH_PORT", 8123))
+CH_USER = os.environ["CH_USER"]
+CH_PASSWORD = os.environ["CH_PASSWORD"]
+CH_DATABASE = os.environ["CH_DATABASE"]
+
+events_consumed = 0
+errors_count = 0
+ch_client = None
+
+
+def get_ch_client():
+    global ch_client
+    if ch_client is None:
+        ch_client = clickhouse_connect.get_client(
+            host=CH_HOST,
+            port=CH_PORT,
+            username=CH_USER,
+            password=CH_PASSWORD,
+            database=CH_DATABASE,
+        )
+        # Create table if not exists
+        ch_client.command("""
+            CREATE TABLE IF NOT EXISTS iot_events (
+                event_id String,
+                sensor_id String,
+                device_type String,
+                location String,
+                value Float64,
+                unit String,
+                timestamp DateTime64(3)
+            ) ENGINE = MergeTree()
+            ORDER BY (timestamp, sensor_id)
+        """)
+        print("[CONSUMER] ClickHouse table ready")
+    return ch_client
+
+
+def insert_batch(events):
+    if not events:
+        return
+    rows = [[
+        e["event_id"],
+        e["sensor_id"],
+        e["device_type"],
+        e["location"],
+        e["value"],
+        e["unit"],
+        e["timestamp"],
+    ] for e in events]
+    try:
+        client = get_ch_client()
+        client.insert("iot_events", rows,
+            column_names=["event_id", "sensor_id", "device_type", "location", "value", "unit", "timestamp"])
+    except Exception as e:
+        print(f"[CONSUMER] Insert error: {e}")
+        raise
+
+
+def consume_loop():
+    global events_consumed, errors_count
+
+    consumer = KafkaConsumer(
+        KAFKA_TOPIC,
+        bootstrap_servers=KAFKA_BROKERS,
+        security_protocol="SASL_PLAINTEXT",
+        sasl_mechanism="PLAIN",
+        sasl_plain_username=KAFKA_USERNAME,
+        sasl_plain_password=KAFKA_PASSWORD,
+        group_id=KAFKA_GROUP_ID,
+        auto_offset_reset="earliest",
+        value_deserializer=lambda m: json.loads(m.decode("utf-8")),
+        max_poll_records=20,
+    )
+
+    print(f"[CONSUMER] Listening on topic '{KAFKA_TOPIC}'...")
+
+    batch = []
+    for message in consumer:
+        try:
+            event = message.value
+            batch.append(event)
+            events_consumed += 1
+
+            if len(batch) >= 10:
+                insert_batch(batch)
+                print(f"[CONSUMER] Inserted {len(batch)} events into ClickHouse")
+                batch = []
+        except Exception as e:
+            errors_count += 1
+            print(f"[CONSUMER] Error: {e}")
+
+
+@app.route("/")
+def index():
+    query = "SELECT count() FROM iot_events"
+    try:
+        client = get_ch_client()
+        total = client.query(query).result_rows[0][0]
+    except Exception:
+        total = 0
+
+    return jsonify({
+        "service": "iot-consumer",
+        "status": "running",
+        "kafka_topic": KAFKA_TOPIC,
+        "events_consumed": events_consumed,
+        "events_in_db": total,
+        "errors": errors_count,
+    })
+
+
+@app.route("/health")
+def health():
+    return jsonify({"status": "ok", "events_consumed": events_consumed})
+
+
+if __name__ == "__main__":
+    threading.Thread(target=consume_loop, daemon=True).start()
+    port = int(os.environ.get("PORT", 5000))
+    app.run(host="0.0.0.0", port=port)
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..68072d3
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,4 @@
+Flask
+gunicorn
+kafka-python
+clickhouse-connect