| 7 | # GetRecords, GetShardIterator, DescribeStream, ListShards, and ListStreams Actions on your stream in IAM. |
| 8 | |
| 9 | def lambda_handler(event, context): |
| 10 | # TODO implement |
| 11 | |
| 12 | token = os.environ.get("INFLUXDB_TOKEN") |
| 13 | org = os.environ.get("INFLUXDB_ORG") |
| 14 | url = os.environ.get("INFLUXDB_URL") |
| 15 | client = influxdb_client.InfluxDBClient(url=url, token=token, org=org) |
| 16 | |
| 17 | bucket=os.environ.get("INFLUXDB_BUCKET") |
| 18 | |
| 19 | write_api = client.write_api(write_options=SYNCHRONOUS) |
| 20 | |
| 21 | for record in event['Records']: |
| 22 | print(record) |
| 23 | string = base64.decodebytes(record["kinesis"]["data"].encode('utf-8')).decode('utf-8') |
| 24 | data = json.loads(string) |
| 25 | print(data) |
| 26 | |
| 27 | point = ( |
| 28 | Point("event") |
| 29 | .tag("http_event", data['req_path']) |
| 30 | .field("_raw", data['_raw']) |
| 31 | .field("_time", data['_time']) |
| 32 | .field("req_method", data['req_method']) |
| 33 | .field("req_path", data['req_path']) |
| 34 | .field("req_proto", data['req_proto']) |
| 35 | .field("req_flags", data['req_flags']) |
| 36 | .field("resp_status", data['resp_status']) |
| 37 | .field("resp_size", data['resp_size']) |
| 38 | ) |
| 39 | write_api.write(bucket=bucket, org="dev", record=point) |
| 40 | |
| 41 | |
| 42 | return { |
| 43 | 'statusCode': 200, |
| 44 | 'body': json.dumps('Hello from Lambda!') |
| 45 | } |