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
130
131
132
133
134
135
136
|
import ipaddress
import jinja2
import os
import subprocess
class Shell():
@staticmethod
def run_shell_command(cmd):
p = subprocess.Popen([cmd],
shell=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env={})
out, err = p.communicate()
if p.returncode != 0:
raise Exception("Error running shell command:\n%s\n%s" % (out, err))
return out, err
class Jinja():
@staticmethod
def render(template_name, **kwargs):
cwd = os.path.dirname(os.path.abspath(__file__))
loader = jinja2.FileSystemLoader([".", "/"])
extensions = ["jinja2.ext.with_", "jinja2.ext.loopcontrols"]
environment = jinja2.Environment(trim_blocks = True,
lstrip_blocks = True,
extensions = extensions,
loader = loader)
template_names = ["%s/jinja/%s" % (cwd, template_name)]
template = environment.select_template(template_names)
return template.render(kwargs)
class RawQuery():
@staticmethod
def dict_fetchall(cursor):
rows = cursor.fetchall()
if len(rows) == 0:
return None
else:
columns = [col[0] for col in cursor.description]
return [dict(zip(columns, row)) for row in rows]
@staticmethod
def query_single_value(cursor, query, column_name):
cursor.execute(query)
rows = RawQuery.dict_fetchall(cursor)
if rows == {}:
return None
else:
return rows[0][column_name]
class Address():
@staticmethod
def coalesce(ip, default):
if ip is None:
return default
else:
return ip
@staticmethod
def get_quads(addr):
return list(filter(lambda x: x, str(addr).split("/")[0].split(":")))
@staticmethod
def address_add(ip, index):
return ipaddress.IPv6Address(ip) + index
"""
Assumes a 5-quad 'subnet' as import input. Adds the 6th quad which is always 40a.
"""
@staticmethod
def add_40a(subnet):
return Address.address_add(subnet, 0x040a00000000)
@staticmethod
def next_address(ip, subnet, block_begin_hex):
beginning_address = Address.coalesce(ip, Address.address_add(subnet, block_begin_hex))
if ip is None:
return beginning_address
else:
return Address.address_add(beginning_address, 0x1)
@staticmethod
def add_cidr(subnet, cidr = 64):
subnet_quads = Address.get_quads(subnet)
return "%s::/%s" % (":".join(subnet_quads), cidr)
@staticmethod
def remove_cidr(subnet):
return ipaddress.IPv6Address(str(subnet).split("/")[0])
@staticmethod
def gateway_to_subnet(gateway, numquads = 5):
# The subnet is always the first 5 quads of the gateway.
# TODO: make sure this holds true in MTCE and labs as well as production.
quads = Address.get_quads(gateway)
return ipaddress.IPv6Address("%s::" % ":".join(quads[:numquads]))
@staticmethod
def gateway_to_subnet_new(gateway):
quads = Address.get_quads(gateway)
return ipaddress.IPv6Address("%s:40a::" % ":".join(quads[:5]))
@staticmethod
def subnet_to_4_quads(subnet):
quads = Address.get_quads(gateway)
return ipaddress.IPv6Address("%s::" % ":".join(quads[:4]))
@staticmethod
def strip_brackets(ip):
return ip.replace("[", "").replace("]", "")
class Hostname():
@staticmethod
def make_oam_hostname(ilo_hostname):
return ilo_hostname.replace("-pe0", "-pe2")
"""
Note: this method does not deal with the index at the end of the
hostname. Doing so would require database access. This is the
responsibility of the service classes. See dnsservice.py
"""
@staticmethod
def make_vip_hostname(ilo_hostname, cluster_name):
index = int(cluster_name[-3:])
return "%s-%s" % (ilo_hostname.replace("-pe0", "-pe1")[:-4], f"{index + 0:03}")
@staticmethod
def increment_hostname(hostname):
index = int(hostname[-3:])
return "%s-%s" % (hostname[:-4], f"{index + 1:03}")
|
