in source/code/actions/ec2_create_snapshot_action.py [0:0]
def create_volume_snapshot(self, volume):
def create_snapshot(vol, snapshot_description):
snapshot_id = ""
try:
create_snapshot_resp = self.ec2_client.create_snapshot_with_retries(DryRun=self._dryrun_, VolumeId=vol,
Description=snapshot_description)
self.result["volumes"][vol] = {}
self.result["volumes"][vol]["create_snapshot"] = create_snapshot_resp
snapshot_id = create_snapshot_resp["SnapshotId"]
self.result["volumes"][vol]["snapshot"] = snapshot_id
self._logger_.info(INFO_SNAPSHOT_CREATED, snapshot_id)
except Exception as ex:
if self._dryrun_:
self._logger_.info(str(ex))
self.result["volumes"][volume]["create_snapshot"] = str(ex)
else:
raise ex
return snapshot_id
def set_snapshot_tags(snap, vol, dev):
try:
tags = get_tags_for_volume_snapshot(vol, dev)
if self.set_snapshot_name:
snapshot_name = self.build_str_from_template(parameter_name=PARAM_NAME,
tag_variables={
TAG_PLACEHOLDER_INSTANCE_ID: self.instance_id,
TAG_PLACEHOLDER_VOLUME_ID: volume
})
if snapshot_name == "":
dt = self._datetime_.utcnow()
snapshot_name = SNAPSHOT_NAME.format(volume, dt.year, dt.month, dt.day, dt.hour, dt.minute)
prefix = self.build_str_from_template(parameter_name=PARAM_SNAPSHOT_NAME_PREFIX,
tag_variables={
TAG_PLACEHOLDER_INSTANCE_ID: self.instance_id,
TAG_PLACEHOLDER_VOLUME_ID: volume
})
snapshot_name = prefix + snapshot_name
tags["Name"] = snapshot_name
self._logger_.info(INFO_SNAPSHOT_NAME, snapshot_name)
if len(tags) > 0:
self._logger_.info(INFO_CREATE_TAGS, safe_json(tags, indent=3))
tagging.set_ec2_tags(ec2_client=self.ec2_client,
resource_ids=[snap],
tags=tags,
can_delete=False,
logger=self._logger_)
if snap not in self.result["snapshots"]:
self.result["snapshots"][snap] = {}
self.result["snapshots"][snap]["tags"] = tags
self._logger_.info(INFO_TAGS_CREATED)
except Exception as ex:
if self._dryrun_:
self._logger_.debug(str(ex))
self.result["volumes"][volume]["create_tags"] = str(ex)
else:
raise ex
def get_tags_for_volume_snapshot(vol, dev):
vol_tags = self.copied_instance_tagfilter.pairs_matching_any_filter(self.tags_on_instance)
tags_on_volume = self.all_volume_tags.get(vol, {})
vol_tags.update(self.copied_volume_tagfilter.pairs_matching_any_filter(tags_on_volume))
vol_tags.update(
self.build_tags_from_template(parameter_name=PARAM_SNAPSHOT_TAGS,
tag_variables={
TAG_PLACEHOLDER_INSTANCE_ID: self.instance_id,
TAG_PLACEHOLDER_VOLUME_ID: volume,
TAG_PLACEHOLDER_DEVICE: dev
}))
vol_tags[actions.marker_snapshot_tag_source_source_volume_id()] = volume
return vol_tags
device = self.volumes[volume]
self.result[volume] = {"device": device}
description = self.build_str_from_template(parameter_name=PARAM_SNAPSHOT_DESCRIPTION,
tag_variables={
TAG_PLACEHOLDER_INSTANCE_ID: self.instance_id,
TAG_PLACEHOLDER_VOLUME_ID: volume,
TAG_PLACEHOLDER_DEVICE: device
})
if description == "":
description = SNAPSHOT_DESCRIPTION.format(self._task_, "root " if volume == self.root_volume else "", volume, device,
self.instance_id)
self._logger_.info(INFO_CREATE_SNAPSHOT, volume, "root " if volume == self.root_volume else "", device, self.instance_id)
snapshot = create_snapshot(volume, description)
set_snapshot_tags(snapshot, volume, device)