diff --git a/Solutions/Kafka-Alert-Demo/KafkaConsumer-Python-Demo.py b/Solutions/Kafka-Alert-Demo/KafkaConsumer-Python-Demo.py new file mode 100644 index 00000000..bc9d556b --- /dev/null +++ b/Solutions/Kafka-Alert-Demo/KafkaConsumer-Python-Demo.py @@ -0,0 +1,64 @@ +from kafka import KafkaConsumer +import redis +import json +import smtplib +from email.mime.text import MIMEText +from email.mime.multipart import MIMEMultipart + +# Kafka 和 Redis 配置 +KAFKA_SERVER = 'localhost:9092' +REDIS_HOST = 'localhost' +REDIS_PORT = 6379 +ALARM_TOPIC = 'alarm_topic' + +# 邮件配置 +SMTP_SERVER = 'smtp.example.com' +SMTP_PORT = 587 +EMAIL_ADDRESS = 'alert@example.com' +EMAIL_PASSWORD = 'your_password' +RECIPIENT_EMAIL = 'recipient@example.com' + +# 初始化 Kafka 消费者和 Redis 客户端 +consumer = KafkaConsumer(ALARM_TOPIC, bootstrap_servers=KAFKA_SERVER) +redis_client = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, db=0) + +# 邮件发送函数 +def send_email(alarm): + msg = MIMEMultipart() + msg['From'] = EMAIL_ADDRESS + msg['To'] = RECIPIENT_EMAIL + msg['Subject'] = f"Alarm Notification - {alarm['level']}" + body = f""" + Alarm ID: {alarm['alarm_id']} + Level: {alarm['level']} + Message: {alarm['message']} + Source: {alarm['source']} + Timestamp: {alarm['timestamp']} + """ + msg.attach(MIMEText(body, 'plain')) + with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server: + server.starttls() + server.login(EMAIL_ADDRESS, EMAIL_PASSWORD) + server.sendmail(EMAIL_ADDRESS, RECIPIENT_EMAIL, msg.as_string()) + print(f"Email sent for alarm: {alarm['alarm_id']}") + +# 告警处理函数 +def process_alarm(message, deduplication=True): + alarm_data = json.loads(message.value) + alarm_id = alarm_data['alarm_id'] + timestamp = alarm_data['timestamp'] + key = f"{alarm_id}:{timestamp}" + + # 去重逻辑 + if deduplication: + if not redis_client.exists(key): + redis_client.setex(key, 3600, "1") # 设置1小时过期 + send_email(alarm_data) + else: + print(f"Duplicate alarm discarded: {alarm_id}") + else: + send_email(alarm_data) + +# 消费 Kafka 消息 +for message in consumer: + process_alarm(message, deduplication=True) # 控制去重 diff --git a/Solutions/Kafka-Alert-Demo/KafkaProducer-Python-Demo.py b/Solutions/Kafka-Alert-Demo/KafkaProducer-Python-Demo.py new file mode 100644 index 00000000..53a70ff8 --- /dev/null +++ b/Solutions/Kafka-Alert-Demo/KafkaProducer-Python-Demo.py @@ -0,0 +1,26 @@ +from kafka import KafkaProducer +import json +import time +import random + +producer = KafkaProducer( + bootstrap_servers='localhost:9092', + value_serializer=lambda v: json.dumps(v).encode('utf-8') +) + +alarm_levels = ["INFO", "WARNING", "CRITICAL"] + +def generate_alarm(): + return { + "alarm_id": random.randint(1000, 9999), + "timestamp": int(time.time()), + "level": random.choice(alarm_levels), + "message": "System load high", + "source": "Server-01" + } + +while True: + alarm = generate_alarm() + producer.send('alarm_topic', alarm) + print(f"Produced: {alarm}") + time.sleep(5) diff --git a/Solutions/Kafka-Alert-Demo/Readme.md b/Solutions/Kafka-Alert-Demo/Readme.md new file mode 100644 index 00000000..29cb964b --- /dev/null +++ b/Solutions/Kafka-Alert-Demo/Readme.md @@ -0,0 +1,16 @@ +# 方案设计概述 + +以下是基于 Kafka 和 Python 实现的告警系统方案,其中 Kafka 集群用于接收告警消息,消费者处理消息并根据需求去重或直接发送邮件通知。 + +- Kafka 集群:作为消息的推送和消费端点。 +- 生产者:模拟告警消息写入 Kafka 的 alarm_topic。 +- 消费者:从 Kafka 消费告警消息,执行去重逻辑(基于 Redis),然后发送邮件。 +- 邮件通知:通过 SMTP 发送告警邮件。 + +# 环境需求 + +- Kafka 集群 (至少 1 个 broker) +- Python (3.x) +- Kafka-Python 库 +- Redis (可选,用于去重) +- smtplib (Python 标准库) diff --git a/Solutions/Kafka-Alert-Demo/Setup-Kafka-Cluster.sh b/Solutions/Kafka-Alert-Demo/Setup-Kafka-Cluster.sh new file mode 100644 index 00000000..5a722ede --- /dev/null +++ b/Solutions/Kafka-Alert-Demo/Setup-Kafka-Cluster.sh @@ -0,0 +1,9 @@ +kubectl create namespace kafka +helm install kafka bitnami/kafka --namespace kafka \ + --set replicaCount=3 \ + --set persistence.enabled=true \ + --set persistence.size=8Gi \ + --set externalZookeeper.enabled=false \ + --set zookeeper.enabled=true +kubectl get pods --namespace kafka +kubectl get svc --namespace kafka diff --git a/Solutions/Kafka-Alert-Demo/Setup-Redis-Cluster.sh b/Solutions/Kafka-Alert-Demo/Setup-Redis-Cluster.sh new file mode 100644 index 00000000..658f8539 --- /dev/null +++ b/Solutions/Kafka-Alert-Demo/Setup-Redis-Cluster.sh @@ -0,0 +1,8 @@ +kubectl create namespace redis +helm install redis bitnami/redis-cluster --namespace redis \ + --set cluster.enabled=true \ + --set cluster.nodes=6 \ + --set persistence.enabled=true \ + --set persistence.size=8Gi +kubectl get pods --namespace redis +kubectl get svc --namespace redis