-
Notifications
You must be signed in to change notification settings - Fork 1
/
instances.py
executable file
·129 lines (107 loc) · 4.04 KB
/
instances.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
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
119
120
121
122
123
124
125
126
127
128
129
#!/usr/bin/env python3
import datetime
import json
from dataclasses import dataclass
from pathlib import Path
import boto3
import click
from summon import read_regions_config, read_aws_defaults
@dataclass
class RunningInstance:
ip_address: str
state: str
coach: str
url: str
creation_date: str
region_name: str
@click.command()
@click.option(
"--region-name",
help=f"the aws region name (default eu-north-1): {', '.join(read_regions_config().keys())}",
default="eu-north-1"
)
@click.option(
"--aws-profile",
default="default",
help="the aws profile, if you dont use the default"
)
@click.option(
"--coach",
help="only show instances owned by this person",
)
def main(region_name, aws_profile, coach=None):
instances = all_instances(region_name, aws_profile)
if coach:
instances = [instance for instance in instances if instance.coach.strip() == coach]
print('\n'.join(print_instances(instances)))
def all_instances(region_name, aws_profile):
url_stem = read_aws_defaults(profile_name=aws_profile)["url_stem"]
instances = []
if region_name == "all":
all_regions = read_regions_config(profile_name=aws_profile).keys()
for region_name in all_regions:
response = get_sammancoach_machines(region_name, aws_profile)
instances.extend(instances_from_response(response, url_stem))
else:
response = get_sammancoach_machines(region_name, aws_profile)
instances.extend(instances_from_response(response, url_stem))
return instances
def print_instances(instances):
rows = []
for instance in instances:
rows.append(print_instance(instance))
return rows
def get_sammancoach_machines(region_name, aws_profile):
session = boto3.Session(profile_name=aws_profile, region_name=region_name)
client = session.client('ec2')
custom_filter = [{
'Name': 'tag:SammanCoach',
'Values': ['*']}]
response = client.describe_instances(Filters=custom_filter)
return response
def instances_from_response(obj, url_stem):
machines = []
for reservation in obj['Reservations']:
for instance in reservation['Instances']:
tags = instance['Tags']
public_ip = instance.get("PublicIpAddress", "not available")
attach_time = try_find_attach_time(instance)
attach_time_short = attach_time.partition('T')[0] if attach_time else '-'
machine_data = {
"State": instance["State"]["Name"],
"IP": public_ip,
"AttachTime": attach_time_short,
"Placement": instance['Placement']['AvailabilityZone'],
}
for machine in tags:
key = machine['Key']
value = machine['Value']
machine_data[key] = value
machines.append(machine_data)
instances = []
for machine in machines:
if url_stem in machine['Name']:
instance = RunningInstance(ip_address=f"{machine['IP']:16}",
state=f"{machine['State']:12}",
coach=f"{machine['SammanCoach']:12}",
url=f"https://{machine['Name']:42}",
creation_date=f"{machine['AttachTime']}",
region_name=machine['Placement'],
)
instances.append(instance)
return instances
def print_instance(instance: RunningInstance):
return f"{instance.ip_address} {instance.state} {instance.coach} {instance.url} {instance.creation_date} {instance.region_name}"
def try_find_attach_time(instance):
try:
time_value = instance["BlockDeviceMappings"][0]["Ebs"]["AttachTime"]
if isinstance(time_value, datetime.datetime):
return time_value.isoformat()
else:
return time_value
except:
return None
if __name__ == '__main__':
import logging
logging.basicConfig(level=logging.INFO)
main()