in source/lambda/configuration/instance_schedule.py [0:0]
def get_desired_state(self, instance, logger=None, dt=None, check_adjacent_periods=True):
"""
Test if an instance should be running at a specific moment in this schedule
:param instance: the instance to test
:param logger: logger for logging output of scheduling logic
:param dt: date time to use for scheduling, use None for using the time specified in the constructor of the schedule
:param check_adjacent_periods: check for adjacent periods in a schedule
:return: desired state, instance type and name of the active period of the schedule if the state is running
"""
# gets the local time using the configured timezone
def get_check_time(time):
check_time = time if time else self.schedule_dt
return check_time.astimezone(pytz.timezone(self.timezone))
# actions for desired state is running
def handle_running_state(inst, periods):
# used to determining most nearest period if more than one period returns a running state in a schedule
def latest_starttime(p1, p2):
if p1["period"].begintime is None:
return p2
if p2["period"].begintime is None:
return p1
return p1 if p1["period"].begintime > p2["period"].begintime else p2
# test if we need to change the type of the instance
def requires_adjust_instance_size(desired_instance_type, checked_instance):
return checked_instance.allow_resize and desired_instance_type is not None and checked_instance.is_running and \
desired_instance_type != checked_instance.instancetype
# reduce is removed from python3, replace by minimal implementation for python3 compatibility
def _reduce(fn, items):
if items is None or len(list(items)) == 0:
return None
else:
result = items[0]
i = 1
while i < len(items):
result = fn(result, items[i])
i += 1
return result
# nearest period in schedule with running state
current_running_period = _reduce(latest_starttime, periods)
multiple_active_periods = len(list(periods)) > 1
self._log_debug(DEBUG_ACTIVE_PERIOD_IN_SCHEDULE.format("s" if multiple_active_periods else "", self.name,
",".join('"' + per["period"].name + '"' for per in periods)))
if multiple_active_periods:
self._log_debug(DEBUG_USED_PERIOD.format(current_running_period["period"].name))
desired_state = InstanceSchedule.STATE_RUNNING
desired_type = current_running_period["instancetype"] if inst.allow_resize else None
# check if the instance type matches the desired type, if not set the status to stopped if the instance is currently
# and the instance will be started with the desired type at the next invocation
if requires_adjust_instance_size(desired_type, inst):
desired_state = InstanceSchedule.STATE_STOPPED_FOR_RESIZE
self._log_debug(DEBUG_SET_DESIRED_INSTANCE_TYPE, inst.instancetype, desired_type, desired_state)
return desired_state, desired_type, current_running_period["period"].name
# actions for desired state is any state
def handle_any_state():
desired_state = InstanceSchedule.STATE_ANY
self._log_debug(DEBUG_STATE_ANY, self.name, desired_state)
return desired_state, None, None
# actions for desired state is stopped
def handle_stopped_state():
desired_state = InstanceSchedule.STATE_STOPPED
self._log_debug(DEBUG_NO_RUNNING_PERIODS, self.name, desired_state)
return desired_state, None, None
# actions if there is an override value set for the schema
def handle_override_status():
desired_state = InstanceSchedule.STATE_RUNNING if self.override_status == configuration.OVERRIDE_STATUS_RUNNING \
else InstanceSchedule.STATE_STOPPED
self._log_debug(DEBUG_OVERRIDE_STATUS, self.override_status, desired_state)
return desired_state, None, "override_status"
self._logger = logger
# always on or off
if self.override_status is not None:
return handle_override_status()
# test if time is withing any period of the schedule
localized_time = get_check_time(dt)
self._log_debug(DEBUG_USED_TIME_FOR_SCHEDULE, localized_time.strftime("%c"))
# get the desired state for every period in the schedule
periods_with_desired_states = self.get_periods_with_desired_states(localized_time)
# get periods from the schema that have a running state
periods_with_running_state = [p for p in periods_with_desired_states if p["state"] == InstanceSchedule.STATE_RUNNING]
if any(periods_with_running_state):
return handle_running_state(instance, periods_with_running_state)
period_with_any_state = filter(lambda period: period["state"] == InstanceSchedule.STATE_ANY, periods_with_desired_states)
if any(period_with_any_state):
return handle_any_state()
if len(periods_with_desired_states) > 1 and check_adjacent_periods:
self._log_debug("Checking for adjacent running periods at current time")
self._log_debug("Checking states for previous minute")
last_minute_running_periods = [p for p in self.get_periods_with_desired_states(localized_time - timedelta(minutes=1)) if
p["state"] == InstanceSchedule.STATE_RUNNING]
self._log_debug("Running period(s) for previous minute {}",
",".join([p["period"].name for p in last_minute_running_periods]))
if len(last_minute_running_periods) > 0:
self._log_debug("Checking states for next minute")
next_minute_running_periods = [p for p in
self.get_periods_with_desired_states(localized_time + timedelta(minutes=1)) if
p["state"] == InstanceSchedule.STATE_RUNNING]
self._log_debug("Running period(s) for next minute {}",
",".join([p["period"].name for p in next_minute_running_periods]))
if len(next_minute_running_periods):
self._log_debug("Adjacent periods found, keep instance in running state")
return handle_running_state(instance, last_minute_running_periods)
return handle_stopped_state()