<s> import re <EOL> import threading <EOL> from oslo_config import cfg <EOL> from oslo_log import log as logging <EOL> from sahara import conductor as c <EOL> from sahara import context <EOL> from sahara . i18n import _ <EOL> from sahara . i18n import _LE <EOL> from sahara . i18n import _LW <EOL> from sahara . plugins import provisioning as plugin_base <EOL> from sahara . utils import cluster_progress_ops as cpo <EOL> from sahara . utils . openstack import base as b <EOL> from sahara . utils . openstack import cinder <EOL> from sahara . utils . openstack import nova <EOL> from sahara . utils import poll_utils <EOL> conductor = c . API <EOL> LOG = logging . getLogger ( __name__ ) <EOL> CONF = cfg . CONF <EOL> def _get_timeout_for_disk_preparing ( cluster ) : <EOL> configs = cluster . cluster_configs . to_dict ( ) <EOL> option_name = plugin_base . DISKS_PREPARING_TIMEOUT . name <EOL> option_target = plugin_base . DISKS_PREPARING_TIMEOUT . applicable_target <EOL> try : <EOL> return int ( configs [ option_target ] [ option_name ] ) <EOL> except Exception : <EOL> return int ( plugin_base . DISKS_PREPARING_TIMEOUT . default_value ) <EOL> def _is_xfs_enabled ( cluster ) : <EOL> configs = cluster . cluster_configs . to_dict ( ) <EOL> option_name = plugin_base . XFS_ENABLED . name <EOL> option_target = plugin_base . XFS_ENABLED . applicable_target <EOL> try : <EOL> return bool ( configs [ option_target ] [ option_name ] ) <EOL> except Exception : <EOL> return bool ( plugin_base . XFS_ENABLED . default_value ) <EOL> def _get_os_distrib ( remote ) : <EOL> return remote . execute_command ( '<STR_LIT>' ) [ <NUM_LIT:1> ] . strip ( ) . lower ( ) <EOL> def _check_installed_xfs ( instance ) : <EOL> redhat = "<STR_LIT>" <EOL> debian = "<STR_LIT>" <EOL> cmd_map = { <EOL> "<STR_LIT>" : redhat , <EOL> "<STR_LIT>" : redhat , <EOL> "<STR_LIT>" : redhat , <EOL> "<STR_LIT>" : debian , <EOL> '<STR_LIT>' : debian <EOL> } <EOL> with instance . remote ( ) as r : <EOL> distro = _get_os_distrib ( r ) <EOL> if not cmd_map . get ( distro ) : <EOL> LOG . warning ( <EOL> _LW ( "<STR_LIT>" <EOL> "<STR_LIT>" ) . format ( distro = distro ) ) <EOL> return False <EOL> try : <EOL> r . execute_command ( cmd_map . get ( distro ) , run_as_root = True ) <EOL> return True <EOL> except Exception as e : <EOL> LOG . warning ( <EOL> _LW ( "<STR_LIT>" ) . format ( reason = e ) ) <EOL> return False <EOL> def _can_use_xfs ( instances ) : <EOL> cluster = instances [ <NUM_LIT:0> ] . cluster <EOL> if not _is_xfs_enabled ( cluster ) : <EOL> return False <EOL> for instance in instances : <EOL> if not _check_installed_xfs ( instance ) : <EOL> return False <EOL> return True <EOL> def mount_to_instances ( instances ) : <EOL> if len ( instances ) == <NUM_LIT:0> : <EOL> return <EOL> use_xfs = _can_use_xfs ( instances ) <EOL> for instance in instances : <EOL> with context . set_current_instance_id ( instance . instance_id ) : <EOL> devices = _find_instance_devices ( instance ) <EOL> if devices : <EOL> cpo . add_provisioning_step ( <EOL> instance . cluster_id , <EOL> _ ( "<STR_LIT>" ) . format ( <EOL> inst_name = instance . instance_name ) , len ( devices ) ) <EOL> formatted_devices = [ ] <EOL> lock = threading . Lock ( ) <EOL> with context . ThreadGroup ( ) as tg : <EOL> for device in devices : <EOL> tg . spawn ( '<STR_LIT>' % device , _format_device , <EOL> instance , device , use_xfs , formatted_devices , <EOL> lock ) <EOL> conductor . instance_update ( <EOL> context . current ( ) , instance , <EOL> { "<STR_LIT>" : len ( formatted_devices ) } ) <EOL> for idx , dev in enumerate ( formatted_devices ) : <EOL> _mount_volume_to_node ( instance , idx + <NUM_LIT:1> , dev , use_xfs ) <EOL> def _find_instance_devices ( instance ) : <EOL> with instance . remote ( ) as r : <EOL> code , attached_info = r . execute_command ( <EOL> "<STR_LIT>" ) <EOL> attached_dev = attached_info . split ( ) <EOL> code , mounted_info = r . execute_command ( <EOL> "<STR_LIT>" ) <EOL> mounted_dev = mounted_info . split ( ) <EOL> for dev in attached_dev [ : ] : <EOL> idx = re . sub ( "<STR_LIT>" , "<STR_LIT>" , dev ) <EOL> if idx : <EOL> if dev in mounted_dev : <EOL> attached_dev . remove ( re . sub ( "<STR_LIT>" , "<STR_LIT>" , dev ) ) <EOL> attached_dev . remove ( dev ) <EOL> for dev in attached_dev [ : ] : <EOL> if re . sub ( "<STR_LIT>" , "<STR_LIT>" , dev ) : <EOL> if re . sub ( "<STR_LIT>" , "<STR_LIT>" , dev ) in attached_dev : <EOL> attached_dev . remove ( dev ) <EOL> attached_dev = [ dev for dev in attached_dev if dev not in mounted_dev ] <EOL> return attached_dev <EOL> @ cpo . event_wrapper ( mark_successful_on_exit = True ) <EOL> def _mount_volume_to_node ( instance , index , device , use_xfs ) : <EOL> LOG . debug ( "<STR_LIT>" . format ( device = device ) ) <EOL> mount_point = instance . node_group . volume_mount_prefix + str ( index ) <EOL> _mount_volume ( instance , device , mount_point , use_xfs ) <EOL> LOG . debug ( "<STR_LIT>" ) <EOL> def _format_device ( <EOL> instance , device , use_xfs , formatted_devices = None , lock = None ) : <EOL> with instance . remote ( ) as r : <EOL> try : <EOL> timeout = _get_timeout_for_disk_preparing ( instance . cluster ) <EOL> fs_opts = '<STR_LIT>' <EOL> command = '<STR_LIT>' % ( fs_opts , device ) <EOL> if use_xfs : <EOL> command = '<STR_LIT>' % device <EOL> r . execute_command ( command , timeout = timeout ) <EOL> if lock : <EOL> with lock : <EOL> formatted_devices . append ( device ) <EOL> except Exception as e : <EOL> LOG . warning ( <EOL> _LW ( "<STR_LIT>" ) . format ( <EOL> dev = device , reason = e ) ) <EOL> cpo . add_fail_event ( instance , e ) <EOL> def _mount_volume ( instance , device_path , mount_point , use_xfs ) : <EOL> with instance . remote ( ) as r : <EOL> try : <EOL> timeout = _get_timeout_for_disk_preparing ( instance . cluster ) <EOL> r . execute_command ( '<STR_LIT>' % mount_point ) <EOL> mount_opts = '<STR_LIT>' <EOL> if use_xfs : <EOL> mount_opts = "<STR_LIT>" <EOL> r . execute_command ( '<STR_LIT>' % <EOL> ( mount_opts , device_path , mount_point ) , <EOL> timeout = timeout ) <EOL> r . execute_command ( <EOL> '<STR_LIT>' % device_path ) <EOL> except Exception : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) ) <EOL> raise <EOL> def detach_from_instance ( instance ) : <EOL> for volume_id in instance . volumes : <EOL> _detach_volume ( instance , volume_id ) <EOL> _delete_volume ( volume_id ) <EOL> @ poll_utils . poll_status ( <EOL> '<STR_LIT>' , _ ( "<STR_LIT>" ) , sleep = <NUM_LIT:2> ) <EOL> def _await_detach ( volume_id ) : <EOL> volume = cinder . get_volume ( volume_id ) <EOL> if volume . status not in [ '<STR_LIT>' , '<STR_LIT:error>' ] : <EOL> return False <EOL> return True <EOL> def _detach_volume ( instance , volume_id ) : <EOL> volume = cinder . get_volume ( volume_id ) <EOL> try : <EOL> LOG . debug ( "<STR_LIT>" . format ( id = volume_id ) ) <EOL> b . execute_with_retries ( nova . client ( ) . volumes . delete_server_volume , <EOL> instance . instance_id , volume_id ) <EOL> except Exception : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) . format ( id = volume . id ) ) <EOL> detach_timeout = CONF . timeouts . detach_volume_timeout <EOL> LOG . debug ( "<STR_LIT>" . format ( <EOL> timeout = detach_timeout , id = volume_id ) ) <EOL> _await_detach ( volume_id ) <EOL> def _delete_volume ( volume_id ) : <EOL> LOG . debug ( "<STR_LIT>" . format ( volume = volume_id ) ) <EOL> volume = cinder . get_volume ( volume_id ) <EOL> try : <EOL> b . execute_with_retries ( volume . delete ) <EOL> except Exception : <EOL> LOG . error ( _LE ( "<STR_LIT>" ) . format ( <EOL> volume = volume . id ) ) </s>
