ExternalVendorCode/sondehub-infra
1
0
Fork 0
mirror of https://github.com/projecthorus/sondehub-infra.git synced 2025-04-15 14:46:39 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

sqs to elk, sonde to s3

Browse Source
This commit is contained in:
Michaela 2021-05-20 18:43:50 +10:00
parent f3cc33c98a
commit d0b010d3a8
6 changed files with 451 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

77
main.tf
View File

@ -128,6 +128,8 @@ EOF
max_session_duration = 3600
}
resource "aws_iam_role" "sign_socket" {
name = "sign_socket"
assume_role_policy = <<EOF
@ -198,6 +200,26 @@ resource "aws_iam_policy" "IAMManagedPolicy" {
EOF
}
resource "aws_sns_topic" "sonde_telem" {
name = "sonde-telem"
delivery_policy = <<EOF
{
"http": {
"defaultHealthyRetryPolicy": {
"minDelayTarget": 5,
"maxDelayTarget": 30,
"numRetries": 100,
"numMaxDelayRetries": 0,
"numNoDelayRetries": 3,
"numMinDelayRetries": 0,
"backoffFunction": "linear"
},
"disableSubscriptionOverrides": false
}
}
EOF
}
resource "aws_iam_policy" "IAMManagedPolicy2" {
name = "aws-iot-role-es_795847808"
path = "/service-role/"
@ -514,9 +536,6 @@ resource "aws_lambda_function" "LambdaFunction" {
role = aws_iam_role.IAMRole5.arn
runtime = "python3.7"
timeout = 30
tracing_config {
mode = "Active"
}
environment {
variables = {
"IOT_ENDPOINT" = data.aws_iot_endpoint.endpoint.endpoint_address
@ -538,9 +557,6 @@ resource "aws_lambda_function" "station" {
role = aws_iam_role.IAMRole5.arn
runtime = "python3.7"
timeout = 10
tracing_config {
mode = "Active"
}
environment {
variables = {
"IOT_ENDPOINT" = data.aws_iot_endpoint.endpoint.endpoint_address
@ -562,9 +578,7 @@ resource "aws_lambda_function" "get_sondes" {
role = aws_iam_role.IAMRole5.arn
runtime = "python3.7"
timeout = 30
tracing_config {
mode = "Active"
}
environment {
variables = {
"ES" = "es.${local.domain_name}"
@ -586,9 +600,7 @@ resource "aws_lambda_function" "listeners" {
role = aws_iam_role.IAMRole5.arn
runtime = "python3.7"
timeout = 30
tracing_config {
mode = "Active"
}
environment {
variables = {
"ES" = "es.${local.domain_name}"
@ -611,9 +623,7 @@ resource "aws_lambda_function" "datanew" {
role = aws_iam_role.IAMRole5.arn
runtime = "python3.7"
timeout = 30
tracing_config {
mode = "Active"
}
environment {
variables = {
"ES" = "es.${local.domain_name}"
@ -635,9 +645,7 @@ resource "aws_lambda_function" "predictions" {
role = aws_iam_role.IAMRole5.arn
runtime = "python3.7"
timeout = 30
tracing_config {
mode = "Active"
}
environment {
variables = {
"ES" = "es.${local.domain_name}"
@ -659,9 +667,7 @@ resource "aws_lambda_function" "get_telem" {
role = aws_iam_role.IAMRole5.arn
runtime = "python3.7"
timeout = 30
tracing_config {
mode = "Active"
}
environment {
variables = {
"ES" = "es.${local.domain_name}"
@ -683,9 +689,7 @@ resource "aws_lambda_function" "sign_socket" {
role = aws_iam_role.sign_socket.arn
runtime = "python3.7"
timeout = 10
tracing_config {
mode = "Active"
}
environment {
variables = {
"IOT_ENDPOINT" = data.aws_iot_endpoint.endpoint.endpoint_address
@ -712,9 +716,7 @@ resource "aws_lambda_function" "history" {
"ES" = "es.${local.domain_name}"
}
}
tracing_config {
mode = "Active"
}
layers = [
"arn:aws:lambda:us-east-1:${data.aws_caller_identity.current.account_id}:layer:xray-python:1",
"arn:aws:lambda:us-east-1:${data.aws_caller_identity.current.account_id}:layer:iot:3"
@ -1060,12 +1062,9 @@ resource "aws_elasticsearch_domain" "ElasticsearchDomain" {
dedicated_master_count = 3
dedicated_master_enabled = false
dedicated_master_type = "t3.small.elasticsearch"
instance_count = 6
instance_type = "t3.medium.elasticsearch"
zone_awareness_enabled = true
zone_awareness_config {
availability_zone_count = 3
}
instance_count = 1
instance_type = "r5.xlarge.elasticsearch"
zone_awareness_enabled = false
}
cognito_options {
enabled = true
@ -1102,13 +1101,13 @@ EOF
ebs_options {
ebs_enabled = true
volume_type = "gp2"
volume_size = 60
volume_size = 250
}
log_publishing_options {
cloudwatch_log_group_arn = "arn:aws:logs:us-east-1:143841941773:log-group:/aws/aes/domains/sondes-v2/application-logs"
enabled = true
log_type = "ES_APPLICATION_LOGS"
}
log_publishing_options {
cloudwatch_log_group_arn = "arn:aws:logs:us-east-1:143841941773:log-group:/aws/aes/domains/sondes-v2/application-logs"
enabled = true
log_type = "ES_APPLICATION_LOGS"
}
}
data "aws_kms_key" "es" {
key_id = "alias/aws/es"

1
sonde-to-s3/README.md Normal file
View File

@ -0,0 +1 @@
Lambda function which reads off SQS queue and processes sonde data to S3.

84
sonde-to-s3/lambda_function.py Normal file
View File

@ -0,0 +1,84 @@
import json
import boto3
import os
import uuid
import hashlib
BUCKET = "sondehub-open-data"
def values_to_hash(payload):
fields = [
"type",
"serial",
"frame",
"datetime",
"lat",
"lon",
"alt",
"subtype",
"temp",
"humidity",
"pressure",
"vel_h",
"vel_v",
"heading",
"sats",
"batt",
"burst_timer",
"xdata"
]
output = ""
for field in fields:
if field in payload:
output += str(payload[field])
return output
s3 = boto3.client('s3')
def lambda_handler(event, context):
payloads = {}
for record in event['Records']:
sns_message = json.loads(record["body"])
payload = json.loads(sns_message["Message"])
body = json.dumps(payload)
id = str(uuid.uuid4())
hash = hashlib.sha256(values_to_hash(payload).encode("utf-8")).hexdigest()
filenames = [
f"date/{payload['datetime']}-{payload['serial']}-{id}.json",
f"serial/{payload['serial']}/{payload['datetime']}-{id}.json",
f"serial-hashed/{payload['serial']}/{payload['datetime']}-{hash}.json"
]
for filename in filenames:
s3.put_object(
Bucket=BUCKET,
Body=body,
Key=filename
)
if __name__ == "__main__":
demo_event = {
"Records": [
{
"messageId": "262d4090-e23b-4907-b677-3c94334dc899",
"receiptHandle": "AQEBL1FXHS4m+Om59KZH9ayxC5VBqDEDh6DgXUZuBhV2uQJS312bhOTpLvptuCCIWaeLkfHU+7NajqV2kTVhnz5lehE/zfQ8OU1jqqm+cHxyul99MxA7K7+C+ww2Ri9KSbgaAgqvZzcLbwpW8rP0MNhrBcIQAE5Pz1urfTZKx1RVnv/XQHbR2ARPwocOzk2yEexa0y2f7FedS4F10gju8Ypp0Zr4DSRb1zUkES3QJGiSJakaO1QJT5npRySjAd0CUSPXw7IDTejolfGkItQG5eMRx0enELTUDv8LPsHJkr7ha3DHNfbvxTtdk406nWFn8U8DW515emp7+Y+AD469OnceIMdVC62GHwrpMkedXzLEH0C8TOXHQ+WuRkhR1dauwKqO",
"body": "{\n \"Type\" : \"Notification\",\n \"MessageId\" : \"65147554-e06d-5324-a87d-2da107fea807\",\n \"TopicArn\" : \"arn:aws:sns:us-east-1:143841941773:sonde-telem\",\n \"Message\" : \"{\\\"software_name\\\":\\\"radiosonde_auto_rx\\\",\\\"software_version\\\":\\\"1.5.1\\\",\\\"uploader_callsign\\\":\\\"BIOWL1\\\",\\\"uploader_position\\\":\\\"52.014417,8.47351\\\",\\\"uploader_antenna\\\":\\\"SirioCX395\\\",\\\"time_received\\\":\\\"2021-04-18T07:52:37.196266Z\\\",\\\"datetime\\\":\\\"2021-04-18T07:52:53.001000Z\\\",\\\"manufacturer\\\":\\\"Vaisala\\\",\\\"type\\\":\\\"RS41\\\",\\\"serial\\\":\\\"meowmeowtest\\\",\\\"subtype\\\":\\\"RS41-SGP\\\",\\\"frame\\\":12781,\\\"lat\\\":50.65064,\\\"lon\\\":6.60805,\\\"alt\\\":2954.44289,\\\"temp\\\":-9.3,\\\"humidity\\\":75.4,\\\"pressure\\\":709.79,\\\"vel_v\\\":-2.85326,\\\"vel_h\\\":8.53055,\\\"heading\\\":236.0122,\\\"sats\\\":9,\\\"batt\\\":2.7,\\\"frequency\\\":405.3,\\\"burst_timer\\\":25423,\\\"snr\\\":12.5,\\\"user-agent\\\":\\\"Amazon CloudFront\\\",\\\"position\\\":\\\"50.65064,6.60805\\\",\\\"upload_time_delta\\\":-0.713689,\\\"uploader_alt\\\":340}\",\n \"Timestamp\" : \"2021-04-18T07:52:51.776Z\",\n \"SignatureVersion\" : \"1\",\n \"Signature\" : \"qXuYwDAGPYYLjKXfDtF69AWKDEhhz9MXlqxO2nBwJ/dgOqNSUZtDPqOYSuge3jVCoTSRY5qGw38gg2G+JnEbJd8SVvp9GRsFre8MKWu8T0obq3rj8S0YAh7dTqi4EILIMmi2KziasCDQlrVuZvCSgPnC+hYF3GByI626QW6m3a4E2igclvbE+O6x6qvVDKwmf/eh+8LRiH1PCrEckiXthnr+qOCiTcstyZoOqMOShJBun9k0DK07+Yf1tYDPSHnqZSIaOvAMSjIKKXfGCkel3SWieO7Zgk7xQuo9Z1bcV8Miu4uEvge4G9HKU3S41zaVcQjYvEhQLxxgd1x3HxXImA==\",\n \"SigningCertURL\" : \"https://sns.us-east-1.amazonaws.com/SimpleNotificationService-010a507c1833636cd94bdb98bd93083a.pem\",\n \"UnsubscribeURL\" : \"https://sns.us-east-1.amazonaws.com/?Action=Unsubscribe&SubscriptionArn=arn:aws:sns:us-east-1:143841941773:sonde-telem:1a52ac41-6e17-43da-bfb6-114577c94ca6\"\n}",
"attributes": {
"ApproximateReceiveCount": "2",
"SentTimestamp": "1618732371814",
"SenderId": "AIDAIT2UOQQY3AUEKVGXU",
"ApproximateFirstReceiveTimestamp": "1618732640317"
},
"messageAttributes": {},
"md5OfMessageAttributes": None,
"md5OfBody": "a0191fc5ea3705340c088e457c31095b",
"eventSource": "aws:sqs",
"eventSourceARN": "arn:aws:sqs:us-east-1:143841941773:to-elk",
"awsRegion": "us-east-1"
}
]
}
lambda_handler(demo_event, {})

52
sqs-to-elk/lambda_function.py Normal file
View File

@ -0,0 +1,52 @@
import json
from botocore.awsrequest import AWSRequest
from botocore.endpoint import URLLib3Session
from botocore.auth import SigV4Auth
import boto3
import botocore.credentials
import os
HOST = os.getenv("ES")
def es_request(payload, path, method):
# get aws creds
session = boto3.Session()
headers = {"Host": HOST, "Content-Type": "application/json"}
request = AWSRequest(
method="POST", url=f"https://{HOST}/{path}", data=payload, headers=headers
)
SigV4Auth(boto3.Session().get_credentials(), "es", "us-east-1").add_auth(request)
session = URLLib3Session()
r = session.send(request.prepare())
return json.loads(r.text)
def lambda_handler(event, context):
payloads = {}
for record in event['Records']:
sns_message = json.loads(record["body"])
payload = json.loads(sns_message["Message"])
index = payload['datetime'][:7]
if index not in payloads: # create index if not exists
payloads[index] = []
payloads[index].append(payload)
for index in payloads:
body=""
for payload in payloads[index]:
body += "{\"index\":{}}\n" + json.dumps(payload) + "\n"
body += "\n"
result = es_request(body, f"telm-{index}/_doc/_bulk", "POST")
if 'errors' in result and result['errors'] == True:
error_types = [x['index']['error']['type'] for x in result['items'] if 'error' in x['index']] # get all the error types
error_types = [a for a in error_types if a != 'mapper_parsing_exception'] # filter out mapper failures since they will never succeed
if error_types:
print(event)
print(result)
raise RuntimeError

133
sqs_to_elk.tf Normal file
View File

@ -0,0 +1,133 @@
data "archive_file" "sqs_to_elk" {
type = "zip"
source_file = "sqs-to-elk/lambda_function.py"
output_path = "${path.module}/build/sqs-to-elk.zip"
}
resource "aws_iam_role" "sqs_to_elk" {
path = "/service-role/"
name = "sqs-to-elk"
assume_role_policy = <<EOF
{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {
"Service": "lambda.amazonaws.com"
},
"Action": "sts:AssumeRole"
}]
}
EOF
max_session_duration = 3600
}
resource "aws_iam_role_policy" "sqs_to_elk" {
name = "sqs_to_elk"
role = aws_iam_role.sqs_to_elk.name
policy = <<EOF
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "logs:CreateLogGroup",
"Resource": "arn:aws:logs:us-east-1:${data.aws_caller_identity.current.account_id}:*"
},
{
"Effect": "Allow",
"Action": [
"logs:CreateLogStream",
"logs:PutLogEvents"
],
"Resource": [
"arn:aws:logs:us-east-1:${data.aws_caller_identity.current.account_id}:log-group:/aws/lambda/*"
]
},
{
"Effect": "Allow",
"Action": "es:*",
"Resource": "*"
},
{
"Effect": "Allow",
"Action": "sqs:*",
"Resource": "*"
}
]
}
EOF
}
resource "aws_lambda_function" "sqs_to_elk" {
function_name = "sqs-to-elk"
handler = "lambda_function.lambda_handler"
filename = "${path.module}/build/sqs-to-elk.zip"
source_code_hash = data.archive_file.sqs_to_elk.output_base64sha256
publish = true
memory_size = 128
role = aws_iam_role.sqs_to_elk.arn
runtime = "python3.8"
timeout = 5
reserved_concurrent_executions = 30
environment {
variables = {
"ES" = aws_route53_record.Route53RecordSet7.fqdn
}
}
}
resource "aws_lambda_event_source_mapping" "sqs_to_elk" {
event_source_arn = aws_sqs_queue.sqs_to_elk.arn
function_name = aws_lambda_function.sqs_to_elk.arn
batch_size = 20
maximum_batching_window_in_seconds = 15
}
resource "aws_sns_topic_subscription" "sqs_to_elk" {
topic_arn = aws_sns_topic.sonde_telem.arn
protocol = "sqs"
endpoint = aws_sqs_queue.sqs_to_elk.arn
}
resource "aws_sqs_queue" "sqs_to_elk" {
name = "to-elk"
receive_wait_time_seconds = 1
message_retention_seconds = 1209600 # 14 days
}
resource "aws_sqs_queue_policy" "sqs_to_elk" {
queue_url = aws_sqs_queue.sqs_to_elk.id
policy = <<EOF
{
"Version": "2008-10-17",
"Id": "__default_policy_ID",
"Statement": [
{
"Sid": "__owner_statement",
"Effect": "Allow",
"Principal": {
"AWS": "arn:aws:iam::${data.aws_caller_identity.current.account_id}:root"
},
"Action": "SQS:*",
"Resource": "${aws_sqs_queue.sqs_to_elk.arn}"
},
{
"Sid": "to-elk",
"Effect": "Allow",
"Principal": {
"AWS": "*"
},
"Action": "SQS:SendMessage",
"Resource": "${aws_sqs_queue.sqs_to_elk.arn}",
"Condition": {
"ArnLike": {
"aws:SourceArn": "${aws_sns_topic.sonde_telem.arn}"
}
}
}
]
}
EOF
}

143
sqs_to_s3.tf Normal file
View File

@ -0,0 +1,143 @@
data "archive_file" "sqs_to_s3" {
type = "zip"
source_file = "sonde-to-s3/lambda_function.py"
output_path = "${path.module}/build/sonde-to-s3.zip"
}
resource "aws_iam_role" "sqs_to_s3" {
path = "/service-role/"
name = "sqs_to_s3"
assume_role_policy = <<EOF
{
"Version": "2012-10-17",
"Statement": [{
"Effect": "Allow",
"Principal": {
"Service": "lambda.amazonaws.com"
},
"Action": "sts:AssumeRole"
}]
}
EOF
max_session_duration = 3600
}
resource "aws_iam_role_policy" "sqs_to_s3" {
name = "sqs_to_s3"
role = aws_iam_role.sqs_to_s3.name
policy = <<EOF
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "logs:CreateLogGroup",
"Resource": "arn:aws:logs:us-east-1:${data.aws_caller_identity.current.account_id}:*"
},
{
"Effect": "Allow",
"Action": [
"logs:CreateLogStream",
"logs:PutLogEvents"
],
"Resource": [
"arn:aws:logs:us-east-1:${data.aws_caller_identity.current.account_id}:log-group:/aws/lambda/*"
]
},
{
"Effect": "Allow",
"Action": [
"ec2:DescribeInstances",
"ec2:CreateNetworkInterface",
"ec2:AttachNetworkInterface",
"ec2:DescribeNetworkInterfaces",
"ec2:DeleteNetworkInterface"
],
"Resource": "*"
},
{
"Effect": "Allow",
"Action": "s3:*",
"Resource": "*"
},
{
"Effect": "Allow",
"Action": "sqs:*",
"Resource": "*"
}
]
}
EOF
}
resource "aws_lambda_function" "sqs_to_s3" {
function_name = "sqs_to_s3"
handler = "lambda_function.lambda_handler"
filename = "${path.module}/build/sonde-to-s3.zip"
source_code_hash = data.archive_file.sqs_to_s3.output_base64sha256
publish = true
memory_size = 128
role = aws_iam_role.sqs_to_s3.arn
runtime = "python3.8"
timeout = 30
reserved_concurrent_executions = 30
vpc_config {
security_group_ids = ["sg-772f357f"]
subnet_ids = ["subnet-5c34ec6d", "subnet-7b1c3836", "subnet-204b052e", "subnet-de4ddeff", "subnet-408d1c1f", "subnet-a7f460c1"]
}
}
resource "aws_lambda_event_source_mapping" "sqs_to_s3" {
event_source_arn = aws_sqs_queue.sqs_to_s3.arn
function_name = aws_lambda_function.sqs_to_s3.arn
batch_size = 40
maximum_batching_window_in_seconds = 15
}
resource "aws_sns_topic_subscription" "sqs_to_s3" {
topic_arn = aws_sns_topic.sonde_telem.arn
protocol = "sqs"
endpoint = aws_sqs_queue.sqs_to_s3.arn
}
resource "aws_sqs_queue" "sqs_to_s3" {
name = "to-s3"
receive_wait_time_seconds = 1
message_retention_seconds = 1209600 # 14 days
}
resource "aws_sqs_queue_policy" "sqs_to_s3" {
queue_url = aws_sqs_queue.sqs_to_s3.id
policy = <<EOF
{
"Version": "2008-10-17",
"Id": "__default_policy_ID",
"Statement": [
{
"Sid": "__owner_statement",
"Effect": "Allow",
"Principal": {
"AWS": "arn:aws:iam::${data.aws_caller_identity.current.account_id}:root"
},
"Action": "SQS:*",
"Resource": "${aws_sqs_queue.sqs_to_s3.arn}"
},
{
"Sid": "to-s3",
"Effect": "Allow",
"Principal": {
"AWS": "*"
},
"Action": "SQS:SendMessage",
"Resource": "${aws_sqs_queue.sqs_to_s3.arn}",
"Condition": {
"ArnLike": {
"aws:SourceArn": "${aws_sns_topic.sonde_telem.arn}"
}
}
}
]
}
EOF
}