Amazon Kinesis Agent for Windows を使用して、集中化された Microsoft Exchange サーバーのログ管理

    "Sources": [
        {
            "Id": "MessageTracking-LogsSource",
            "SourceType": "ExchangeLogSource",
            "Directory": "C:\\Program Files\\Microsoft\\Exchange Server\\V15\\TransportRoles\\Logs\\MessageTracking",
            "FileNameFilter": "*.log",
            "TimeZoneKind": "UTC",
            "TimeStampField": "date-time" //Optional.ExchangeLogSource can automatically detect if the TimestampField name is "date-time" or "DateTime".For other names, please specify
        }
    ]

"Sinks": [
        {
            "Id": "MessageTracking-Kinesis-Sink",
            "SinkType": "KinesisStream",
            "Region": "us-west-2",
            "RoleARN": "arn:aws:iam::<another aws account>::role/exch-kinesis-log", // only if logs are sent to Kinesis Data Stream in another account.
            "StreamName": "ex-messagetracking",
            "Format": "json"
        }
    ]

   "Pipes": [
        {
            "Id": "MessageTracking-Kinesis-Pipe",
            "SourceRef": "MessageTracking-LogsSource",
            "SinkRef": "MessageTracking-Kinesis-Sink"
        }
    ]

{
      "date-time": "2018-10-22T10:53:13.404Z",
      "client-ip": "10.00.00.00",
      "client-hostname": "ExchangeServer01",
      "server-ip": "10.00.00.00",
      "server-hostname": "ExchangeServer01",
      "source-context": ";250 2.0.0 OK;ClientSubmitTime:2018-10-22T10:53:10.680Z",
      "connector-id": "Intra-Organization SMTP Send Connector",
      "source": "SMTP",
      "event-id": "SEND",
      "internal-message-id": "157882997807893",
      "message-id": "<6d606502441648f993b7b6d1b8fbb5fb@ExchangeServer01>",
      "network-message-id": "9b3f4489-a158-4126-0d41-08d6380c8f0f",
      "recipient-address": "recipient@Somedomain.com",
      "recipient-status": "250 2.1.5 Recipient OK",
      "sender-address": "sender@Somedomain.com",
      "return-path": "return-path@Somedomain.com",
      ….
    }

import datetime
import logging
import boto3
import os
from aws_requests_auth.aws_auth import AWSRequestsAuth
from elasticsearch import Elasticsearch, RequestsHttpConnection

#Constants Declarations
epoch = datetime.datetime.utcfromtimestamp(0)
session = boto3.Session()
credentials = session.get_credentials().get_frozen_credentials()
es_host = 'search-ex-messagetracking-xxxxxxxxxxxxxxxxxxxxxxxxxx.us-west-2.es.amazonaws.com'

#Main function
def lambda_handler(event, context):
    now_time = datetime.datetime.now()
    query_end_time = unix_time_millis(now_time)
    query_start_time = unix_time_millis(now_time - datetime.timedelta(minutes=15))
          
    awsauth = AWSRequestsAuth(
        aws_access_key=credentials.access_key,
        aws_secret_access_key=credentials.secret_key,
        aws_token=credentials.token,
        aws_host=es_host,
        aws_region=session.region_name,
        aws_service='es'
    )

    es = Elasticsearch(
        hosts=[{'host': es_host, 'port': 443}],
        http_auth=awsauth,
        use_ssl=True,
        verify_certs=True,
        connection_class=RequestsHttpConnection
    )
    
    querybody = construct_agg_query("sender-address.keyword", query_start_time, query_end_time)
    log2CW("TopSenders","NoOfMsgsByTopSender",query_es(es,querybody))

# Function Declarations 

def query_es(es_param,querybody_param):
    es_results = es_param.search(index="timestamp-*", body=querybody_param)
    parsed_value = es_results['aggregations']['3']['buckets']
    print(parsed_value)
    print(parsed_value[0]['doc_count'])
    return parsed_value[0]['doc_count']

def unix_time_millis(dt):
    return int((dt - epoch).total_seconds() * 1000.0) 
    
def construct_agg_query(query_keyword, start_time, end_time):
    return_query_body = {
                            "query": {
                                "bool": {
                                "must": [
                                    {
                                    "match_all": {}
                                    },
                                    {
                                    "range": {
                                        "date-time": {
                                        "gte": start_time,
                                        "lte": end_time,
                                        "format": "epoch_millis"
                                        }
                                    }
                                    }
                                ],
                                "must_not": []
                                }
                            },
                            "size": 0,
                            "_source": {
                                "excludes": []
                            },
                            "aggs": {
                                "3": {
                                "terms": {
                                    "field": query_keyword,
                                    "size": 5,
                                    "order": {
                                    "_count": "desc"
                                    }
                                }
                                }
                            }
                        }
    return return_query_body
    
def log2CW (dimension_value, error_code, error_value):
    cloudwatch = boto3.client('cloudwatch')
    response = cloudwatch.put_metric_data(
        MetricData = [
            {
                'MetricName': error_code,
                'Dimensions': [
                    {
                        'Name': 'TransportService',
                        'Value': dimension_value
                    }
                ],
                'Unit': 'None',
                'Value': error_value
            }
        ],
        Namespace = 'Exchange/TransportService'
    )