Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self, arguments):
self.log = logging.getLogger(
'{0}.{1}'.format(__name__, self.__class__.__name__)
)
# Pop extra_arguments now so we can log it separately
extra_arguments = arguments.pop('extra_arguments', [])
header = ' WATCHMAKER RUN '
header = header.rjust((40 + len(header) // 2), '#').ljust(80, '#')
self.log.info(header)
self.log.debug('Watchmaker Version: %s', __version__)
self.log.debug('Parameters: %s', arguments)
self.log.debug('Extra Parameters: %s', extra_arguments)
# Pop remaining arguments used by watchmaker.Client itself
self.default_config = os.path.join(static.__path__[0], 'config.yaml')
self.no_reboot = arguments.pop('no_reboot', False)
self.config_path = arguments.pop('config_path')
self.log_dir = arguments.pop('log_dir')
self.log_level = arguments.pop('log_level')
# Get the system params
self.system = platform.system().lower()
self._set_system_params()
self.log.debug('System Type: %s', self.system)
self.log.debug('System Parameters: %s', self.system_params)
# All remaining arguments are worker_args
worker_args = arguments
# Convert extra_arguments to a dict and merge it with worker_args.
def default_config():
"""Return default configuration for watchmaker."""
config_path = os.path.join(static.__path__[0], 'config.yaml')
with open(config_path, 'r') as stream:
return yaml.safe_load(stream)
:param s3: Should an s3 bucket be used for the installation files.
:type s3: bool
:param config_path: Path to YAML configuration file.
:type config_path: basestring
:param stream: Enables self.logger to a file.
:type stream: bool
:param log_path: Path to logfile for stream self.logger.
:type log_path: basestring
"""
self.kwargs = {}
self.noreboot = noreboot
self.s3 = s3
self.system = platform.system()
self.config_path = config_path
self.default_config = os.path.join(static.__path__[0], 'config.yaml')
self.log_path = log_path
self.config = None
self.system_params = None
self.system_drive = None
self.execution_scripts = None
self.logger = logging.getLogger()
if stream and os.path.exists(log_path):
logging.basicConfig(filename=os.path.join(self.log_path,
'watchmaker-{0}.log'.format(str(datetime.date.today()))),
format='%(levelname)s:\t%(message)s',
level=logging.DEBUG)
self.logger = logging.getLogger()
self.logger.info('\n\n\n{0}'.format(datetime.datetime.now()))
elif stream:
self.logger.error('{0} does not exist'.format(log_path))
raise WatchmakerException(msg)
try:
salt_files_dir = salt_content_glob[0]
except IndexError:
msg = 'Salt content glob path \'{0}\' not' \
' found in {1}'.format(
self.salt_content_path,
self.salt_content)
self.log.critical(msg)
raise WatchmakerException(msg)
watchmaker.utils.copy_subdirectories(
salt_files_dir, extract_dir, self.log)
bundled_content = os.sep.join(
(static.__path__[0], 'salt', 'content')
)
watchmaker.utils.copy_subdirectories(
bundled_content, extract_dir, self.log)
with codecs.open(
os.path.join(self.salt_conf_path, 'minion'),
'r+',
encoding="utf-8"
) as fh_:
salt_conf = yaml.safe_load(fh_)
salt_conf.update(self.salt_file_roots)
fh_.seek(0)
yaml.safe_dump(salt_conf, fh_, default_flow_style=False)