消费 Demo-第三方软件

import json
import time
import threading
import queue
from kafka import KafkaConsumer
import requests

# 日志消费参数配置
cls_consumer_config = {
    'bootstrap_servers': ['kafkaconsumer-ap-guangzhou.cls.tencentcs.com:9096'],
    'topic_name': '1254139626-XXXXXXXXX-XXXX-XXXX-bdd00b95d678', #消费主题 ID，请在 Kafka 协议消费的控制台复制。
    'group_id': 'xhm_consumer_group',
    'auto_offset_reset': 'earliest',
    'security_protocol': 'SASL_PLAINTEXT',
    'sasl_mechanism': 'PLAIN',
    'sasl_plain_username': "e60e9e91-XXXX-XXXX-XXXXX-f19fb6f4e626",#配置为 ${LogSetID}，即日志集 ID。
    'sasl_plain_password': "AKIDnTG#SfehOq3",#配置为 ${SecretId}#${SecretKey}，
    'api_version': (2, 5, 1)
}

# Splunk HEC 配置
SPLUNK_CONFIG = {
    'url': 'http:#192.168.1.1:8088/services/collector',#splunk HEC endpoint,请在Splunk侧获取
    'token': 'Splunk 1329e5cc-2c0d-XXXX-XXXX-332c753f54bd', #HEC token,请在Splunk侧获取
    'max_retries': 3,  # 失败重试次数
    'retry_delay': 1  # 重试延迟(秒)
}

# 创建线程安全的队列
log_queue = queue.Queue(maxsize=1000)  # 根据内存调整队列大小


def splunk_sender():
    """Splunk日志发送线程"""
    session = requests.Session()
    session.headers.update({
        'Authorization': SPLUNK_CONFIG['token'],
        'Content-Type': 'application/json'
    })

    while True:
        try:
            # 从队列获取日志
            log_data = log_queue.get()
            if log_data is None:  # 终止信号
                break

            # 构造Splunk事件
            event = {
                'event': {
                    'message': log_data.decode('utf-8'),
                    'kafka_topic': cls_consumer_config['topic_name']
                }
            }

            # 发送日志到Splunk（带重试机制）
            for attempt in range(SPLUNK_CONFIG['max_retries'] + 1):
                try:
                    response = session.post(
                        SPLUNK_CONFIG['url'],
                        data=json.dumps(event),
                        verify=False,  # 跳过SSL验证（-k参数）
                        timeout=5
                    )
                    response.raise_for_status()
                    print(f"Successfully sent to Splunk, status: {response.status_code}")
                    break  #成功则跳出重试循环
                except Exception as e:
                    if attempt < SPLUNK_CONFIG['max_retries']:
                        print(f"Attempt {attempt + 1} failed: {str(e)}, retrying...")
                        time.sleep(SPLUNK_CONFIG['retry_delay'])
                    else:
                        print(f"Failed to send to Splunk after {SPLUNK_CONFIG['max_retries']} attempts: {str(e)}")
                        # 可以添加死信队列处理或本地存储逻辑

        except Exception as e:
            print(f"Error in splunk_sender: {str(e)}")
        finally:
            log_queue.task_done()


def consume_messages():
    """Kafka消费主线程"""
    consumer = KafkaConsumer(
        cls_consumer_config['topic_name'],
        bootstrap_servers=cls_consumer_config['bootstrap_servers'],
        group_id=cls_consumer_config['group_id'],
        auto_offset_reset=cls_consumer_config['auto_offset_reset'],
        security_protocol=cls_consumer_config['security_protocol'],
        sasl_mechanism=cls_consumer_config['sasl_mechanism'],
        sasl_plain_username=cls_consumer_config['sasl_plain_username'],
        sasl_plain_password=cls_consumer_config['sasl_plain_password'],
        api_version=cls_consumer_config['api_version']
    )

    print('Start consuming messages...')
    try:
        for message in consumer:
            print(
                f"Received message from Topic: {message.topic}, Partition: {message.partition}, Offset: {message.offset}")
            # 将消息放入队列（阻塞直到有空位）
            log_queue.put(message.value)
    except Exception as e:
        print(f"Kafka consumer error: {str(e)}")
    finally:
        consumer.close()


if __name__ == "__main__":
    # 启动Splunk发送线程
    sender_thread = threading.Thread(target=splunk_sender, daemon=True)
    sender_thread.start()

    try:
        consume_messages()
    except KeyboardInterrupt:
        print("Shutting down...")
    finally:
        # 等待队列处理完毕
        log_queue.join()
        # 发送终止信号
        log_queue.put(None)
        sender_thread.join(timeout=10)
        print("All done.")

  input { 
  kafka {
        # cls kafka 协议消费控制台给出的的主题名称，例如 XXXXXX-633a268c-XXXX-4a4c-XXXX-7a9a1a7baXXXX，可在控制台复制
        topics => "您的消费主题"
        # 服务地址+端口，外网端口9096，内网端口9095，例子是内网消费，请根据您的实际情况填写
        bootstrap_servers => "kafkaconsumer-${region}.cls.tencentyun.com:9095"
        group_id => "您的消费组名称"
        security_protocol => "SASL_PLAINTEXT"
        sasl_mechanism => "PLAIN"
        # 用户名是日志集合 ID，例如 ca5cXXXXdd2e-4ac0af12-92d4b677d2c6 
        # 密码是用户的 SecretId#SecretKey 组合的字符串，例如 AKID********************************#XXXXuXtymIXT0Lac，注意不要丢失#。建议使用子账号密钥，主账号为子账号授权时，遵循最小权限原则，即子账号的访问策略中的 action、resource 都配置为最小范围，可以满足操作即可。
        sasl_jaas_config => "org.apache.kafka.common.security.plain.PlainLoginModule required username='${logsetID}' password='${SecretId}#${SecretKey}';"
    }
}
output {
    syslog {
        facility => "local7"
        host => "ecollector-eu.devo.io" 
        port => "443"
        severity => "informational"
        appname => "my.app.oracle.test"   
        protocol => "ssl-tcp"
        ssl_cert => "/Users/Bob/logstash/ca/domain.crt"
        ssl_key => "/Users/Bob/logstash/ca/domain.key"
        ssl_cacert => "/Users/Bob/logstash/ca/chain.crt"
        codec => line {
            format => "%{name},%{id}"
            }
        }
}