1、将本地代码打包上传S3并使用创建的lambda函数从S3中加载代码：
    打包代码：
    python程序运行需要一个独立的运行环境，在本地将所需要的所有依赖包下载打包，一块上传：

    1. # 安装依赖
    2. root@kali-orderplus:~# apt-get -y install python-pip zip
    3. root@kali-orderplus:~# pip install virtualenv
    4. # 配置一个虚拟环境
    5. root@kali-orderplus:~# mkdir -p /var/s3-to-es && cd /var/s3-to-es
    6. root@kali-orderplus:/var/s3-to-es# virtualenv /var/s3-to-es
    7. New python executable in /var/s3-to-es/bin/python
    8. Installing setuptools, pip, wheel...
    9. done.
    10. root@kali-orderplus:/var/s3-to-es# cd /var/s3-to-es && source bin/activate
    11. (s3-to-es) root@kali-orderplus:/var/s3-to-es# pip install requests_aws4auth -t .
    12. (s3-to-es) root@kali-orderplus:/var/s3-to-es# pip freeze > requirements.txt
    13. # python脚本添加执行权限
    14. (s3-to-es) root@kali-orderplus:/var/s3-to-es# chmod 754 s3-to-es.py
    15. # 打包lambda运行环境
    16. (s3-to-es) root@kali-orderplus:/var/s3-to-es# zip -r /var/s3-to-es.zip *
    17. # 通过aws s3工具上传代码压缩包到S3桶中
    18. (s3-to-es) root@kali-orderplus:/var/s3-to-es# aws s3 cp /var/s3-to-es.zip s3://cf.us.xxxx.com/test-cflog/

    python脚本详情：
    carbon (1).png

    1. #!/usr/bin/python3
    2. # -*- coding: utf-8 -*-
    3. from __future__ import print_function
    4. import boto3
    5. import json
    6. import datetime
    7. import gzip
    8. import urllib
    9. import urllib3
    10. import logging
    11. from pprint import pprint
    12. from requests_aws4auth import AWS4Auth
    13. from botocore.vendored import requests
    14. from io import BytesIO
    16. """
    17. Can Override the global variables using Lambda Environment Parameters
    18. """
    19. globalVars  = {}
    20. globalVars['Owner']                 = "Mystique"
    21. globalVars['Environment']           = "Prod"
    22. globalVars['awsRegion']             = "us-west-1"
    23. globalVars['tagName']               = "serverless-s3-to-es-log-ingester"
    24. globalVars['service']               = "es"
    25. globalVars['esIndexPrefix']         = "s3-to-es-"
    26. globalVars['esIndexDocType']        = "s3_to_es_docs"
    27. globalVars['esHosts']               = {
    28.                                         'test': '' ,
    29.                                         'prod': 'https://search-cloudflare-log-t7duyuq4dgxjw4x4fn7jj7wina.us-west-1.es.amazonaws.com'
    30.                                         }
    32. # Initialize Logger
    33. logger = logging.getLogger()
    34. logger.setLevel(logging.INFO)
    36. def indexDocElement(es_Url, awsauth, docData):
    37.     try:
    38.         #headers = { "Content-Type": "application/json" }
    39.         headers = {'Content-type': 'application/json', 'Accept': 'text/plain'}
    40.         resp = requests.post(es_Url, auth=awsauth, headers=headers, json=docData)
    41.         if resp.status_code == 201:
    42.             logger.info('INFO: Successfully inserted element into ES')
    43.         else:
    44.             logger.error('FAILURE: Unable to index element')
    45.     except Exception as e:
    46.         logger.error('ERROR: {0}'.format( str(e) ) )
    47.         logger.error('ERROR: Unable to index line:"{0}"'.format(docData['content']) )
    48.         print (e)
    51. def lambda_handler(event, context):
    52.     s3 = boto3.client('s3')
    53.     credentials = boto3.Session().get_credentials()
    54.     awsauth = AWS4Auth( credentials.access_key, 
    55.                         credentials.secret_key, 
    56.                         globalVars['awsRegion'], 
    57.                         globalVars['service'], 
    58.                         session_token=credentials.token
    59.                     )
    61.     logger.info("Received event: " + json.dumps(event, indent=2))
    63.     try:    
    64.         bucket = event['Records'][0]['s3']['bucket']['name']
    65.         key = urllib.parse.unquote_plus(event['Records'][0]['s3']['object']['key'])
    67.         # Get documet (obj) form S3
    68.         obj = s3.get_object(Bucket=bucket, Key=key)
    70.     except Exception as e:
    71.         logger.error('ERROR: {0}'.format( str(e) ) )
    72.         logger.error('ERROR: Unable able to GET object:{0} from S3 Bucket:{1}. Verify object exists.'.format(key, bucket) )
    74.     if (key.endswith('.gz')) or (key.endswith('.tar.gz')):
    75.         mycontentzip = gzip.GzipFile(fileobj=BytesIO(obj['Body'].read())).read()
    76.         lines = mycontentzip.decode("utf-8").replace("'", '"')
    77.         # print('unziped file')
    78.     else:
    79.         lines = obj['Body'].read().decode("utf-8").replace("'", '"')
    81.     logger.info('SUCCESS: Retreived object from S3')
    83.     # Split (S3 object/Log File) by lines
    84.     lines = lines.splitlines()
    85.     if (isinstance(lines, str)):
    86.         lines = [lines]
    88.     # Index each line to ES Domain
    89.     indexName = globalVars['esIndexPrefix'] + str( datetime.date.today().year ) + '-' + str( datetime.date.today().month )
    90.     es_Url = globalVars['esHosts'].get('prod') + '/' + indexName + '/' + globalVars['esIndexDocType']
    92.     docData = {}
    93.     docData['objectKey']        = str(key)
    94.     docData['createdDate']      = str(obj['LastModified'])
    95.     docData['content_type']     = str(obj['ContentType'])
    96.     docData['content_length']   = str(obj['ContentLength'])
    98.     for line in lines:
    99.         # docData['content'] = str(line)
    100.         docData['content'] = json.loads(line)
    101.         indexDocElement(es_Url, awsauth, docData )
    102.     logger.info('SUCCESS: Successfully indexed the entire doc into ES')
    104. if __name__ == '__main__':
    105.     lambda_handler(None, None)