-
Notifications
You must be signed in to change notification settings - Fork 6
/
konf.py
executable file
·318 lines (250 loc) · 9.28 KB
/
konf.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
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
#!/usr/bin/env python3
import argparse
import os
import jinja2
import yaml
BASE_DIR = os.getenv("SNAP", ".")
TEMPLATES_DIR = BASE_DIR + "/templates"
# Custom Jinja2 functions
@jinja2.pass_context
def get_site_name(context, add_staging_prefix=False):
"""Return the metadata name to use for the K8s objects."""
domain = context["domain"].split(".")
if add_staging_prefix and context["deployment_env"] == "staging":
# Return the staging name if there is one specified
if "name" in context["data"].get("staging", {}):
return context["data"]["staging"]["name"]
else:
domain.insert(-2, "staging")
elif context["name"]:
# Return the production name if there is one specified
return context["name"]
return "-".join(domain)
@jinja2.pass_context
def get_environment_domain(context):
"""Return the environment domain."""
domain = context["domain"]
if context["deployment_env"] == "staging":
# Return the staging domain if there is one specified
if "domain" in context["data"].get("staging", {}):
domain = context["data"]["staging"]["domain"]
else:
domain = add_environment_prefix(context, domain)
return domain
@jinja2.pass_context
def add_environment_prefix(context, s):
"""Return the domain with the staging prefix if needed."""
domain = s.split(".")
if context["deployment_env"] == "staging":
domain.insert(-2, "staging")
return ".".join(domain)
@jinja2.pass_context
def is_apex_domain(context, s):
"""Check if the given string is an apex/base domain"""
return s.count(".") == 1
class Konf:
def __init__(
self,
values_file,
env,
local_qa,
docker_tag,
database_url,
labels,
overrides,
new_k8s_version,
nginx_proxy_body_size,
):
self.deployment_env = env
self.values_file = values_file
self.local_qa = local_qa
self.docker_tag = docker_tag
self.database_url = database_url
self.overrides = overrides
self.labels = {
key: value for key, value in (label.split("=") for label in labels)
}
self.nginx_proxy_body_size = nginx_proxy_body_size
self.new_k8s_version = new_k8s_version
# Load project data
self.load_values()
if self.database_url:
# Replace database URL in main deployment and in routes
def _replace_database_url(envs, database_url):
for index, env in enumerate(envs):
if "DATABASE_URL" == env["name"]:
envs.pop(index)
break
envs.append({"name": "DATABASE_URL", "value": database_url})
# Replace in top level "env" definition
_replace_database_url(
self.values.get("env", []), self.database_url
)
# Replace in routes "env" definitions
if self.deployment_env in self.values:
routes = self.values[self.deployment_env].get("routes", [])
for route in routes:
_replace_database_url(route.get("env"), self.database_url)
def load_values(self):
raise NotImplementedError
def render(self, template_file):
"""Returns templates rendered."""
# Init Jinja2 environment
template_loader = jinja2.FileSystemLoader(TEMPLATES_DIR)
jinja_env = jinja2.Environment(
loader=template_loader, undefined=jinja2.StrictUndefined
)
# Add custom jinja2 functions and filters
jinja_env.globals["get_site_name"] = get_site_name
jinja_env.globals["get_environment_domain"] = get_environment_domain
jinja_env.filters["add_environment_prefix"] = add_environment_prefix
jinja_env.filters["is_apex_domain"] = is_apex_domain
template = jinja_env.get_template(template_file)
return template.render(
name=self.name,
domain=self.domain,
tag=self.tag,
data=self.values,
labels=self.labels,
namespace=self.namespace,
deployment_env=self.deployment_env,
nginx_proxy_body_size=self.nginx_proxy_body_size,
new_k8s_version=self.new_k8s_version,
)
class KonfCronJob(Konf):
def load_values(self):
"""This reads the cronjob values from the yaml file
Parameters:
values_file (io.TextIOWrapper): project values
env (string): production or staging
local_qa (boolean): Override values for local qa
docker_tag (string): Override docker tag value
"""
self.values = yaml.load(self.values_file, Loader=yaml.FullLoader)
for override in self.overrides:
key, value = override.split("=")
self.values[key] = value
self.name = self.values.get("name")
self.domain = None
# Set deployment environment namespace
self.namespace = self.deployment_env
# QA overrides
if self.local_qa:
self.namespace = "default"
if self.docker_tag:
self.tag = self.docker_tag
def render(self, template_file="cronjob.yaml"):
if self.new_k8s_version:
template_file = "cronjob-new.yaml"
return super(KonfCronJob, self).render(template_file)
class KonfSite(Konf):
def load_values(self):
"""This reads the project values from the yaml file
Parameters:
values_file (io.TextIOWrapper): project values
env (string): production or staging
local_qa (boolean): Override values for local qa
docker_tag (string): Override docker tag value
"""
self.values = yaml.load(self.values_file, Loader=yaml.FullLoader)
# Use staging overrides for demos
if self.deployment_env == "demo":
staging_values = self.values.get("staging", {})
staging_values.pop("routes", None)
staging_values.pop("nginxConfigurationSnippet", None)
staging_values.pop("nginxServerSnippet", None)
self.values.update(staging_values)
for override in self.overrides:
key, value = override.split("=")
self.values[key] = value
self.name = self.values.get("name")
self.domain = self.values["domain"]
# Environment overrides
environment_config = self.values.get(self.deployment_env, {})
# Extend the env section instead of overriding it
if environment_config.get("env") and self.values.get("env"):
env_names = [e["name"] for e in environment_config["env"]]
# If the same value is defined in the general section and
# in the environment section, we discard the general one in
# favour of the environment (production,staging,demo) value
for item in self.values["env"]:
if item["name"] not in env_names:
environment_config["env"].append(item)
self.values.update(environment_config)
# Set deployment environment namespace
self.namespace = self.deployment_env
# QA overrides
if self.local_qa or self.deployment_env == "demo":
self.namespace = "default"
self.values["replicas"] = 1
for route in self.values.get("routes", []):
route.update({"replicas": 1})
if self.docker_tag:
self.tag = self.docker_tag
def render(self, template_file="site.yaml"):
return super(KonfSite, self).render(template_file)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
description="Render kubernetes configurations for each projects"
)
parser.add_argument(
"--type",
type=str,
help="Type of deployment configuration to generate",
choices=["Site", "CronJob"],
default="Site",
dest="konf_type",
)
parser.add_argument(
"--local-qa", action="store_true", default=False, dest="local_qa"
)
parser.add_argument(
"env",
type=str,
help="Deployment environment.",
choices=["production", "staging", "demo"],
)
parser.add_argument(
"values_file",
type=argparse.FileType("r"),
help="Project configuration values",
)
parser.add_argument(
"--tag",
type=str,
help="Docker tag to deploy",
default="latest",
dest="docker_tag",
)
parser.add_argument("--database-url", type=str, default=None)
parser.add_argument(
"-o",
type=str,
nargs="+",
default=[],
dest="overrides",
)
parser.add_argument(
"-l",
type=str,
nargs="+",
default=[],
dest="labels",
)
parser.add_argument(
"--nginx-proxy-body-size",
type=str,
nargs="?",
dest="nginx_proxy_body_size",
)
# TODO: make this the default behavior once we move away from k8s < v1.21
parser.add_argument(
"--new-version",
action="store_true",
default=False,
dest="new_k8s_version",
)
args = vars(parser.parse_args())
konf_type = args.pop("konf_type")
projectConfig = globals()["Konf" + konf_type](**args)
print(projectConfig.render())