<s> """<STR_LIT>""" <EOL> import datetime <EOL> import functools <EOL> import itertools <EOL> import threading <EOL> import time <EOL> from ganeti import constants <EOL> import qa_config <EOL> import qa_error <EOL> from qa_instance_utils import GetGenericAddParameters <EOL> import qa_job_utils <EOL> import qa_logging <EOL> import qa_utils <EOL> MAX_JOB_SUBMISSION_DURATION = <NUM_LIT> <EOL> class _JobQueueDriver ( object ) : <EOL> """<STR_LIT>""" <EOL> _UNKNOWN_STATUS = "<STR_LIT>" <EOL> class _JobEntry ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , job_id , running_fn , success_fn ) : <EOL> self . job_id = job_id <EOL> self . running_fn = running_fn <EOL> self . success_fn = success_fn <EOL> def __str__ ( self ) : <EOL> return str ( self . job_id ) <EOL> def __init__ ( self ) : <EOL> self . _jobs = { } <EOL> self . _running_notified = set ( ) <EOL> self . _jobs_per_status = { } <EOL> self . _lock = threading . RLock ( ) <EOL> def AddJob ( self , job_id , running_fn = None , success_fn = None ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> self . _jobs [ job_id ] = _JobQueueDriver . _JobEntry ( job_id , <EOL> running_fn , <EOL> success_fn ) <EOL> self . _jobs_per_status . setdefault ( self . _UNKNOWN_STATUS , [ ] ) . append ( job_id ) <EOL> def _FetchJobStatuses ( self ) : <EOL> """<STR_LIT>""" <EOL> job_statuses = qa_job_utils . GetJobStatuses ( self . _GetJobIds ( ) ) <EOL> new_statuses = { } <EOL> for job_id , status in job_statuses . items ( ) : <EOL> new_statuses . setdefault ( status , [ ] ) . append ( self . _jobs [ int ( job_id ) ] ) <EOL> self . _jobs_per_status = new_statuses <EOL> def _GetJobIds ( self ) : <EOL> return list ( self . _jobs . keys ( ) ) <EOL> def _GetJobsInStatuses ( self , statuses ) : <EOL> """<STR_LIT>""" <EOL> ret = [ ] <EOL> for state in statuses : <EOL> ret . extend ( self . _jobs_per_status . get ( state , [ ] ) ) <EOL> return ret <EOL> def _UpdateJobStatuses ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _FetchJobStatuses ( ) <EOL> error_jobs = self . _GetJobsInStatuses ( [ constants . JOB_STATUS_ERROR ] ) <EOL> if error_jobs : <EOL> raise qa_error . Error ( <EOL> "<STR_LIT>" % [ job . job_id for job in error_jobs ] ) <EOL> for job in self . _GetJobsInStatuses ( [ constants . JOB_STATUS_RUNNING , <EOL> constants . JOB_STATUS_SUCCESS ] ) : <EOL> if job . job_id not in self . _running_notified : <EOL> if job . running_fn is not None : <EOL> job . running_fn ( self , job . job_id ) <EOL> self . _running_notified . add ( job . job_id ) <EOL> for job in self . _GetJobsInStatuses ( [ constants . JOB_STATUS_SUCCESS ] ) : <EOL> if job . success_fn is not None : <EOL> job . success_fn ( self , job . job_id ) <EOL> del self . _jobs [ job . job_id ] <EOL> def _HasPendingJobs ( self ) : <EOL> """<STR_LIT>""" <EOL> with self . _lock : <EOL> self . _UpdateJobStatuses ( ) <EOL> uncompleted_jobs = self . _GetJobsInStatuses ( <EOL> constants . JOB_STATUS_ALL - constants . JOBS_FINALIZED ) <EOL> unknown_jobs = self . _GetJobsInStatuses ( [ self . _UNKNOWN_STATUS ] ) <EOL> return len ( uncompleted_jobs ) > <NUM_LIT:0> or len ( unknown_jobs ) > <NUM_LIT:0> <EOL> def WaitForCompletion ( self ) : <EOL> """<STR_LIT>""" <EOL> while self . _HasPendingJobs ( ) : <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> with self . _lock : <EOL> if self . _jobs : <EOL> raise qa_error . Error ( <EOL> "<STR_LIT>" % self . _GetJobIds ( ) ) <EOL> def _AcquireAllInstances ( ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> while True : <EOL> instance = qa_config . AcquireInstance ( ) <EOL> yield instance <EOL> except qa_error . OutOfInstancesError : <EOL> pass <EOL> def _AcquireAllNodes ( ) : <EOL> """<STR_LIT>""" <EOL> exclude = [ ] <EOL> try : <EOL> while True : <EOL> node = qa_config . AcquireNode ( exclude = exclude ) <EOL> exclude . append ( node ) <EOL> yield node <EOL> except qa_error . OutOfNodesError : <EOL> pass <EOL> def _ExecuteJobSubmittingCmd ( cmd ) : <EOL> """<STR_LIT>""" <EOL> start = datetime . datetime . now ( ) <EOL> result = qa_job_utils . ExecuteJobProducingCommand ( cmd ) <EOL> duration = qa_utils . TimedeltaToTotalSeconds ( datetime . datetime . now ( ) - start ) <EOL> if duration > MAX_JOB_SUBMISSION_DURATION : <EOL> print ( qa_logging . FormatWarning ( <EOL> "<STR_LIT>" % <EOL> ( cmd , duration , MAX_JOB_SUBMISSION_DURATION ) ) ) <EOL> return result <EOL> def _SubmitInstanceCreationJob ( instance , disk_template = None ) : <EOL> """<STR_LIT>""" <EOL> if disk_template is None : <EOL> disk_template = qa_config . GetDefaultDiskTemplate ( ) <EOL> try : <EOL> cmd = ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" % qa_config . get ( "<STR_LIT>" ) , <EOL> "<STR_LIT>" % disk_template ] + <EOL> GetGenericAddParameters ( instance , disk_template ) ) <EOL> cmd . append ( instance . name ) <EOL> instance . SetDiskTemplate ( disk_template ) <EOL> return _ExecuteJobSubmittingCmd ( cmd ) <EOL> except : <EOL> instance . Release ( ) <EOL> raise <EOL> def _SubmitInstanceRemoveJob ( instance ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> cmd = ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> cmd . append ( instance . name ) <EOL> return _ExecuteJobSubmittingCmd ( cmd ) <EOL> finally : <EOL> instance . Release ( ) <EOL> def _TestParallelInstanceCreationAndRemoval ( max_instances = None , <EOL> disk_template = None , <EOL> custom_job_driver = None ) : <EOL> """<STR_LIT>""" <EOL> job_driver = custom_job_driver or _JobQueueDriver ( ) <EOL> def _CreateSuccessFn ( instance , job_driver , _ ) : <EOL> job_id = _SubmitInstanceRemoveJob ( instance ) <EOL> job_driver . AddJob ( job_id ) <EOL> instance_generator = _AcquireAllInstances ( ) <EOL> if max_instances is not None : <EOL> instance_generator = itertools . islice ( instance_generator , max_instances ) <EOL> for instance in instance_generator : <EOL> job_id = _SubmitInstanceCreationJob ( instance , disk_template = disk_template ) <EOL> job_driver . AddJob ( <EOL> job_id , success_fn = functools . partial ( _CreateSuccessFn , instance ) ) <EOL> if custom_job_driver is None : <EOL> job_driver . WaitForCompletion ( ) <EOL> def TestParallelMaxInstanceCreationPerformance ( ) : <EOL> """<STR_LIT>""" <EOL> _TestParallelInstanceCreationAndRemoval ( ) <EOL> def TestParallelNodeCountInstanceCreationPerformance ( ) : <EOL> """<STR_LIT>""" <EOL> nodes = list ( _AcquireAllNodes ( ) ) <EOL> _TestParallelInstanceCreationAndRemoval ( max_instances = len ( nodes ) ) <EOL> qa_config . ReleaseManyNodes ( nodes ) <EOL> def CreateAllInstances ( ) : <EOL> """<STR_LIT>""" <EOL> job_driver = _JobQueueDriver ( ) <EOL> instances = list ( _AcquireAllInstances ( ) ) <EOL> for instance in instances : <EOL> job_id = _SubmitInstanceCreationJob ( instance ) <EOL> job_driver . AddJob ( job_id ) <EOL> job_driver . WaitForCompletion ( ) <EOL> return instances <EOL> def RemoveAllInstances ( instances ) : <EOL> """<STR_LIT>""" <EOL> job_driver = _JobQueueDriver ( ) <EOL> for instance in instances : <EOL> job_id = _SubmitInstanceRemoveJob ( instance ) <EOL> job_driver . AddJob ( job_id ) <EOL> job_driver . WaitForCompletion ( ) <EOL> def TestParallelModify ( instances ) : <EOL> """<STR_LIT>""" <EOL> job_driver = _JobQueueDriver ( ) <EOL> new_min_mem = qa_config . get ( constants . BE_MAXMEM ) <EOL> for instance in instances : <EOL> cmd = ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" % ( constants . BE_MINMEM , new_min_mem ) ] ) <EOL> cmd . append ( instance . name ) <EOL> job_driver . AddJob ( _ExecuteJobSubmittingCmd ( cmd ) ) <EOL> cmd = ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> cmd . append ( instance . name ) <EOL> job_driver . AddJob ( _ExecuteJobSubmittingCmd ( cmd ) ) <EOL> cmd = ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" % ( constants . BE_MINMEM , new_min_mem ) ] ) <EOL> cmd . append ( instance . name ) <EOL> job_driver . AddJob ( _ExecuteJobSubmittingCmd ( cmd ) ) <EOL> job_driver . WaitForCompletion ( ) <EOL> def TestParallelInstanceOSOperations ( instances ) : <EOL> """<STR_LIT>""" <EOL> OPS = [ "<STR_LIT:start>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> job_driver = _JobQueueDriver ( ) <EOL> def _SubmitNextOperation ( instance , start , idx , job_driver , _ ) : <EOL> if idx == len ( OPS ) : <EOL> return <EOL> op_idx = ( start + idx ) % len ( OPS ) <EOL> next_fn = functools . partial ( _SubmitNextOperation , instance , start , idx + <NUM_LIT:1> ) <EOL> if OPS [ op_idx ] == "<STR_LIT>" and instance . disk_template == constants . DT_DISKLESS : <EOL> next_fn ( job_driver , None ) <EOL> return <EOL> elif OPS [ op_idx ] == "<STR_LIT>" : <EOL> shutdown_cmd = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , instance . name ] <EOL> cmd = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , instance . name ] <EOL> job_driver . AddJob ( _ExecuteJobSubmittingCmd ( shutdown_cmd ) , <EOL> running_fn = lambda _ , __ : job_driver . AddJob ( <EOL> _ExecuteJobSubmittingCmd ( cmd ) , <EOL> running_fn = next_fn ) ) <EOL> else : <EOL> cmd = [ "<STR_LIT>" , OPS [ op_idx ] , "<STR_LIT>" ] <EOL> if OPS [ op_idx ] == "<STR_LIT>" : <EOL> cmd . append ( "<STR_LIT>" ) <EOL> cmd . append ( instance . name ) <EOL> job_id = _ExecuteJobSubmittingCmd ( cmd ) <EOL> job_driver . AddJob ( job_id , running_fn = next_fn ) <EOL> for start , instance in enumerate ( instances ) : <EOL> _SubmitNextOperation ( instance , start % len ( OPS ) , <NUM_LIT:0> , job_driver , None ) <EOL> job_driver . WaitForCompletion ( ) <EOL> def TestParallelInstanceQueries ( instances ) : <EOL> """<STR_LIT>""" <EOL> threads = qa_job_utils . QAThreadGroup ( ) <EOL> for instance in instances : <EOL> cmd = [ "<STR_LIT>" , "<STR_LIT:info>" , instance . name ] <EOL> info_thread = qa_job_utils . QAThread ( qa_utils . AssertCommand , [ cmd ] , { } ) <EOL> threads . Start ( info_thread ) <EOL> cmd = [ "<STR_LIT>" , "<STR_LIT:list>" ] <EOL> list_thread = qa_job_utils . QAThread ( qa_utils . AssertCommand , [ cmd ] , { } ) <EOL> threads . Start ( list_thread ) <EOL> threads . JoinAndReraise ( ) <EOL> def TestJobQueueSubmissionPerformance ( ) : <EOL> """<STR_LIT>""" <EOL> MAX_CLUSTER_INFO_SECONDS = <NUM_LIT> <EOL> job_driver = _JobQueueDriver ( ) <EOL> submission_durations = [ ] <EOL> def _VerifySubmissionDuration ( duration_seconds ) : <EOL> if len ( submission_durations ) >= <NUM_LIT:10> : <EOL> avg_duration = sum ( submission_durations ) / len ( submission_durations ) <EOL> max_duration = avg_duration * <NUM_LIT> <EOL> if duration_seconds > max_duration : <EOL> print ( qa_logging . FormatWarning ( <EOL> "<STR_LIT>" % <EOL> ( duration_seconds , max_duration ) ) ) <EOL> else : <EOL> submission_durations . append ( duration_seconds ) <EOL> def _SubmitDelayJob ( count ) : <EOL> for _ in range ( count ) : <EOL> cmd = [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> start = datetime . datetime . now ( ) <EOL> job_id = _ExecuteJobSubmittingCmd ( cmd ) <EOL> duration_seconds = qa_utils . TimedeltaToTotalSeconds ( datetime . datetime . now ( ) - start ) <EOL> _VerifySubmissionDuration ( duration_seconds ) <EOL> job_driver . AddJob ( job_id ) <EOL> threads = qa_job_utils . QAThreadGroup ( ) <EOL> for _ in range ( <NUM_LIT:10> ) : <EOL> thread = qa_job_utils . QAThread ( _SubmitDelayJob , [ <NUM_LIT:20> ] , { } ) <EOL> threads . Start ( thread ) <EOL> threads . JoinAndReraise ( ) <EOL> qa_utils . AssertCommand ( [ "<STR_LIT>" , "<STR_LIT:info>" ] , <EOL> max_seconds = MAX_CLUSTER_INFO_SECONDS ) <EOL> job_driver . WaitForCompletion ( ) <EOL> def TestParallelDRBDInstanceCreationPerformance ( ) : <EOL> """<STR_LIT>""" <EOL> assert qa_config . IsTemplateSupported ( constants . DT_DRBD8 ) <EOL> nodes = list ( _AcquireAllNodes ( ) ) <EOL> _TestParallelInstanceCreationAndRemoval ( max_instances = len ( nodes ) * <NUM_LIT:2> , <EOL> disk_template = constants . DT_DRBD8 ) <EOL> qa_config . ReleaseManyNodes ( nodes ) <EOL> def TestParallelPlainInstanceCreationPerformance ( ) : <EOL> """<STR_LIT>""" <EOL> assert qa_config . IsTemplateSupported ( constants . DT_PLAIN ) <EOL> nodes = list ( _AcquireAllNodes ( ) ) <EOL> _TestParallelInstanceCreationAndRemoval ( max_instances = len ( nodes ) * <NUM_LIT:2> , <EOL> disk_template = constants . DT_PLAIN ) <EOL> qa_config . ReleaseManyNodes ( nodes ) <EOL> def _TestInstanceOperationInParallelToInstanceCreation ( * cmds ) : <EOL> """<STR_LIT>""" <EOL> def _SubmitNextCommand ( cmd_idx , job_driver , _ ) : <EOL> if cmd_idx >= len ( cmds ) : <EOL> return <EOL> job_id = _ExecuteJobSubmittingCmd ( cmds [ cmd_idx ] ) <EOL> job_driver . AddJob ( <EOL> job_id , success_fn = functools . partial ( _SubmitNextCommand , cmd_idx + <NUM_LIT:1> ) ) <EOL> assert qa_config . IsTemplateSupported ( constants . DT_DRBD8 ) <EOL> assert len ( cmds ) > <NUM_LIT:0> <EOL> job_driver = _JobQueueDriver ( ) <EOL> _SubmitNextCommand ( <NUM_LIT:0> , job_driver , None ) <EOL> _TestParallelInstanceCreationAndRemoval ( max_instances = <NUM_LIT:1> , <EOL> disk_template = constants . DT_DRBD8 , <EOL> custom_job_driver = job_driver ) <EOL> job_driver . WaitForCompletion ( ) <EOL> def TestParallelInstanceFailover ( instance ) : <EOL> """<STR_LIT>""" <EOL> _TestInstanceOperationInParallelToInstanceCreation ( <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> instance . name ] ) <EOL> def TestParallelInstanceMigration ( instance ) : <EOL> """<STR_LIT>""" <EOL> _TestInstanceOperationInParallelToInstanceCreation ( <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , instance . name ] ) <EOL> def TestParallelInstanceReplaceDisks ( instance ) : <EOL> """<STR_LIT>""" <EOL> _TestInstanceOperationInParallelToInstanceCreation ( <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> instance . name ] ) <EOL> def TestParallelInstanceReboot ( instance ) : <EOL> """<STR_LIT>""" <EOL> _TestInstanceOperationInParallelToInstanceCreation ( <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , instance . name ] ) <EOL> def TestParallelInstanceReinstall ( instance ) : <EOL> """<STR_LIT>""" <EOL> qa_utils . AssertCommand ( [ "<STR_LIT>" , "<STR_LIT>" , instance . name ] ) <EOL> _TestInstanceOperationInParallelToInstanceCreation ( <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , instance . name ] ) <EOL> qa_utils . AssertCommand ( [ "<STR_LIT>" , "<STR_LIT:start>" , instance . name ] ) <EOL> def TestParallelInstanceRename ( instance ) : <EOL> """<STR_LIT>""" <EOL> qa_utils . AssertCommand ( [ "<STR_LIT>" , "<STR_LIT>" , instance . name ] ) <EOL> new_instance = qa_config . AcquireInstance ( ) <EOL> try : <EOL> _TestInstanceOperationInParallelToInstanceCreation ( <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , instance . name , new_instance . name ] , <EOL> [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , new_instance . name , instance . name ] ) <EOL> finally : <EOL> new_instance . Release ( ) <EOL> qa_utils . AssertCommand ( [ "<STR_LIT>" , "<STR_LIT:start>" , instance . name ] ) </s>
