Add Kafka-Alert-Demo Codes

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)  # 控制去重

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)

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 标准库)

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

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