使用消费组通过Splunk HEC或Syslog投递日志到SIEM-日志服务-阿里云

当需要投递日志到 SIEM 时，可通过部署一个能够连接日志服务与 SIEM 的应用程序，利用 SLS 消费组拉取日志，并使用 Splunk HEC 或 Syslog 将数据推送到 SIEM，实现云上日志与本地安全分析平台的集成。

业务背景

企业通常将安全信息和事件管理（SIEM）平台（如 Splunk、QRadar）部署在本地数据中心，且不向公网暴露接收端口以保证安全。当业务上云后，云上资源产生的日志数据又需要纳入本地 SIEM 进行统一的监控、审计和威胁分析。因此在不降低本地系统安全性的前提下，需要建立一条从日志服务到本地 SIEM 的日志投递管道，以实现云上日志的投递。

投递流程

在数据投递场景中，建议采用日志服务消费组来实现实时消费，并利用 Splunk 的 API（HTTP事件收集，HEC）或 Syslog over TCP/TLS 将日志传输至 SIEM。

核心逻辑

日志拉取：基于消费组构建程序，从日志服务拉取数据。此机制支持并发消费和故障转移。
- 并发与吞吐
  - 可通过多次启动程序来实现并发效果。多个消费者属于同一消费组，且名称均不相同（消费者名以进程ID为后缀）。
  - 一个分区（Shard）只能被一个消费者消费，因此并发上限为Shard数量。例如一个日志库有10个分区，那么最多有10个消费者同时消费。
  - 在理想网络条件下：
    - 单个消费者（约占用20%单核CPU）可达10 MB/s原始日志消费速率。
    - 10个消费者可消费100 MB/s原始日志。
- 高可用性
  - 消费组将检测点（Checkpoint）存储于服务端。
  - 当某一消费者实例终止运行，另一个消费者实例将自动接管并从断点继续消费。因此可在不同机器上启动消费者，当一台机器故障的情况下，其他机器上的消费者便可以自动从断点继续消费。
  - 可在不同机器启动大于Shard数量的消费者以作备用。
数据转发：程序收到日志后，根据配置进行格式化，并发送到本地 SIEM。

准备工作

创建RAM用户及授权：该RAM需要拥有AliyunLogFullAccess的权限。
网络要求：程序所在机器需要能访问日志服务Endpoint域名，且与SIEM 处于相同网络环境。
- Endpoint域名获取方式：
  1. 登录日志服务控制台，在Project列表中，单击目标Project。
  2. 单击Project名称右侧的进入项目概览页面。
  3. 在访问域名中复制公网域名。Endpoint域名为https://+公网域名。
环境要求：准备Python 3运行环境，并安装Python sdk。
1. 安装日志服务Python sdk：pip install -U aliyun-log-python-sdk。
2. 验证安装结果：pip show aliyun-log-python-sdk，返回如下信息表示安装成功。
```
Name: aliyun-log-python-sdk
Version: 0.9.12
Summary: Aliyun log service Python client SDK
Home-page: https://github.com/aliyun/aliyun-log-python-sdk
Author: Aliyun
```

实施步骤

步骤一：应用程序准备

日志服务提供Splunk HEC 与 Syslog 两种投递方式，请选择对应程序示例进行配置。

Splunk HEC ：HTTP 事件收集器 (HEC) 基于Token，通过 HTTP 以高效安全的方式将多种数据格式的日志直接发送到 Splunk。
Syslog：常见的日志通道，兼容大多数SIEM，支持文本格式。

Splunk HEC

投递日志数据至Splunk时，可以参考sync_data.py进行配置，代码主要由三部分内容组成：

main()方法：主程序控制逻辑。

get_option() 方法：消费配置项。

基本配置项：包括日志服务连接配置和消费组配置。
消费组的高级选项：性能调参，不推荐修改。
SIEM（Splunk）相关参数与选项。

若在数据投递过程中涉及数据清洗（如行过滤、列裁剪和数据规整等）时，可以通过添加SPL语句添加规则，参考如下：

# SPL 语句
    query = "* | where instance_id in ('instance-1', 'instance-2')"
# 基于规则创建消费，相比普通消费在参数列表最后增加了参数 query
    option = LogHubConfig(endpoint, accessKeyId, accessKey, project, logstore, consumer_group, consumer_name,
                          cursor_position=CursorPosition.SPECIAL_TIMER_CURSOR,
                          cursor_start_time=cursor_start_time,
                          heartbeat_interval=heartbeat_interval,
                          data_fetch_interval=data_fetch_interval,
                          query=query)

SyncData(ConsumerProcessorBase)：内容包含如何从日志服务获取数据并投递到Splunk，请仔细阅读代码中相关注释并根据需求调整。

完整代码如下：

sync_data.py

# -*- coding: utf-8 -*-
import os
import logging
from logging.handlers import RotatingFileHandler
from aliyun.log.consumer import *
from aliyun.log.pulllog_response import PullLogResponse
from multiprocessing import current_process
import json
import socket
import requests

# 配置程序日志文件，以便后续测试或者诊断问题
root = logging.getLogger()
handler = RotatingFileHandler("{0}_{1}.log".format(os.path.basename(__file__), current_process().pid), maxBytes=100*1024*1024, backupCount=5)
handler.setFormatter(logging.Formatter(fmt='[%(asctime)s] - [%(threadName)s] - {%(module)s:%(funcName)s:%(lineno)d} %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S'))
root.setLevel(logging.INFO)
root.addHandler(handler)
root.addHandler(logging.StreamHandler())

logger = logging.getLogger(__name__)


class SyncData(ConsumerProcessorBase):
    """
    这个消费者从日志服务消费数据并发送给Splunk
    """
    def __init__(self, splunk_setting=None):
        
        """初始化并验证Splunk连通性"""
        super(SyncData, self).__init__()   # remember to call base's init

        assert splunk_setting, ValueError("You need to configure settings of remote target")
        assert isinstance(splunk_setting, dict), ValueError("The settings should be dict to include necessary address and confidentials.")

        self.option = splunk_setting
        self.timeout = self.option.get("timeout", 120)

        # 测试Splunk连通性
        s = socket.socket()
        s.settimeout(self.timeout)
        s.connect((self.option["host"], self.option['port']))

        self.r = requests.session()
        self.r.max_redirects = 1
        self.r.verify = self.option.get("ssl_verify", True)
        self.r.headers['Authorization'] = "Splunk {}".format(self.option['token'])
        self.url = "{0}://{1}:{2}/services/collector".format("http" if not self.option.get('https') else "https", self.option['host'], self.option['port'])

        self.default_fields = {}
        if self.option.get("sourcetype"):
            self.default_fields['sourcetype'] = self.option.get("sourcetype")
        if self.option.get("source"):
            self.default_fields['source'] = self.option.get("source")
        if self.option.get("index"):
            self.default_fields['index'] = self.option.get("index")

    def process(self, log_groups, check_point_tracker):
        logs = PullLogResponse.loggroups_to_flattern_list(log_groups, time_as_str=True, decode_bytes=True)
        logger.info("Get data from shard {0}, log count: {1}".format(self.shard_id, len(logs)))
        for log in logs:
            # 修改此处代码：替换为实际的日志发送到远端的同步处理代码
            # 日志格式为字典类型，示例如下（注意：所有字符串必须为Unicode编码）:
            #    Python3: {"__time__": "12312312", "__topic__": "topic", "field1": "value1", "field2": "value2"}
            event = {}
            event.update(self.default_fields)
            event['time'] = log[u'__time__']
            del log['__time__']

            json_topic = {"actiontrail_audit_event": ["event"] }
            topic = log.get("__topic__", "")
            if topic in json_topic:
                try:
                    for field in json_topic[topic]:
                        log[field] = json.loads(log[field])
                except Exception as ex:
                    pass
            event['event'] = json.dumps(log)

            data = json.dumps(event, sort_keys=True)

            try:
                req = self.r.post(self.url, data=data, timeout=self.timeout)
                req.raise_for_status()
            except Exception as err:
                logger.debug("Failed to connect to remote Splunk server ({0}). Exception: {1}".format(self.url, err))
                raise err

                #根据需要，添加一些重试或者报告。

        logger.info("Complete send data to remote")

        self.save_checkpoint(check_point_tracker)


def get_option():
    ##########################
    # 基本配置项
    ##########################

    # 从环境变量中加载日志服务参数与选项
    accessKeyId = os.environ.get('SLS_AK_ID', '')
    accessKey = os.environ.get('SLS_AK_KEY', '')
    endpoint = os.environ.get('SLS_ENDPOINT', '')
    project = os.environ.get('SLS_PROJECT', '')
    logstore = os.environ.get('SLS_LOGSTORE', '')
    consumer_group = os.environ.get('SLS_CG', '')

    assert endpoint and accessKeyId and accessKey and project and logstore and consumer_group, \
        ValueError("endpoint/access_id/key/project/logstore/consumer_group/name cannot be empty")

    ##########################
    # 消费组的高级选项
    ##########################

    # 一般不建议修改消费者名称，尤其是需要进行并发消费时。
    consumer_name = "{0}-{1}".format(consumer_group, current_process().pid)

    # 消费的起点。这个参数在首次运行程序的时候有效，后续再次运行时将从上一次消费的保存点继续消费。
    # 可以使用“begin”、“end”，或者特定的ISO时间格式。
    cursor_start_time = "2018-12-26 0:0:0"

    # 心跳时长，当服务器在2倍时间内没有收到特定Shard的心跳报告时，服务器会认为对应消费者离线并重新调配任务。
    # 当网络环境不佳时，不建议将时长设置的比较小。
    heartbeat_interval = 20

    # 消费数据的最大间隔，如果数据生成的速度很快，不需要调整这个参数
    data_fetch_interval = 1
    
    # 构建一个消费组和消费者
    option = LogHubConfig(endpoint, accessKeyId, accessKey, project, logstore, consumer_group, consumer_name,
                          cursor_position=CursorPosition.SPECIAL_TIMER_CURSOR,
                          cursor_start_time=cursor_start_time,
                          heartbeat_interval=heartbeat_interval,
                          data_fetch_interval=data_fetch_interval)
"""
    基于规则构建消费者时可使用如下代码:
    自定义 SPL 语句
    query = "* | where instance_id in ('instance-1', 'instance-2')"
    基于规则构建消费，相比普通消费在参数列表最后增加了参数 query
    option = LogHubConfig(endpoint, accessKeyId, accessKey, project, logstore, consumer_group, consumer_name,
                          cursor_position=CursorPosition.SPECIAL_TIMER_CURSOR,
                          cursor_start_time=cursor_start_time,
                          heartbeat_interval=heartbeat_interval,
                          data_fetch_interval=data_fetch_interval,
                          query=query)
    """

    # Splunk选项
    settings = {
                "host": "1.2.3.4",
                "port": 80,
                "token": "a0*****123",
                'https': False,              # 可选, bool
                'timeout': 120,             # 可选, int
                'ssl_verify': True,         # 可选, bool
                "sourcetype": "",            # 可选, sourcetype
                "index": "",                # 可选, index
                "source": "",               # 可选, source
            }

    return  option, settings

#主程序控制逻辑 
def main():
    option, settings = get_option()

    logger.info("*** start to consume data...")
    worker = ConsumerWorker(SyncData, option, args=(settings,) )
    worker.start(join=True)

if __name__ == '__main__':
    main()

Syslog

Syslog主要基于RFC5424和RFC3164定义相关日志格式规范，推荐使用RFC5424协议。理论上TCP和UDP都支持Syslog，可以较好的保证数据传输稳定性，RFC5424协议也定义了TLS的安全传输层，当SIEM支持TCP通道或者TLS通道时建议优先使用。

当需要投递日志数据至SIEM时可以参考sync_data.py进行配置，代码主要由三部分内容组成：

main()方法：主程序控制逻辑。
get_monitor_option() 方法：消费配置项。
- 基本配置项：包括日志服务连接配置和消费组配置。
- 消费组的高级选项：性能调参，不推荐修改。
- SIEM的Syslog server相关参数与选项。
  - Syslog facility：程序组件，此处选择syslogclient.FAC_USER作为默认组件。
  - Syslog severity：日志级别，可根据需求设置指定内容的日志级别。此处选择syslogclient.SEV_INFO。
  - 若SIEM支持基于TCP或TLS的Syslog通道，请配置proto为TLS及配置正确的SSL证书。
SyncData(ConsumerProcessorBase)：内容包含如何从日志服务获取数据投递到SIEM Syslog服务器，请仔细阅读代码中相关注释并根据需求调整。

完整代码如下：

sync_data.py

# -*- coding: utf-8 -*-

import os
import logging
from logging.handlers import RotatingFileHandler
from aliyun.log.consumer import *
from aliyun.log.pulllog_response import PullLogResponse
from multiprocessing import current_process
import aliyun.log.ext.syslogclient as syslogclient
from aliyun.log.ext.syslogclient import SyslogClientRFC5424 as SyslogClient
import six
from datetime import datetime

# 配置程序日志文件，以便后续测试或者诊断问题
root = logging.getLogger()
handler = RotatingFileHandler("{0}_{1}.log".format(os.path.basename(__file__), current_process().pid), maxBytes=100*1024*1024, backupCount=5)
handler.setFormatter(logging.Formatter(fmt='[%(asctime)s] - [%(threadName)s] - {%(module)s:%(funcName)s:%(lineno)d} %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S'))
root.setLevel(logging.INFO)
root.addHandler(handler)
root.addHandler(logging.StreamHandler())

logger = logging.getLogger(__name__)


class SyncData(ConsumerProcessorBase):
    """
   消费者从日志服务消费数据并发送给Syslog server
    """
    def __init__(self, target_setting=None):
        """
        初始化并验证Syslog server连通性
        """

        super(SyncData, self).__init__()   # remember to call base's init

        assert target_setting, ValueError("You need to configure settings of remote target")
        assert isinstance(target_setting, dict), ValueError("The settings should be dict to include necessary address and confidentials.")

        self.option = target_setting
        self.protocol = self.option['protocol']
        self.timeout = int(self.option.get('timeout', 120))
        self.sep = self.option.get('sep', "||")
        self.host = self.option["host"]
        self.port = int(self.option.get('port', 514))
        self.cert_path=self.option.get('cert_path', None)

        # 测试连通性 
        with SyslogClient(self.host, self.port, proto=self.protocol, timeout=self.timeout, cert_path=self.cert_path) as client:
            pass

    def process(self, log_groups, check_point_tracker):
        logs = PullLogResponse.loggroups_to_flattern_list(log_groups, time_as_str=True, decode_bytes=True)
        logger.info("Get data from shard {0}, log count: {1}".format(self.shard_id, len(logs)))

        try:
            with SyslogClient(self.host, self.port, proto=self.protocol, timeout=self.timeout, cert_path=self.cert_path) as client:
                for log in logs:
                    # 将日志发送到远端的同步代码置于此处
                    # 日志格式为字典类型，示例如下（注意：所有字符串必须为Unicode编码）:
                    #    Python2: {"__time__": "12312312", "__topic__": "topic", u"field1": u"value1", u"field2": u"value2"}
                    #    Python3: {"__time__": "12312312", "__topic__": "topic", "field1": "value1", "field2": "value2"}
                
                    timestamp = datetime.fromtimestamp(int(log[u'__time__']))
                    del log['__time__']

                    io = six.StringIO()
                    first = True
                    # 可以根据需要修改格式化内容，这里使用Key=Value传输，并使用默认的双竖线（||）进行分割
                    for k, v in six.iteritems(log):
                        io.write("{0}{1}={2}".format(self.sep, k, v))

                    data = io.getvalue()
                    # 可以根据需要修改facility或者severity
                    client.log(data, facility=self.option.get("facility", None), severity=self.option.get("severity", None), timestamp=timestamp, program=self.option.get("tag", None), hostname=self.option.get("hostname", None))

        except Exception as err:
            logger.debug("Failed to connect to remote syslog server ({0}). Exception: {1}".format(self.option, err))
            # 需要添加一些错误处理的代码，例如重试或者通知等 
            raise err

        logger.info("Complete send data to remote")

        self.save_checkpoint(check_point_tracker)


def get_monitor_option():
    ##########################
    # 基本配置项
    ##########################

    # 从环境变量中加载日志服务参数与选项
    endpoint = os.environ.get('SLS_ENDPOINT', '')
    accessKeyId = os.environ.get('SLS_AK_ID', '')
    accessKey = os.environ.get('SLS_AK_KEY', '')
    project = os.environ.get('SLS_PROJECT', '')
    logstore = os.environ.get('SLS_LOGSTORE', '')
    consumer_group = os.environ.get('SLS_CG', '')

    assert endpoint and accessKeyId and accessKey and project and logstore and consumer_group, \
        ValueError("endpoint/access_id/key/project/logstore/consumer_group/name cannot be empty")

    ##########################
    # 消费组的高级选项
    ##########################

    # 一般不建议修改消费者名称，尤其是需要进行并发消费时。
    consumer_name = "{0}-{1}".format(consumer_group, current_process().pid)

    # 消费的起点。这个参数在首次运行程序的时候有效，后续再次运行时将从上一次消费的保存点继续消费。
    # 可以使用“begin”、“end”，或者特定的ISO时间格式。
    cursor_start_time = "2019-1-1 0:0:0+8:00"

    # 心跳时长，当服务器在2倍时间内没有收到特定Shard的心跳报告时，服务器会认为对应消费者离线并重新调配任务。
    # 当网络环境不佳时，不建议将时长设置的比较小。
    heartbeat_interval = 20

    # 消费数据的最大间隔，如果数据生成的速度很快，不需要调整这个参数
    data_fetch_interval = 1

    # 构建一个消费组和消费者
    option = LogHubConfig(endpoint, accessKeyId, accessKey, project, logstore, consumer_group, consumer_name,
                          cursor_position=CursorPosition.SPECIAL_TIMER_CURSOR,
                          cursor_start_time=cursor_start_time,
                          heartbeat_interval=heartbeat_interval,
                          data_fetch_interval=data_fetch_interval)

    # Syslog server相关参数与选项
    settings = {
                "host": "1.2.3.4", # 必选
                "port": 514,       # 必选, 端口
                "protocol": "tcp", # 必选, TCP、UDP或TLS（仅Python3）
                "sep": "||",      # 必选, key=value键值对的分隔符，这里用双竖线（||）分隔
                "cert_path": None,  # 可选，TLS的证书位置
                "timeout": 120,   # 可选，超时时间，默认120秒
                "facility": syslogclient.FAC_USER,  # 可选，可以参考其他syslogclient.FAC_*的值
                "severity": syslogclient.SEV_INFO,  # 可选，可以参考其他syslogclient.SEV_*的值
                "hostname": None, # 可选，机器名，默认选择本机机器名
                "tag": None # 可选，标签，默认是短划线（-）
    }

    return option, settings

#主程序控制逻辑
def main():
    option, settings = get_monitor_option()

    logger.info("*** start to consume data...")
    worker = ConsumerWorker(SyncData, option, args=(settings,) )
    worker.start(join=True)


if __name__ == '__main__':
    main()

步骤二：配置环境变量

程序配置完成后，进行表格中系统环境变量配置。

环境变量名	取值	示例
SLS_ENDPOINT	登录日志服务控制台，在Project列表中，单击目标Project。单击Project名称右侧的进入项目概览页面。在访问域名中复制公网域名，拼接为`https://`+公网域名。若Endpoint前缀配置为`https://`，如`https://cn-beijing.log.aliyuncs.com`，则程序自动使用HTTPS加密与日志服务连接。服务器证书`*.aliyuncs.com`由GlobalSign签发，一般机器会自动信任此证书。若机器不信任此证书，通过Certificate installation下载并安装。	`https://cn-beijing.log.aliyuncs.com`
SLS_PROJECT	在日志服务控制台，复制目标Project名称。	my-sls-project-one
SLS_LOGSTORE	在日志服务控制台，复制目标Logstore名称。	my-sls-logstore-a1
SLS_AK_ID	建议使用RAM账号的AccessKey ID。重要阿里云账号的AccessKey拥有所有API的访问权限，建议使用RAM用户的AccessKey进行API访问或日常运维。强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码中，否则可能导致AccessKey泄露，威胁您账号下所有资源的安全。	L***ky
SLS_AK_KEY	建议使用RAM账号的AccessKey Secret。	x***Xl
SLS_CG	消费组名，可以简单命名为"syc_data"，若消费组不存在，程序会自动创建。	syc_data

步骤三：启动并验证

启动多消费者进行并发消费，支持的最大并发数等于总 Shard 数。

# 启动第一个消费者进程
nohup python3 sync_data.py &
# 启动第二个消费者进程
nohup python3 sync_data.py &

在日志服务控制台查看消费组状态。
1. 在Project列表区域，单击目标Project。在日志存储 > 日志库页签中，单击目标Logstore的图标，然后单击数据消费的图标。
2. 在消费组列表中，单击目标消费组，在Consumer Group状态页面，查看每个Shard消费数据的客户端和时间。
通过云监控查看消费组延迟情况并配置告警。

常见问题

出现`ConsumerGroupQuotaExceed`错误

此错误表示超出限制，单个日志库（Logstore）配置消费组上限为30个，请在日志服务控制台删除无用消费组。