-
Notifications
You must be signed in to change notification settings - Fork 2
/
albatross.py
72 lines (55 loc) · 2.15 KB
/
albatross.py
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
import logging
import boto3
import json
import os
import urllib.parse as urlparse
from typing import Optional
import constants
import aws
import slack
logger = logging.getLogger()
logger.setLevel(logging.WARNING)
# TODO(vishalkuo): handle try except better
# TODO(vishalkuo): create a generic context with client/resource.
def handle(event, context):
try:
slack_signature = event["headers"]["X-Slack-Signature"]
slack_request_timestamp = event["headers"]["X-Slack-Request-Timestamp"]
if not slack.verify(slack_signature, slack_request_timestamp, event["body"]):
response = {"statusCode": 401, "body": "Unauthorized"}
return response
body = urlparse.parse_qs(event["body"])
if not body["text"] or body["text"][0] not in constants.COMMANDS:
text = f"Invalid command, please try `{constants.COMMANDS}`"
return {"statusCode": 200, "body": json.dumps({"text": text})}
cmd = body["text"][0]
body = ""
client = boto3.client("ec2")
sns = boto3.client("sns")
if cmd == constants.STATUS:
body = _process_status(client)
elif cmd == constants.DOWN:
body = _process_down(client)
elif cmd == constants.UP:
body = _process_up(sns)
response = {"text": body, "response_type": "in_channel"}
return {"statusCode": 200, "body": json.dumps(response)}
except Exception as e:
response = {"statusCode": 500, "body": f"Error: {e}"}
return response
def _process_down(client) -> str:
server = aws.find_devserver(client, non_terminated=True)
if not server:
return "No devserver found"
if server["State"]["Name"] != "running":
return "Server not running"
client.stop_instances(InstanceIds=[server["InstanceId"]])
return "Stopping..."
def _process_status(client) -> str:
server = aws.find_devserver(client, non_terminated=True)
if not server:
return "Not found, please start"
return server["State"]["Name"]
def _process_up(sns) -> str:
sns.publish(TopicArn=constants.SNS_TOPIC, Message=constants.UP)
return "Start request received!"