in source/lambda/schedulers/ec2_service.py [0:0]
def get_schedulable_instances(self, kwargs):
self._session = kwargs[schedulers.PARAM_SESSION]
context = kwargs[schedulers.PARAM_CONTEXT]
region = kwargs[schedulers.PARAM_REGION]
account = kwargs[schedulers.PARAM_ACCOUNT]
self._logger = kwargs[schedulers.PARAM_LOGGER]
tagname = kwargs[schedulers.PARAM_CONFIG].tag_name
config = kwargs[schedulers.PARAM_CONFIG]
self.schedules_with_hibernation = [s.name for s in config.schedules.values() if s.hibernate]
self._logger.info("Enable SSM Maintenance window is set to {}", config.enable_SSM_maintenance_windows)
if config.enable_SSM_maintenance_windows:
# calling the get maintenance window for this account and region.
self._logger.debug("load the ssm maintenance windows for account {}, and region {}", account, region)
self._ssm_maintenance_windows = self.ssm_maintenance_windows(self._session, context, account, region)
self._logger.debug("finish loading the ssm maintenance windows")
client = get_client_with_retries("ec2", ["describe_instances"], context=context, session=self._session,
region=region)
def is_in_schedulable_state(ec2_inst):
state = ec2_inst["state"] & 0xFF
return state in Ec2Service.EC2_SCHEDULABLE_STATES
jmes = "Reservations[*].Instances[*].{InstanceId:InstanceId, EbsOptimized:EbsOptimized, Tags:Tags, " \
"InstanceType:InstanceType,State:State}[]" + \
"|[?Tags]|[?contains(Tags[*].Key, '{}')]".format(tagname)
args = {}
number_of_instances = 0
instances = []
done = False
self._logger.info(INF_FETCHING_INSTANCES, account, region)
while not done:
ec2_resp = client.describe_instances_with_retries(**args)
for reservation_inst in jmespath.search(jmes, ec2_resp):
inst = self._select_instance_data(instance=reservation_inst, tagname=tagname, config=config)
number_of_instances += 1
if is_in_schedulable_state(inst):
instances.append(inst)
self._logger.debug(DEBUG_SELECTED_INSTANCE, inst[schedulers.INST_ID],
inst[schedulers.INST_STATE_NAME])
else:
self._logger.debug(DEBUG_SKIPPED_INSTANCE, inst[schedulers.INST_ID],
inst[schedulers.INST_STATE_NAME])
if "NextToken" in ec2_resp:
args["NextToken"] = ec2_resp["NextToken"]
else:
done = True
self._logger.info(INF_FETCHED_INSTANCES, number_of_instances, len(instances))
return instances