From 3a9eef88cd426202db4b7166ad04673d58dcd97c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E2=80=9CNaeel=E2=80=9D?= Date: Tue, 21 Jul 2026 18:17:18 +0400 Subject: [PATCH] =?UTF-8?q?fix:=20Kafka=20=E2=86=92=20Redis=20+=20MongoDB?= =?UTF-8?q?=20(no=20ClickHouse)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- requirements.txt | 3 +- site/app.py | 144 +++++++++++++++++++++++------------------------ 2 files changed, 74 insertions(+), 73 deletions(-) diff --git a/requirements.txt b/requirements.txt index 854d296..fa0962d 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,4 +1,5 @@ Flask>=3.0 gunicorn>=21.2 kafka-python>=2.0 -clickhouse-connect>=0.7 +redis>=5.0 +pymongo>=4.0 diff --git a/site/app.py b/site/app.py index d58f346..48077fe 100644 --- a/site/app.py +++ b/site/app.py @@ -1,6 +1,6 @@ """ IoT Consumer — Flask app that reads IoT events from Kafka (TLS + SASL) -and inserts them into ClickHouse. +and stores them in Redis (cache) + MongoDB (permanent). Env vars (set by Terraform): KAFKA_BROKERS — bootstrap server (host:port) @@ -11,11 +11,11 @@ Env vars (set by Terraform): KAFKA_USER_CRT — user certificate (PEM) KAFKA_USER_KEY — user private key (PEM) 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 + REDIS_HOST — Redis host + REDIS_PORT — Redis port (default 6379) + REDIS_PASSWORD — Redis password + MONGO_URI — MongoDB connection URI + MONGO_DB — MongoDB database name """ import json @@ -25,7 +25,8 @@ import threading from flask import Flask, jsonify from kafka import KafkaConsumer -import clickhouse_connect +import redis +import pymongo app = Flask(__name__) @@ -39,11 +40,12 @@ KAFKA_USER_CRT = os.environ.get("KAFKA_USER_CRT", "") KAFKA_USER_KEY = os.environ.get("KAFKA_USER_KEY", "") 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"] +REDIS_HOST = os.environ["REDIS_HOST"] +REDIS_PORT = int(os.environ.get("REDIS_PORT", 6379)) +REDIS_PASSWORD = os.environ.get("REDIS_PASSWORD", "") + +MONGO_URI = os.environ["MONGO_URI"] +MONGO_DB = os.environ.get("MONGO_DB", "iot") # --- TLS certs → temp files --- _cert_dir = tempfile.mkdtemp(prefix="kafka-certs-") @@ -60,56 +62,68 @@ _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) +# --- Clients --- events_consumed = 0 errors_count = 0 -ch_client = None +redis_client = None +mongo_client = None +mongo_coll = 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, +def get_redis(): + global redis_client + if redis_client is None: + redis_client = redis.Redis( + host=REDIS_HOST, + port=REDIS_PORT, + password=REDIS_PASSWORD or None, + decode_responses=True, ) - 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 + print("[CONSUMER] Redis connected") + return redis_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 get_mongo(): + global mongo_client, mongo_coll + if mongo_client is None: + mongo_client = pymongo.MongoClient(MONGO_URI) + db = mongo_client[MONGO_DB] + mongo_coll = db["iot_events"] + mongo_coll.create_index("timestamp", expireAfterSeconds=604800) + print("[CONSUMER] MongoDB connected") + return mongo_coll + + +def store_event(event): + r = get_redis() + sensor_id = event["sensor_id"] + value = event["value"] + unit = event["unit"] + ts = event["timestamp"] + + # Redis: latest value per sensor + r.hset("iot:latest", sensor_id, json.dumps({ + "value": value, "unit": unit, "location": event["location"], "timestamp": ts, + })) + + # Redis: counter per sensor type + r.hincrby("iot:counters", event["device_type"], 1) + + # Redis: sorted set for recent events (last 1000) + r.zadd("iot:recent", {json.dumps(event): 0}) + r.zremrangebyrank("iot:recent", 0, -1001) + + # MongoDB: permanent storage + coll = get_mongo() + coll.insert_one({ + "event_id": event["event_id"], + "sensor_id": sensor_id, + "device_type": event["device_type"], + "location": event["location"], + "value": value, + "unit": unit, + "timestamp": ts, + }) def consume_loop(): @@ -135,20 +149,14 @@ def consume_loop(): kwargs["ssl_check_hostname"] = False consumer = KafkaConsumer(KAFKA_TOPIC, **kwargs) - print(f"[CONSUMER] Listening on topic '{KAFKA_TOPIC}'...") - batch = [] for message in consumer: try: - event = message.value - batch.append(event) + store_event(message.value) events_consumed += 1 - - if len(batch) >= 10: - insert_batch(batch) - print(f"[CONSUMER] Inserted {len(batch)} events into ClickHouse") - batch = [] + if events_consumed % 10 == 0: + print(f"[CONSUMER] Processed {events_consumed} events") except Exception as e: errors_count += 1 print(f"[CONSUMER] Error: {e}") @@ -156,19 +164,11 @@ def consume_loop(): @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, })