Use HTTPS to send logs to an SIEM system

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()

• Configurations
  • The log file of the program: This file is used to debug the program or diagnose any exceptions that may occur.
  • The basic configurations of the program: These include the configurations used to connect to Log Service and those of the consumer group.
  • The advanced configurations of the consumer group: These configurations are used to debug performance. We recommend that you do not modify them.
  • The parameters and options related to the SIEM (Splunk).
• A code example of the program

  You must follow the annotations of the code to customize it to meet your needs.

  #encoding: utf8
  import os
  import logging
  from logging.handlers import RotatingFileHandler
  
  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__)
  
  def get_option():
      ##########################
      # Basic options
      ##########################
  
      # Load parameters and options of SLS from environment variables.
      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', '')
  
      # This parameter specifies the starting time when data consumption is initialized.
      # You can use begin or end when you set this parameter. The time can be set to use a specific ISO format.
      cursor_start_time = "2018-12-26 0:0:0"
  
      ##########################
      # Advanced options
      ##########################
  
      # Do not configure the consumer name, especially when you need to run this program in parallel.
      consumer_name = "{0}-{1}".format(consumer_group, current_process().pid)
  
      # Set the heartbeat interval. If the server does not receive the heartbeat report from the specified shard within two times of the specified interval, the server determines the consumer is off line and will reassign its task to another consumer.
      # We recommend that you set a greater interval when the network provides low performance.
      heartbeat_interval = 20
  
      # The maximum interval between two times of data consumption. If data are generated quickly, do not adjust this setting.
      data_fetch_interval = 1
  
      # Create a consumer group that contains a consumer.
      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)
  
      # Splunk options
      settings = {
                  "host": "10.1.2.3",
                  "port": 80,
                  "token": "a023nsdu123123123",
                  'https': False,              # optional. bool
                  'timeout': 120,             # optional. int
                  'ssl_verify': True,         # optional. bool
                  "sourcetype": "",            # optional.
                  "index": "",                # optional.
                  "source": "",               # optional.
              }
  
      return option, settings

from aliyun.log.consumer import *
from aliyun.log.pulllog_response import PullLogResponse
from multiprocessing import current_process
import time
import json
import socket
import requests

class SyncData(ConsumerProcessorBase):
    """
    This consumer obtains data from Log Service and send the data to Splunk.
    """
    def __init__(self, splunk_setting):
      """Initiate the Splunk server and its network connectivity."""
        super(SyncData, self).__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)

        # Test Splunk connectivity.
        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/event".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:
            # Send data to Splunk.
            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}", self.url, err)

                    # TODO: Add the code to process errors. For example, you can add the code to retry or send a notification in response to an error.
            raise err

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

        self.save_checkpoint(check_point_tracker)

export SLS_ENDPOINT=<Endpoint of your region>
export SLS_AK_ID=<YOUR AK ID>
export SLS_AK_KEY=<YOUR AK KEY>
export SLS_PROJECT=<SLS Project Name>
export SLS_LOGSTORE=<SLS Logstore Name>
export SLS_CG=<SLS consumer group name. You can set it to syc_data.>

pypy3 sync_data.py

exeuctor, options, settings = get_option()

    logger.info("*** start to consume data...")
    workers = []

    for option in options:
        worker = ConsumerWorker(SyncData, option, args=(settings,) )
        workers.append(worker)
        worker.start()

    try:
        for i, worker in enumerate(workers):
            while worker.is_alive():
                worker.join(timeout=60)
            logger.info("worker project: {0} logstore: {1} exit unexpected, try to shutdown it".format(
                options[i].project, options[i].logstore))
            worker.shutdown()
    except KeyboardInterrupt:
        logger.info("*** try to exit **** ")
        for worker in workers:
            worker.shutdown()

        # wait for all workers to shutdown before shutting down executor
        for worker in workers:
            while worker.is_alive():
                worker.join(timeout=60)

    exeuctor.shutdown()

• You can log on to the Log Service console to view the status of a consumer group. For more information, see View consumer group status.
• You can log on to the CloudMonitor to view delays associated with consumer groups and configure corresponding alarms. For more information, see Consumer group delay.

nohup pypy3 sync_data.py &
nohup pypy3 sync_data.py &
nohup pypy3 sync_data.py &
...

• The bandwidth and rate to receive data is not limited.
• PyPy3 is used to run the following example codes.
• One consumer uses about 10% resources of a single-core CPU.