/usr/lib/python3/dist-packages/provisioningserver/drivers/power/fence_cdu.py is in python3-maas-provisioningserver 2.0.0~beta3+bzr4941-0ubuntu1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 | # Copyright 2015-2016 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Fence CDU Power Driver."""
__all__ = []
import re
from time import sleep
from provisioningserver.drivers.power import (
PowerDriver,
PowerError,
)
from provisioningserver.utils import shell
from provisioningserver.utils.shell import (
call_and_check,
ExternalProcessError,
select_c_utf8_locale,
)
class FenceCDUPowerDriver(PowerDriver):
name = 'fence_cdu'
description = "Fence CDU Power Driver."
settings = []
def detect_missing_packages(self):
if not shell.has_command_available('fence_cdu'):
return ['fence-agents']
return []
def _issue_fence_cdu_command(
self, command, fence_cdu=None, power_address=None, power_id=None,
power_user=None, power_pass=None, **extra):
"""Issue fence_cdu command for the given power change."""
try:
stdout = call_and_check([
fence_cdu, '-a', power_address, '-n', power_id, '-l',
power_user, '-p', power_pass, '-o', command],
env=select_c_utf8_locale())
except ExternalProcessError as e:
# XXX 2016-01-08 newell-jensen, bug=1532310:
# fence-agents fence_action method returns an exit code
# of 2, by default, for querying power status while machine
# is OFF.
if e.returncode == 2 and command == 'status':
return "Status: OFF\n"
else:
raise PowerError(
"Fence CDU failed issuing command %s for Power ID %s: %s"
% (command, power_id, e.output_as_unicode))
else:
return stdout.decode("utf-8")
def power_on(self, system_id, context):
"""Power ON Fence CDU power_id."""
if self.power_query(system_id, context) == 'on':
self.power_off(system_id, context)
sleep(1)
if self.power_query(system_id, context) != 'off':
raise PowerError(
"Fence CDU unable to power off Power ID %s."
% context['power_id'])
self._issue_fence_cdu_command('on', **context)
def power_off(self, system_id, context):
"""Power OFF Fence CDU power_id."""
self._issue_fence_cdu_command('off', **context)
def power_query(self, system_id, context):
"""Power QUERY Fence CDU power_id."""
re_status = re.compile(
r"Status: \s* \b(ON|OFF)\b",
re.VERBOSE | re.IGNORECASE)
query_output = self._issue_fence_cdu_command('status', **context)
# Power query output is `Status: OFF\n` or `Status: ON\n`
match = re_status.match(query_output)
if match is None:
raise PowerError(
"Fence CDU obtained unexpected response to query of "
"Power ID %s: %r" % (context['power_id'], query_output))
else:
return match.group(1).lower()
|