Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
try:
ensure_experiment_is_valid(experiment)
except ChaosException as x:
logger.error(str(x))
logger.debug(x)
ctx.exit(1)
experiment["dry"] = dry
journal = run_experiment(experiment, settings=settings)
has_deviated = journal.get("deviated", False)
has_failed = journal["status"] != "completed"
with io.open(journal_path, "w") as r:
json.dump(
journal, r, indent=2, ensure_ascii=False, default=encoder)
if journal["status"] == "completed":
notify(settings, RunFlowEvent.RunCompleted, journal)
elif has_failed:
notify(settings, RunFlowEvent.RunFailed, journal)
if has_deviated:
notify(settings, RunFlowEvent.RunDeviated, journal)
if (has_failed or has_deviated) and not no_exit:
ctx.exit(1)
return journal
logging.CRITICAL: ForegroundColors.RED
}
fmt = "%(color)s[%(asctime)s %(levelname)s]%(end_color)s %(message)s"
if verbose:
log_level = logging.DEBUG
fmt = "%(color)s[%(asctime)s %(levelname)s] "\
"[%(module)s:%(lineno)d]%(end_color)s %(message)s"
formatter = LogFormatter(
fmt=fmt, datefmt="%Y-%m-%d %H:%M:%S", colors=colors)
if log_format == 'json':
fmt = "(process) (asctime) (levelname) (module) (lineno) (message)"
if context_id:
fmt = "(context_id) {}".format(fmt)
formatter = jsonlogger.JsonFormatter(
fmt, json_default=encoder, timestamp=True)
# sadly, no other way to specify the name of the default logger publicly
LOGZERO_DEFAULT_LOGGER = logger_name
logger = setup_default_logger(level=log_level, formatter=formatter)
if context_id:
logger.addFilter(ChaosToolkitContextFilter(logger_name, context_id))
if log_file:
# always everything as strings in the log file
logger.setLevel(logging.DEBUG)
fmt = "%(color)s[%(asctime)s %(levelname)s] "\
"[%(module)s:%(lineno)d]%(end_color)s %(message)s"
formatter = LogFormatter(fmt=fmt, datefmt="%Y-%m-%d %H:%M:%S",
colors=colors)
logzero.logfile(log_file, formatter=formatter, mode='a',
loglevel=logging.DEBUG)
no_install: bool = False) -> Discovery:
"""Discover capabilities and experiments."""
settings = load_settings(ctx.obj["settings_path"])
try:
notify(settings, DiscoverFlowEvent.DiscoverStarted, package)
discovery = disco(
package_name=package, discover_system=not no_system_info,
download_and_install=not no_install)
except DiscoveryFailed as err:
notify(settings, DiscoverFlowEvent.DiscoverFailed, package, err)
logger.debug("Failed to discover {}".format(package), exc_info=err)
logger.fatal(str(err))
return
with open(discovery_path, "w") as d:
d.write(json.dumps(discovery, indent=2, default=encoder))
logger.info("Discovery outcome saved in {p}".format(
p=discovery_path))
notify(settings, DiscoverFlowEvent.DiscoverCompleted, discovery)
return discovery