Source code for ironic.common.fsm
# Copyright (C) 2014 Yahoo! Inc. All Rights Reserved.
#
# 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.
import functools
from automaton import exceptions as automaton_exceptions
from automaton import machines
"""State machine modelling.
This is being used in the implementation of:
http://specs.openstack.org/openstack/ironic-specs/specs/kilo/new-ironic-state-machine.html
"""
from ironic.common import exception as excp
from ironic.common.i18n import _
def _translate_excp(func):
"""Decorator to translate automaton exceptions into ironic exceptions."""
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except (automaton_exceptions.InvalidState,
automaton_exceptions.NotInitialized,
automaton_exceptions.FrozenMachine,
automaton_exceptions.NotFound) as e:
raise excp.InvalidState(str(e))
except automaton_exceptions.Duplicate as e:
raise excp.Duplicate(str(e))
return wrapper
[docs]
class FSM(machines.FiniteMachine):
"""An ironic state-machine class with some ironic specific additions."""
def __init__(self):
super(FSM, self).__init__()
self._target_state = None
# For now make these raise ironic state machine exceptions until
# a later period where these should(?) be using the raised automaton
# exceptions directly.
add_transition = _translate_excp(machines.FiniteMachine.add_transition)
@property
def target_state(self):
return self._target_state
[docs]
def is_stable(self, state):
"""Is the state stable?
:param state: the state of interest
:raises: InvalidState if the state is invalid
:returns: True if it is a stable state; False otherwise
"""
try:
return self._states[state]['stable']
except KeyError:
raise excp.InvalidState(_("State '%s' does not exist") % state)
[docs]
@_translate_excp
def add_state(self, state, on_enter=None, on_exit=None,
target=None, terminal=None, stable=False):
"""Adds a given state to the state machine.
:param stable: Use this to specify that this state is a stable/passive
state. A state must have been previously defined as
'stable' before it can be used as a 'target'
:param target: The target state for 'state' to go to. Before a state
can be used as a target it must have been previously
added and specified as 'stable'
Further arguments are interpreted as for parent method ``add_state``.
"""
self._validate_target_state(target)
super(FSM, self).add_state(state, terminal=terminal,
on_enter=on_enter, on_exit=on_exit)
self._states[state].update({
'stable': stable,
'target': target,
})
def _post_process_event(self, event, result):
# Clear '_target_state' if we've reached it
if (self._target_state is not None
and self._target_state == self._current.name):
self._target_state = None
# If new state has a different target, update the '_target_state'
if self._states[self._current.name]['target'] is not None:
self._target_state = self._states[self._current.name]['target']
def _validate_target_state(self, target):
"""Validate the target state.
A target state must be a valid state that is 'stable'.
:param target: The target state
:raises: exception.InvalidState if it is an invalid target state
"""
if target is None:
return
if target not in self._states:
raise excp.InvalidState(
_("Target state '%s' does not exist") % target)
if not self.is_stable(target):
raise excp.InvalidState(
_("Target state '%s' is not a 'stable' state") % target)
[docs]
@_translate_excp
def initialize(self, start_state=None, target_state=None):
"""Initialize the FSM.
:param start_state: the FSM is initialized to start from this state
:param target_state: if specified, the FSM is initialized to this
target state. Otherwise use the default target
state
"""
super(FSM, self).initialize(start_state=start_state)
current_state = self._current.name
self._validate_target_state(target_state)
self._target_state = (target_state
or self._states[current_state]['target'])
[docs]
@_translate_excp
def process_event(self, event, target_state=None):
"""process the event.
:param event: the event to be processed
:param target_state: if specified, the 'final' target state for the
event. Otherwise, use the default target state
"""
super(FSM, self).process_event(event)
if target_state:
# NOTE(rloo): _post_process_event() was invoked at the end of
# the above super().process_event() call. At this
# point, the default target state is being used but
# we want to use the specified state instead.
self._validate_target_state(target_state)
self._target_state = target_state