​云函数实践(含代码)：将日志服务的日志投递到自建 Kafka 的 3 个步骤

#!/usr/bin/env python
# -*- coding: utf-8 -*-
 
import time
import logging
import os
import base64
import json
import gzip
import urllib
from kafka import KafkaProducer
from kafka.errors import KafkaError
from StringIO import StringIO

logger = logging.getLogger('kafka')
logger.setLevel(logging.INFO)
 
class ClsToKafka(object):
    """
    CLS 消息投递 kafka
    """

    def __init__(self, host, **kwargs):
        self.host = host

        self.producer = KafkaProducer(bootstrap_servers = [self.host], 
            retries = 10,
            max_in_flight_requests_per_connection = 1, 
            request_timeout_ms = 30000,
            max_block_ms = 60000,
            **kwargs
        )def send(self, topic, records):
        """
        异步生产 kafka 消息
        """

        global count
        count = 0
        def on_send_success(record_metadata):
            global count
            count = count +1

        def on_send_error(excp):
            logger.error('failed to send message', exc_info = excp)

        s_time = time.time()
        
        try:
            ## 修改 1： 原始消息是 JSON 格式（日志服务采集容器服务输出的日志格式是 JSON），每条消息位于 .records（type: List）
            ## for record in records:
            for record in records['records']:
                key = ""# 当 key 为""或者为"None" 时，要传入 key=None，这样 python kafka 库会随机选取一个 partition 写入消息
                if key == "" or key =="None":
                    key = None
                ## 修改 2：record 是 dict，因为原始数据就是 JSON，需要转成 str，否则调用 self.producer.send 会报错 "assert type(value_bytes) in (bytes, bytearray, memoryview, type(None))"
                value = json.dumps(record)

                # 也可以对消息进行处理后再转存
                #value = deal_message(data)

                self.producer.send(topic, key = key, value = value).add_callback(on_send_success).add_errback(on_send_error)

            # block until all async messages are sent
            self.producer.flush()
        except KafkaError as e:
            return e
        finally:
            if self.producer is not None:
                self.producer.close()e_time = time.time()

        return "{} messages delivered in {}s".format(count, e_time - s_time)

# 这里可以对消息进行处理后返回
def deal_message(message):
    return message

def main_handler(event, context):
    kafka_address = os.getenv("kafka_address")
    kafka_topic_name = os.getenv("kafka_topic_name")
  
    kafka_to_kafka = ClsToKafka(
        kafka_address
        #security_protocol = "SASL_PLAINTEXT",
        #sasl_mechanism = "PLAIN",
        #sasl_plain_username = "ckafka-80o10xxx#lkoxx",
        #sasl_plain_password = "ccllxxxx",
        #api_version=(0, 10, 2)
    )event = json.loads(gzip.GzipFile(fileobj=StringIO(event['clslogs']['data'].decode('base64'))).read())# print("type of event: %s" % type(event))
  
    ## 修改 3：直接使用 event 这个字典，便于从字典中获取每条消息的内容
    ## data = json.dumps(event, indent=4, sort_keys=True)
    ## ret = kafka_to_kafka.send(kafka_topic_name,data) 
    ret = kafka_to_kafka.send(kafka_topic_name,event) 
    logger.info(ret)
    return ret
展开 

# ./kafka-console-consumer.sh --bootstrap-server 10.0.0.29:9092 --topic scf_topic --offset latest --partition 0
{"content": "{\\"Accept\\":\\"*/*\\",\\"Body\\":\\"\\",\\"Host\\":\\"header.dev.xxx.cn\\",\\"Method\\":\\"GET\\",\\"Protocol\\":\\"HTTP/1.1\\",\\"Referer\\":\\"\\",\\"RemoteAddr\\":\\"172.16.7.71:37468\\",\\"RequestURI\\":\\"/\\",\\"Type\\":\\"web\\",\\"UserAgent\\":\\"Qcloud-boce\\",\\"X-Forwarded-For\\":\\"58.87.66.69\\"}", "timestamp": 1618716491203428}
{"content": "{\\"Accept\\":\\"*/*\\",\\"Body\\":\\"\\",\\"Host\\":\\"header.dev.xxx.cn\\",\\"Method\\":\\"GET\\",\\"Protocol\\":\\"HTTP/1.1\\",\\"Referer\\":\\"\\",\\"RemoteAddr\\":\\"172.16.7.71:36864\\",\\"RequestURI\\":\\"/\\",\\"Type\\":\\"curl\\",\\"UserAgent\\":\\"Qcloud-curl\\",\\"X-Forwarded-For\\":\\"180.163.9.66\\"}", "timestamp": 1618716494178403}
...
Processed a total of 9 messages

DNS lookup failed for hadoop-29.com:9092, exception was [Errno -2] Name or service not known. Is your advertised.listeners (called advertised.host.name before Kafka 9) correct and resolvable?

DNS lookup failed for hadoop-29.com:9092 (0)