161 lines
4.4 KiB
Python
161 lines
4.4 KiB
Python
"""
|
|
IoT Consumer — Flask app that reads IoT events from Kafka (TLS + SASL)
|
|
and stores them in Redis (cache) + MongoDB (permanent).
|
|
|
|
Env vars (set by Terraform):
|
|
KAFKA_BROKERS — bootstrap server (host:port)
|
|
KAFKA_TOPIC — topic name
|
|
KAFKA_USERNAME — SASL username
|
|
KAFKA_PASSWORD — SASL password
|
|
KAFKA_GROUP_ID — consumer group id
|
|
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
|
|
import os
|
|
import ssl
|
|
import threading
|
|
|
|
from flask import Flask, jsonify
|
|
from kafka import KafkaConsumer
|
|
import redis
|
|
import pymongo
|
|
|
|
app = Flask(__name__)
|
|
|
|
# --- Config ---
|
|
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_GROUP_ID = os.environ.get("KAFKA_GROUP_ID", "iot-consumer-group")
|
|
|
|
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")
|
|
|
|
# --- Clients ---
|
|
events_consumed = 0
|
|
errors_count = 0
|
|
redis_client = None
|
|
mongo_client = None
|
|
mongo_coll = None
|
|
|
|
|
|
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,
|
|
)
|
|
print("[CONSUMER] Redis connected")
|
|
return redis_client
|
|
|
|
|
|
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():
|
|
global events_consumed, errors_count
|
|
|
|
ctx = ssl.create_default_context()
|
|
ctx.check_hostname = False
|
|
ctx.verify_mode = ssl.CERT_NONE
|
|
|
|
consumer = KafkaConsumer(
|
|
KAFKA_TOPIC,
|
|
bootstrap_servers=KAFKA_BROKERS,
|
|
security_protocol="SASL_SSL",
|
|
sasl_mechanism="PLAIN",
|
|
sasl_plain_username=KAFKA_USERNAME,
|
|
sasl_plain_password=KAFKA_PASSWORD,
|
|
ssl_context=ctx,
|
|
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}'...")
|
|
|
|
for message in consumer:
|
|
try:
|
|
store_event(message.value)
|
|
events_consumed += 1
|
|
if events_consumed % 10 == 0:
|
|
print(f"[CONSUMER] Processed {events_consumed} events")
|
|
except Exception as e:
|
|
errors_count += 1
|
|
print(f"[CONSUMER] Error: {e}")
|
|
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return jsonify({
|
|
"service": "iot-consumer",
|
|
"status": "running",
|
|
"kafka_topic": KAFKA_TOPIC,
|
|
"events_consumed": events_consumed,
|
|
"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)
|