mirror of
https://github.com/ARM-software/workload-automation.git
synced 2025-01-31 18:21:14 +00:00
9257a787f9
Metadata is a key-value mapping for arbitrary data, similar to classifiers. Unlike classifiers, metadata does not directly relate to the results of the execution, but to the execution itself, and typically would not be processed by Output Processors in the same way as classifiers. Metadata can also be a lot more free-form in it's value structure; while classifier values are simple scalars, metadata values can be arbitrary POD structures.
629 lines
22 KiB
Python
629 lines
22 KiB
Python
# Copyright 2013-2015 ARM Limited
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
#
|
|
|
|
# pylint: disable=no-member
|
|
|
|
import logging
|
|
import os
|
|
import shutil
|
|
from copy import copy
|
|
from datetime import datetime
|
|
|
|
import wa.framework.signal as signal
|
|
from wa.framework import instrument
|
|
from wa.framework.configuration.core import Status
|
|
from wa.framework.exception import TargetError, HostError, WorkloadError,\
|
|
TargetNotRespondingError, TimeoutError
|
|
from wa.framework.job import Job
|
|
from wa.framework.output import init_job_output
|
|
from wa.framework.output_processor import ProcessorManager
|
|
from wa.framework.resource import ResourceResolver
|
|
from wa.framework.target.manager import TargetManager
|
|
from wa.utils import log
|
|
from wa.utils.misc import merge_config_values, format_duration
|
|
|
|
|
|
class ExecutionContext(object):
|
|
|
|
@property
|
|
def previous_job(self):
|
|
if not self.completed_jobs:
|
|
return None
|
|
return self.completed_jobs[-1]
|
|
|
|
@property
|
|
def next_job(self):
|
|
if not self.job_queue:
|
|
return None
|
|
return self.job_queue[0]
|
|
|
|
@property
|
|
def spec_changed(self):
|
|
if self.previous_job is None and self.current_job is not None: # Start of run
|
|
return True
|
|
if self.previous_job is not None and self.current_job is None: # End of run
|
|
return True
|
|
return self.current_job.spec.id != self.previous_job.spec.id
|
|
|
|
@property
|
|
def spec_will_change(self):
|
|
if self.current_job is None and self.next_job is not None: # Start of run
|
|
return True
|
|
if self.current_job is not None and self.next_job is None: # End of run
|
|
return True
|
|
return self.current_job.spec.id != self.next_job.spec.id
|
|
|
|
@property
|
|
def workload(self):
|
|
if self.current_job:
|
|
return self.current_job.workload
|
|
|
|
@property
|
|
def job_output(self):
|
|
if self.current_job:
|
|
return self.current_job.output
|
|
|
|
@property
|
|
def output(self):
|
|
if self.current_job:
|
|
return self.job_output
|
|
return self.run_output
|
|
|
|
@property
|
|
def output_directory(self):
|
|
return self.output.basepath
|
|
|
|
@property
|
|
def reboot_policy(self):
|
|
return self.cm.run_config.reboot_policy
|
|
|
|
def __init__(self, cm, tm, output):
|
|
self.logger = logging.getLogger('context')
|
|
self.cm = cm
|
|
self.tm = tm
|
|
self.run_output = output
|
|
self.run_state = output.state
|
|
self.target_info = self.tm.get_target_info()
|
|
self.logger.debug('Loading resource discoverers')
|
|
self.resolver = ResourceResolver(cm.plugin_cache)
|
|
self.resolver.load()
|
|
self.job_queue = None
|
|
self.completed_jobs = None
|
|
self.current_job = None
|
|
self.successful_jobs = 0
|
|
self.failed_jobs = 0
|
|
self.run_interrupted = False
|
|
|
|
def start_run(self):
|
|
self.output.info.start_time = datetime.utcnow()
|
|
self.output.write_info()
|
|
self.job_queue = copy(self.cm.jobs)
|
|
self.completed_jobs = []
|
|
self.run_state.status = Status.STARTED
|
|
self.output.status = Status.STARTED
|
|
self.output.write_state()
|
|
|
|
def end_run(self):
|
|
if self.successful_jobs:
|
|
if self.failed_jobs:
|
|
status = Status.PARTIAL
|
|
else:
|
|
status = Status.OK
|
|
else:
|
|
status = Status.FAILED
|
|
self.run_state.status = status
|
|
self.run_output.status = status
|
|
self.run_output.info.end_time = datetime.utcnow()
|
|
self.run_output.info.duration = self.run_output.info.end_time -\
|
|
self.run_output.info.start_time
|
|
self.run_output.write_info()
|
|
self.run_output.write_state()
|
|
self.run_output.write_result()
|
|
|
|
def finalize(self):
|
|
self.tm.finalize()
|
|
|
|
def start_job(self):
|
|
if not self.job_queue:
|
|
raise RuntimeError('No jobs to run')
|
|
self.current_job = self.job_queue.pop(0)
|
|
job_output = init_job_output(self.run_output, self.current_job)
|
|
self.current_job.set_output(job_output)
|
|
self.update_job_state(self.current_job)
|
|
return self.current_job
|
|
|
|
def end_job(self):
|
|
if not self.current_job:
|
|
raise RuntimeError('No jobs in progress')
|
|
self.completed_jobs.append(self.current_job)
|
|
self.update_job_state(self.current_job)
|
|
self.output.write_result()
|
|
self.current_job = None
|
|
|
|
def set_status(self, status, force=False):
|
|
if not self.current_job:
|
|
raise RuntimeError('No jobs in progress')
|
|
self.current_job.set_status(status, force)
|
|
|
|
def extract_results(self):
|
|
self.tm.extract_results(self)
|
|
|
|
def move_failed(self, job):
|
|
self.run_output.move_failed(job.output)
|
|
|
|
def update_job_state(self, job):
|
|
self.run_state.update_job(job)
|
|
self.run_output.write_state()
|
|
|
|
def skip_job(self, job):
|
|
job.status = Status.SKIPPED
|
|
self.run_state.update_job(job)
|
|
self.completed_jobs.append(job)
|
|
|
|
def skip_remaining_jobs(self):
|
|
while self.job_queue:
|
|
job = self.job_queue.pop(0)
|
|
self.skip_job(job)
|
|
self.write_state()
|
|
|
|
def write_state(self):
|
|
self.run_output.write_state()
|
|
|
|
def get_metric(self, name):
|
|
try:
|
|
return self.output.get_metric(name)
|
|
except HostError:
|
|
if not self.current_job:
|
|
raise
|
|
return self.run_output.get_metric(name)
|
|
|
|
def add_metric(self, name, value, units=None, lower_is_better=False,
|
|
classifiers=None):
|
|
if self.current_job:
|
|
classifiers = merge_config_values(self.current_job.classifiers,
|
|
classifiers)
|
|
self.output.add_metric(name, value, units, lower_is_better, classifiers)
|
|
|
|
def get_artifact(self, name):
|
|
try:
|
|
return self.output.get_artifact(name)
|
|
except HostError:
|
|
if not self.current_job:
|
|
raise
|
|
return self.run_output.get_artifact(name)
|
|
|
|
def get_artifact_path(self, name):
|
|
try:
|
|
return self.output.get_artifact_path(name)
|
|
except HostError:
|
|
if not self.current_job:
|
|
raise
|
|
return self.run_output.get_artifact_path(name)
|
|
|
|
def add_artifact(self, name, path, kind, description=None, classifiers=None):
|
|
self.output.add_artifact(name, path, kind, description, classifiers)
|
|
|
|
def add_run_artifact(self, name, path, kind, description=None,
|
|
classifiers=None):
|
|
self.run_output.add_artifact(name, path, kind, description, classifiers)
|
|
|
|
def add_event(self, message):
|
|
self.output.add_event(message)
|
|
|
|
def add_metadata(self, key, *args, **kwargs):
|
|
self.output.add_metadata(key, *args, **kwargs)
|
|
|
|
def update_metadata(self, key, *args):
|
|
self.output.update_metadata(key, *args)
|
|
|
|
def take_screenshot(self, filename):
|
|
filepath = self._get_unique_filepath(filename)
|
|
self.tm.target.capture_screen(filepath)
|
|
if os.path.isfile(filepath):
|
|
self.add_artifact('screenshot', filepath, kind='log')
|
|
|
|
def take_uiautomator_dump(self, filename):
|
|
filepath = self._get_unique_filepath(filename)
|
|
self.tm.target.capture_ui_hierarchy(filepath)
|
|
self.add_artifact('uitree', filepath, kind='log')
|
|
|
|
def record_ui_state(self, basename):
|
|
self.logger.info('Recording screen state...')
|
|
self.take_screenshot('{}.png'.format(basename))
|
|
target = self.tm.target
|
|
if target.os == 'android' or\
|
|
(target.os == 'chromeos' and target.has('android_container')):
|
|
self.take_uiautomator_dump('{}.uix'.format(basename))
|
|
|
|
def initialize_jobs(self):
|
|
new_queue = []
|
|
failed_ids = []
|
|
for job in self.job_queue:
|
|
if job.id in failed_ids:
|
|
# Don't try to initialize a job if another job with the same ID
|
|
# (i.e. same job spec) has failed - we can assume it will fail
|
|
# too.
|
|
self.skip_job(job)
|
|
continue
|
|
|
|
try:
|
|
job.initialize(self)
|
|
except WorkloadError as e:
|
|
job.set_status(Status.FAILED)
|
|
log.log_error(e, self.logger)
|
|
failed_ids.append(job.id)
|
|
|
|
if self.cm.run_config.bail_on_init_failure:
|
|
raise
|
|
else:
|
|
new_queue.append(job)
|
|
|
|
self.job_queue = new_queue
|
|
|
|
def _get_unique_filepath(self, filename):
|
|
filepath = os.path.join(self.output_directory, filename)
|
|
rest, ext = os.path.splitext(filepath)
|
|
i = 1
|
|
new_filepath = '{}-{}{}'.format(rest, i, ext)
|
|
|
|
if not os.path.exists(filepath) and not os.path.exists(new_filepath):
|
|
return filepath
|
|
elif not os.path.exists(new_filepath):
|
|
# new_filepath does not exit, thefore filepath must exit.
|
|
# this is the first collision
|
|
shutil.move(filepath, new_filepath)
|
|
|
|
while os.path.exists(new_filepath):
|
|
i += 1
|
|
new_filepath = '{}-{}{}'.format(rest, i, ext)
|
|
return new_filepath
|
|
|
|
|
|
class Executor(object):
|
|
"""
|
|
The ``Executor``'s job is to set up the execution context and pass to a
|
|
``Runner`` along with a loaded run specification. Once the ``Runner`` has
|
|
done its thing, the ``Executor`` performs some final reporting before
|
|
returning.
|
|
|
|
The initial context set up involves combining configuration from various
|
|
sources, loading of requided workloads, loading and installation of
|
|
instruments and output processors, etc. Static validation of the combined
|
|
configuration is also performed.
|
|
|
|
"""
|
|
# pylint: disable=R0915
|
|
|
|
def __init__(self):
|
|
self.logger = logging.getLogger('executor')
|
|
self.error_logged = False
|
|
self.warning_logged = False
|
|
self.target_manager = None
|
|
|
|
def execute(self, config_manager, output):
|
|
"""
|
|
Execute the run specified by an agenda. Optionally, selectors may be
|
|
used to only selecute a subset of the specified agenda.
|
|
|
|
Params::
|
|
|
|
:state: a ``ConfigManager`` containing processed configuration
|
|
:output: an initialized ``RunOutput`` that will be used to
|
|
store the results.
|
|
|
|
"""
|
|
signal.connect(self._error_signalled_callback, signal.ERROR_LOGGED)
|
|
signal.connect(self._warning_signalled_callback, signal.WARNING_LOGGED)
|
|
|
|
self.logger.info('Initializing run')
|
|
self.logger.debug('Finalizing run configuration.')
|
|
config = config_manager.finalize()
|
|
output.write_config(config)
|
|
|
|
self.logger.info('Connecting to target')
|
|
self.target_manager = TargetManager(config.run_config.device,
|
|
config.run_config.device_config,
|
|
output.basepath)
|
|
|
|
if config_manager.run_config.reboot_policy.perform_initial_reboot:
|
|
self.logger.info('Performing inital reboot.')
|
|
attempts = config_manager.run_config.max_retries
|
|
while attempts:
|
|
try:
|
|
self.target_manager.target.reboot()
|
|
except TargetError as e:
|
|
if attempts:
|
|
attempts -= 1
|
|
else:
|
|
raise e
|
|
else:
|
|
break
|
|
|
|
output.set_target_info(self.target_manager.get_target_info())
|
|
|
|
self.logger.info('Initializing execution context')
|
|
context = ExecutionContext(config_manager, self.target_manager, output)
|
|
|
|
self.logger.info('Generating jobs')
|
|
config_manager.generate_jobs(context)
|
|
output.write_job_specs(config_manager.job_specs)
|
|
output.write_state()
|
|
|
|
self.logger.info('Installing instruments')
|
|
for instrument_name in config_manager.get_instruments(self.target_manager.target):
|
|
instrument.install(instrument_name, context)
|
|
instrument.validate()
|
|
|
|
self.logger.info('Installing output processors')
|
|
pm = ProcessorManager()
|
|
for proc in config_manager.get_processors():
|
|
pm.install(proc, context)
|
|
pm.validate()
|
|
|
|
self.logger.info('Starting run')
|
|
runner = Runner(context, pm)
|
|
signal.send(signal.RUN_STARTED, self)
|
|
try:
|
|
runner.run()
|
|
finally:
|
|
context.finalize()
|
|
self.execute_postamble(context, output)
|
|
signal.send(signal.RUN_COMPLETED, self)
|
|
|
|
def execute_postamble(self, context, output):
|
|
self.logger.info('Done.')
|
|
duration = format_duration(output.info.duration)
|
|
self.logger.info('Run duration: {}'.format(duration))
|
|
num_ran = context.run_state.num_completed_jobs
|
|
status_summary = 'Ran a total of {} iterations: '.format(num_ran)
|
|
|
|
counter = context.run_state.get_status_counts()
|
|
parts = []
|
|
for status in reversed(Status.levels):
|
|
if status in counter:
|
|
parts.append('{} {}'.format(counter[status], status))
|
|
self.logger.info(status_summary + ', '.join(parts))
|
|
|
|
self.logger.info('Results can be found in {}'.format(output.basepath))
|
|
|
|
if self.error_logged:
|
|
self.logger.warn('There were errors during execution.')
|
|
self.logger.warn('Please see {}'.format(output.logfile))
|
|
elif self.warning_logged:
|
|
self.logger.warn('There were warnings during execution.')
|
|
self.logger.warn('Please see {}'.format(output.logfile))
|
|
|
|
def _error_signalled_callback(self, record):
|
|
self.error_logged = True
|
|
signal.disconnect(self._error_signalled_callback, signal.ERROR_LOGGED)
|
|
|
|
def _warning_signalled_callback(self, record):
|
|
self.warning_logged = True
|
|
signal.disconnect(self._warning_signalled_callback, signal.WARNING_LOGGED)
|
|
|
|
def __str__(self):
|
|
return 'executor'
|
|
|
|
__repr__ = __str__
|
|
|
|
|
|
class Runner(object):
|
|
"""
|
|
Triggers running jobs and processing results
|
|
|
|
Takes pre-initialized ExcecutionContext and ProcessorManager. Handles
|
|
actually running the jobs, and triggers the ProcessorManager to handle
|
|
processing job and run results.
|
|
"""
|
|
|
|
def __init__(self, context, pm):
|
|
self.logger = logging.getLogger('runner')
|
|
self.context = context
|
|
self.pm = pm
|
|
self.output = self.context.output
|
|
self.config = self.context.cm
|
|
|
|
def run(self):
|
|
try:
|
|
self.initialize_run()
|
|
self.send(signal.RUN_INITIALIZED)
|
|
|
|
while self.context.job_queue:
|
|
if self.context.run_interrupted:
|
|
raise KeyboardInterrupt()
|
|
with signal.wrap('JOB_EXECUTION', self, self.context):
|
|
self.run_next_job(self.context)
|
|
|
|
except KeyboardInterrupt as e:
|
|
log.log_error(e, self.logger)
|
|
self.logger.info('Skipping remaining jobs.')
|
|
self.context.skip_remaining_jobs()
|
|
except Exception as e:
|
|
message = e.message if e.message else str(e)
|
|
log.log_error(e, self.logger)
|
|
self.logger.error('Skipping remaining jobs due to "{}".'.format(e))
|
|
self.context.skip_remaining_jobs()
|
|
raise e
|
|
finally:
|
|
self.finalize_run()
|
|
self.send(signal.RUN_FINALIZED)
|
|
|
|
def initialize_run(self):
|
|
self.logger.info('Initializing run')
|
|
signal.connect(self._error_signalled_callback, signal.ERROR_LOGGED)
|
|
signal.connect(self._warning_signalled_callback, signal.WARNING_LOGGED)
|
|
self.context.start_run()
|
|
self.pm.initialize()
|
|
with log.indentcontext():
|
|
self.context.initialize_jobs()
|
|
self.context.write_state()
|
|
|
|
def finalize_run(self):
|
|
self.logger.info('Run completed')
|
|
with log.indentcontext():
|
|
for job in self.context.completed_jobs:
|
|
job.finalize(self.context)
|
|
self.logger.info('Finalizing run')
|
|
self.context.end_run()
|
|
self.pm.enable_all()
|
|
self.pm.process_run_output(self.context)
|
|
self.pm.export_run_output(self.context)
|
|
self.pm.finalize()
|
|
signal.disconnect(self._error_signalled_callback, signal.ERROR_LOGGED)
|
|
signal.disconnect(self._warning_signalled_callback, signal.WARNING_LOGGED)
|
|
|
|
def run_next_job(self, context):
|
|
job = context.start_job()
|
|
self.logger.info('Running job {}'.format(job.id))
|
|
|
|
try:
|
|
log.indent()
|
|
if self.context.reboot_policy.reboot_on_each_job:
|
|
self.logger.info('Rebooting on new job.')
|
|
self.context.tm.target.reboot()
|
|
elif self.context.reboot_policy.reboot_on_each_spec and context.spec_changed:
|
|
self.logger.info('Rebooting on new spec.')
|
|
self.context.tm.target.reboot()
|
|
|
|
context.tm.start()
|
|
self.do_run_job(job, context)
|
|
job.set_status(Status.OK)
|
|
except (Exception, KeyboardInterrupt) as e: # pylint: disable=broad-except
|
|
log.log_error(e, self.logger)
|
|
if isinstance(e, KeyboardInterrupt):
|
|
context.run_interrupted = True
|
|
job.set_status(Status.ABORTED)
|
|
raise e
|
|
else:
|
|
job.set_status(Status.FAILED)
|
|
if isinstance(e, TargetNotRespondingError):
|
|
raise e
|
|
elif isinstance(e, TargetError):
|
|
context.tm.verify_target_responsive(context.reboot_policy.can_reboot)
|
|
finally:
|
|
self.logger.info('Completing job {}'.format(job.id))
|
|
self.send(signal.JOB_COMPLETED)
|
|
context.tm.stop()
|
|
context.end_job()
|
|
|
|
log.dedent()
|
|
self.check_job(job)
|
|
|
|
def do_run_job(self, job, context):
|
|
rc = self.context.cm.run_config
|
|
if job.workload.phones_home and not rc.allow_phone_home:
|
|
self.logger.warning('Skipping job {} ({}) due to allow_phone_home=False'
|
|
.format(job.id, job.workload.name))
|
|
self.context.skip_job(job)
|
|
return
|
|
|
|
job.set_status(Status.RUNNING)
|
|
self.send(signal.JOB_STARTED)
|
|
|
|
job.configure_augmentations(context, self.pm)
|
|
|
|
with signal.wrap('JOB_TARGET_CONFIG', self, context):
|
|
job.configure_target(context)
|
|
|
|
try:
|
|
with signal.wrap('JOB_SETUP', self, context):
|
|
job.setup(context)
|
|
except Exception as e:
|
|
job.set_status(Status.FAILED)
|
|
log.log_error(e, self.logger)
|
|
if isinstance(e, TargetError) or isinstance(e, TimeoutError):
|
|
context.tm.verify_target_responsive(context.reboot_policy.can_reboot)
|
|
self.context.record_ui_state('setup-error')
|
|
raise e
|
|
|
|
try:
|
|
|
|
try:
|
|
with signal.wrap('JOB_EXECUTION', self, context):
|
|
job.run(context)
|
|
except KeyboardInterrupt:
|
|
context.run_interrupted = True
|
|
job.set_status(Status.ABORTED)
|
|
raise
|
|
except Exception as e:
|
|
job.set_status(Status.FAILED)
|
|
log.log_error(e, self.logger)
|
|
if isinstance(e, TargetError) or isinstance(e, TimeoutError):
|
|
context.tm.verify_target_responsive(context.reboot_policy.can_reboot)
|
|
self.context.record_ui_state('run-error')
|
|
raise e
|
|
finally:
|
|
try:
|
|
with signal.wrap('JOB_OUTPUT_PROCESSED', self, context):
|
|
job.process_output(context)
|
|
self.pm.process_job_output(context)
|
|
self.pm.export_job_output(context)
|
|
except Exception as e:
|
|
job.set_status(Status.PARTIAL)
|
|
if isinstance(e, TargetError) or isinstance(e, TimeoutError):
|
|
context.tm.verify_target_responsive(context.reboot_policy.can_reboot)
|
|
self.context.record_ui_state('output-error')
|
|
raise
|
|
|
|
except KeyboardInterrupt:
|
|
context.run_interrupted = True
|
|
job.set_status(Status.ABORTED)
|
|
raise
|
|
finally:
|
|
# If setup was successfully completed, teardown must
|
|
# run even if the job failed
|
|
with signal.wrap('JOB_TEARDOWN', self):
|
|
job.teardown(context)
|
|
|
|
def check_job(self, job):
|
|
rc = self.context.cm.run_config
|
|
if job.status in rc.retry_on_status:
|
|
if job.retries < rc.max_retries:
|
|
msg = 'Job {} iteration {} completed with status {}. retrying...'
|
|
self.logger.error(msg.format(job.id, job.iteration, job.status))
|
|
self.retry_job(job)
|
|
self.context.move_failed(job)
|
|
self.context.write_state()
|
|
else:
|
|
msg = 'Job {} iteration {} completed with status {}. '\
|
|
'Max retries exceeded.'
|
|
self.logger.error(msg.format(job.id, job.iteration, job.status))
|
|
self.context.failed_jobs += 1
|
|
else: # status not in retry_on_status
|
|
self.logger.info('Job completed with status {}'.format(job.status))
|
|
if job.status != 'ABORTED':
|
|
self.context.successful_jobs += 1
|
|
else:
|
|
self.context.failed_jobs += 1
|
|
|
|
def retry_job(self, job):
|
|
retry_job = Job(job.spec, job.iteration, self.context)
|
|
retry_job.workload = job.workload
|
|
retry_job.retries = job.retries + 1
|
|
retry_job.set_status(Status.PENDING)
|
|
self.context.job_queue.insert(0, retry_job)
|
|
|
|
def send(self, s):
|
|
signal.send(s, self, self.context)
|
|
|
|
def _error_signalled_callback(self, record):
|
|
self.context.add_event(record.getMessage())
|
|
|
|
def _warning_signalled_callback(self, record):
|
|
self.context.add_event(record.getMessage())
|
|
|
|
def __str__(self):
|
|
return 'runner'
|