forked from localstack/localstack
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathlocalstack.py
More file actions
446 lines (344 loc) · 12.9 KB
/
localstack.py
File metadata and controls
446 lines (344 loc) · 12.9 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
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
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
import json
import os
import sys
from typing import Dict, Optional, TypedDict
import click
from localstack import __version__
from .console import BANNER, console
from .plugin import LocalstackCli, load_cli_plugins
def create_with_plugins() -> LocalstackCli:
"""
Creates a LocalstackCli instance with all cli plugins loaded.
:return: a LocalstackCli instance
"""
cli = LocalstackCli()
cli.group = localstack
load_cli_plugins(cli)
return cli
def _setup_cli_debug():
from localstack import config
from localstack.utils.bootstrap import setup_logging
config.DEBUG = True
os.environ["DEBUG"] = "1"
setup_logging()
@click.group(name="localstack", help="The LocalStack Command Line Interface (CLI)")
@click.version_option(version=__version__, message="%(version)s")
@click.option("--debug", is_flag=True, help="Enable CLI debugging mode")
@click.option("--profile", type=str, help="Set the configuration profile")
def localstack(debug, profile):
if profile:
os.environ["CONFIG_PROFILE"] = profile
if debug:
_setup_cli_debug()
@localstack.group(name="config", help="Inspect your LocalStack configuration")
def localstack_config():
pass
@localstack.group(
name="status",
help="Print status information about the LocalStack runtime",
invoke_without_command=True,
)
@click.pass_context
def localstack_status(ctx):
if ctx.invoked_subcommand is None:
ctx.invoke(localstack_status.get_command(ctx, "docker"))
@localstack_status.command(
name="docker", help="Query information about the LocalStack Docker image and runtime"
)
@click.option("--format", type=click.Choice(["table", "plain", "dict", "json"]), default="table")
def cmd_status_docker(format):
with console.status("Querying Docker status"):
print_docker_status(format)
@localstack_status.command(name="services", help="Query information about running services")
@click.option("--format", type=click.Choice(["table", "plain", "dict", "json"]), default="table")
def cmd_status_services(format):
import requests
from localstack import config
url = config.get_edge_url()
try:
health = requests.get(f"{url}/health")
doc = health.json()
services = doc.get("services", [])
if format == "table":
print_service_table(services)
if format == "plain":
for service, status in services.items():
console.print(f"{service}={status}")
if format == "dict":
console.print(services)
if format == "json":
console.print(json.dumps(services))
except requests.ConnectionError:
error = f"could not connect to LocalStack health endpoint at {url}"
print_error(format, error)
if config.DEBUG:
console.print_exception()
sys.exit(1)
@localstack.command(name="start", help="Start LocalStack")
@click.option("--docker", is_flag=True, help="Start LocalStack in a docker container (default)")
@click.option("--host", is_flag=True, help="Start LocalStack directly on the host")
@click.option("--no-banner", is_flag=True, help="Disable LocalStack banner", default=False)
@click.option(
"-d", "--detached", is_flag=True, help="Start LocalStack in the background", default=False
)
def cmd_start(docker: bool, host: bool, no_banner: bool, detached: bool):
if docker and host:
raise click.ClickException("Please specify either --docker or --host")
if host and detached:
raise click.ClickException("Cannot start detached in host mode")
if not no_banner:
print_banner()
print_version()
console.line()
from localstack.utils import bootstrap
if not no_banner:
if host:
console.log("starting LocalStack in host mode :laptop_computer:")
else:
console.log("starting LocalStack in Docker mode :whale:")
bootstrap.prepare_host()
if not no_banner and not detached:
console.rule("LocalStack Runtime Log (press [bold][yellow]CTRL-C[/yellow][/bold] to quit)")
if host:
bootstrap.start_infra_locally()
else:
if detached:
bootstrap.start_infra_in_docker_detached(console)
else:
bootstrap.start_infra_in_docker()
@localstack.command(name="stop", help="Stop the running LocalStack container")
def cmd_stop():
from localstack import config
from localstack.utils.docker_utils import DOCKER_CLIENT
from ..utils.container_utils.container_client import NoSuchContainer
container_name = config.MAIN_CONTAINER_NAME
try:
DOCKER_CLIENT.stop_container(container_name)
console.print("container stopped: %s" % container_name)
except NoSuchContainer:
console.print("no such container: %s" % container_name)
sys.exit(1)
@localstack.command(name="logs", help="Show the logs of the LocalStack container")
@click.option(
"-f",
"--follow",
is_flag=True,
help="Block the terminal and follow the log output",
default=False,
)
def cmd_logs(follow: bool):
from localstack import config
from localstack.utils.bootstrap import LocalstackContainer
from localstack.utils.common import FileListener
from localstack.utils.docker_utils import DOCKER_CLIENT
container_name = config.MAIN_CONTAINER_NAME
logfile = LocalstackContainer(container_name).logfile
if not DOCKER_CLIENT.is_container_running(container_name):
console.print("localstack container not running")
sys.exit(1)
if not os.path.exists(logfile):
console.print("localstack container logfile not found at %s" % logfile)
sys.exit(1)
if follow:
listener = FileListener(logfile, print)
listener.start()
try:
listener.join()
except KeyboardInterrupt:
pass
finally:
listener.close()
else:
with open(logfile) as fd:
for line in fd:
print(line.rstrip("\n\r"))
@localstack.command(name="wait", help="Wait on the LocalStack container to start")
@click.option(
"-t",
"--timeout",
type=float,
help="The amount of time in seconds to wait before raising a timeout error",
default=None,
)
def cmd_wait(timeout: Optional[float] = None):
from localstack.utils.bootstrap import wait_container_is_ready
if not wait_container_is_ready(timeout=timeout):
raise click.ClickException("timeout")
@localstack_config.command(
name="validate", help="Validate your LocalStack configuration (e.g., your docker-compose.yml)"
)
@click.option(
"--file",
default="docker-compose.yml",
type=click.Path(exists=True, file_okay=True, readable=True),
)
def cmd_config_validate(file):
from rich.panel import Panel
from localstack.utils import bootstrap
try:
if bootstrap.validate_localstack_config(file):
console.print("[green]:heavy_check_mark:[/green] config valid")
sys.exit(0)
else:
console.print("[red]:heavy_multiplication_x:[/red] validation error")
sys.exit(1)
except Exception as e:
console.print(Panel(str(e), title="[red]Error[/red]", expand=False))
console.print("[red]:heavy_multiplication_x:[/red] validation error")
sys.exit(1)
@localstack_config.command(name="show", help="Print the current LocalStack config values")
@click.option("--format", type=click.Choice(["table", "plain", "dict", "json"]), default="table")
def cmd_config_show(format):
# TODO: parse values from potential docker-compose file?
from localstack_ext import config as ext_config
from localstack import config
assert config
assert ext_config
if format == "table":
print_config_table()
elif format == "plain":
print_config_pairs()
elif format == "dict":
print_config_dict()
elif format == "json":
print_config_json()
else:
print_config_pairs() # fall back to plain
def print_config_json():
import json
from localstack import config
console.print(json.dumps(dict(config.collect_config_items())))
def print_config_pairs():
from localstack import config
for key, value in config.collect_config_items():
console.print(f"{key}={value}")
def print_config_dict():
from localstack import config
console.print(dict(config.collect_config_items()))
def print_config_table():
from rich.table import Table
from localstack import config
grid = Table(show_header=True)
grid.add_column("Key")
grid.add_column("Value")
for key, value in config.collect_config_items():
grid.add_row(key, str(value))
console.print(grid)
@localstack.command(name="ssh", help="Obtain a shell in the running LocalStack container")
def cmd_ssh():
from localstack import config
from localstack.utils.docker_utils import DOCKER_CLIENT
from localstack.utils.run import run
if not DOCKER_CLIENT.is_container_running(config.MAIN_CONTAINER_NAME):
raise click.ClickException(
'Expected a running container named "%s", but found none' % config.MAIN_CONTAINER_NAME
)
try:
process = run("docker exec -it %s bash" % config.MAIN_CONTAINER_NAME, tty=True)
process.wait()
except KeyboardInterrupt:
pass
# legacy support
@localstack.group(
name="infra",
help="Manipulate LocalStack infrastructure (legacy)",
)
def infra():
pass
@infra.command("start")
@click.pass_context
@click.option("--docker", is_flag=True, help="Start LocalStack in a docker container (default)")
@click.option("--host", is_flag=True, help="Start LocalStack directly on the host")
def cmd_infra_start(ctx, *args, **kwargs):
ctx.invoke(cmd_start, *args, **kwargs)
class DockerStatus(TypedDict, total=False):
running: bool
runtime_version: str
image_tag: str
image_id: str
image_created: str
container_name: Optional[str]
container_ip: Optional[str]
def print_docker_status(format):
from localstack import config
from localstack.utils import docker_utils
from localstack.utils.bootstrap import (
get_docker_image_details,
get_main_container_ip,
get_main_container_name,
get_server_version,
)
img = get_docker_image_details()
cont_name = config.MAIN_CONTAINER_NAME
running = docker_utils.DOCKER_CLIENT.is_container_running(cont_name)
status = DockerStatus(
runtime_version=get_server_version(),
image_tag=img["tag"],
image_id=img["id"],
image_created=img["created"],
running=running,
)
if running:
status["container_name"] = get_main_container_name()
status["container_ip"] = get_main_container_ip()
if format == "dict":
console.print(status)
if format == "table":
print_docker_status_table(status)
if format == "json":
console.print(json.dumps(status))
if format == "plain":
for key, value in status.items():
console.print(f"{key}={value}")
def print_docker_status_table(status: DockerStatus):
from rich.table import Table
grid = Table(show_header=False)
grid.add_column()
grid.add_column()
grid.add_row("Runtime version", f'[bold]{status["runtime_version"]}[/bold]')
grid.add_row(
"Docker image",
f"tag: {status['image_tag']}, "
f"id: {status['image_id']}, "
f":calendar: {status['image_created']}",
)
cont_status = "[bold][red]:heavy_multiplication_x: stopped"
if status["running"]:
cont_status = (
f"[bold][green]:heavy_check_mark: running[/green][/bold] "
f'(name: "[italic]{status["container_name"]}[/italic]", IP: {status["container_ip"]})'
)
grid.add_row("Runtime status", cont_status)
console.print(grid)
def print_service_table(services: Dict[str, str]):
from rich.table import Table
status_display = {
"running": "[green]:heavy_check_mark:[/green] running",
"starting": ":hourglass_flowing_sand: starting",
"available": "[grey]:heavy_check_mark:[/grey] available",
"error": "[red]:heavy_multiplication_x:[/red] error",
}
table = Table()
table.add_column("Service")
table.add_column("Status")
services = list(services.items())
services.sort(key=lambda item: item[0])
for service, status in services:
if status in status_display:
status = status_display[status]
table.add_row(service, status)
console.print(table)
def print_version():
console.print(" :laptop_computer: [bold]LocalStack CLI[/bold] [blue]%s[/blue]" % __version__)
def print_error(format, error):
if format == "table":
symbol = "[bold][red]:heavy_multiplication_x: ERROR[/red][/bold]"
console.print(f"{symbol}: {error}")
if format == "plain":
console.print(f"error={error}")
if format == "dict":
console.print({"error": error})
if format == "json":
console.print(json.dumps({"error": error}))
def print_banner():
print(BANNER)