-
Notifications
You must be signed in to change notification settings - Fork 326
Expand file tree
/
Copy pathlambda_pinpoint_automation.py
More file actions
118 lines (102 loc) · 3.41 KB
/
lambda_pinpoint_automation.py
File metadata and controls
118 lines (102 loc) · 3.41 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
"""
-*- coding: utf-8 -*-
========================
AWS Lambda
========================
Contributor: Chirag Rathod (Srce Cde)
========================
"""
import os
import sys
import json
import ast
import boto3
import logging
import traceback
logger = logging.getLogger()
logger.setLevel(logging.ERROR)
def process_error():
ex_type, ex_value, ex_traceback = sys.exc_info()
traceback_string = traceback.format_exception(ex_type, ex_value, ex_traceback)
error_msg = json.dumps(
{
"errorType": ex_type.__name__,
"errorMessage": str(ex_value),
"stackTrace": traceback_string,
}
)
return error_msg
def notify_recepients(EMAIL_INFO, public_dns="", START_DEMO=False, STOP_DEMO=False):
client = boto3.client("ses")
if START_DEMO:
subject = "Demo instance is running | Information"
body = """
Hello Team,
<br><br>
The demo instance is running at: {}
<br><br>
Thanks
<br><br>
Srce Cde
""".format(
public_dns
)
if STOP_DEMO:
subject = "Demo instance stopped"
body = """
Hello Team,
<br><br>
The demo instance is stopped!
<br><br>
Thanks
<br><br>
Srce Cde
"""
message = {"Subject": {"Data": subject}, "Body": {"Html": {"Data": body}}}
response = client.send_email(
Source=EMAIL_INFO.get("source"),
Destination={"ToAddresses": EMAIL_INFO.get("recipients")},
Message=message,
)
def get_public_dns(ec2_client, INSTANCE_IDS):
public_dns = []
while True:
dns_response = ec2_client.describe_instances(InstanceIds=INSTANCE_IDS)
for reservations in dns_response["Reservations"]:
for info in reservations["Instances"]:
if (
info.get("InstanceId") in INSTANCE_IDS
and info.get("State").get("Name") == "running"
):
public_dns.append(f"http://{info.get('PublicDnsName')}")
if public_dns:
break
return public_dns
def lambda_handler(event, context):
ec2_client = boto3.client("ec2")
try:
INSTANCE_IDS = ast.literal_eval(os.environ["INSTANCE_IDS"])
EMAIL_INFO = ast.literal_eval(os.environ["EMAIL_INFO"])
except Exception as e:
error_msg = process_error()
logger.error(error_msg)
try:
for record in event["Records"]:
body = json.loads(record["Sns"]["Message"])["messageBody"]
if body == "START_DEMO":
# logic to START demo instances
response = ec2_client.start_instances(InstanceIds=INSTANCE_IDS)
public_dns = get_public_dns(ec2_client, INSTANCE_IDS)
notify_recepients(
EMAIL_INFO, public_dns, START_DEMO=True, STOP_DEMO=False
)
elif body == "STOP_DEMO":
# logic to STOP demo instances
response = ec2_client.stop_instances(InstanceIds=INSTANCE_IDS)
notify_recepients(EMAIL_INFO, START_DEMO=False, STOP_DEMO=True)
else:
logger.error("No action performed!")
except Exception as e:
error_msg = process_error()
logger.error(error_msg)
return {"statusCode": 200, "body": json.dumps("Thanks from Srce Cde!")}