"""State module for managing Snapshots."""
__contracts__ = ["resource"]
from dataclasses import make_dataclass, field
from typing import Dict, Any, List
from dataclasses import field
from dataclasses import make_dataclass
from typing import Any
from typing import Dict
from typing import List
from idem_gcp.tool.gcp.utils import global_absent
# prevent commit hook from removing the import
absent = global_absent
__contracts__ = ["resource"]
[docs]async def present(
hub,
ctx,
name: str,
project: str = None,
resource_id: str = None,
source_disk: str = None,
storage_locations: List[str] = None,
location_hint: str = None,
label_fingerprint: str = None,
description: str = None,
labels: Dict[str, Any] = None,
source_disk_encryption_key: make_dataclass(
"CustomerEncryptionKey",
[
("kms_key_service_account", str, field(default=None)),
("rsa_encrypted_key", str, field(default=None)),
("kms_key_name", str, field(default=None)),
("raw_key", str, field(default=None)),
],
) = None,
chain_name: str = None,
snapshot_type: str = None,
snapshot_encryption_key: make_dataclass(
"CustomerEncryptionKey",
[
("kms_key_service_account", str, field(default=None)),
("rsa_encrypted_key", str, field(default=None)),
("kms_key_name", str, field(default=None)),
("raw_key", str, field(default=None)),
],
) = None,
request_id: str = None,
) -> Dict[str, Any]:
r"""Creates a snapshot in the specified project using the data included in the request.
For regular snapshot creation, consider using this method instead of disks.createSnapshot, as this method supports more features, such as creating snapshots in a project different from the source disk project.
Args:
storage_locations(List[str], Optional):
Cloud Storage bucket storage location of the snapshot (regional or multi-regional). Defaults to None.
location_hint(str, Optional):
An opaque location hint used to place the snapshot close to other resources. This field is for use by internal tools that use the public API. Defaults to None.
label_fingerprint(str, Optional):
A fingerprint for the labels being applied to this snapshot, which is essentially a hash of the labels set used for optimistic locking. The fingerprint is initially generated by Compute Engine and changes after every request to modify or update labels. You must always provide an up-to-date fingerprint hash in order to update or change labels, otherwise the request will fail with error 412 conditionNotMet. To see the latest fingerprint, make a get() request to retrieve a snapshot. Defaults to None.
description(str, Optional):
An optional description of this resource. Provide this property when you create the resource. Defaults to None.
labels(Dict[str, Any], Optional):
Labels to apply to this snapshot. These can be later modified by the setLabels method. Label values may be empty. Defaults to None.
source_disk_encryption_key(Dict[str, Any], Optional):
The customer-supplied encryption key of the source disk. Required if the source disk is protected by a customer-supplied encryption key. Defaults to None.
* kms_key_service_account(str, Optional):
The service account being used for the encryption request for the given KMS key. If absent, the Compute Engine default service account is used. For example: "kmsKeyServiceAccount": "name@project_id.iam.gserviceaccount.com/"
* sha256(str, Optional):
[Output only] The RFC 4648 base64 encoded SHA-256 hash of the customer-supplied encryption key that protects this resource.
* rsa_encrypted_key(str, Optional):
Specifies an RFC 4648 base64 encoded, RSA-wrapped 2048-bit customer-supplied encryption key to either encrypt or decrypt this resource. You can provide either the rawKey or the rsaEncryptedKey. For example: "rsaEncryptedKey": "ieCx/NcW06PcT7Ep1X6LUTc/hLvUDYyzSZPPVCVPTVEohpeHASqC8uw5TzyO9U+Fka9JFH z0mBibXUInrC/jEk014kCK/NPjYgEMOyssZ4ZINPKxlUh2zn1bV+MCaTICrdmuSBTWlUUiFoD D6PYznLwh8ZNdaheCeZ8ewEXgFQ8V+sDroLaN3Xs3MDTXQEMMoNUXMCZEIpg9Vtp9x2oe==" The key must meet the following requirements before you can provide it to Compute Engine: 1. The key is wrapped using a RSA public key certificate provided by Google. 2. After being wrapped, the key must be encoded in RFC 4648 base64 encoding. Gets the RSA public key certificate provided by Google at: https://cloud-certs.storage.googleapis.com/google-cloud-csek-ingress.pem
* kms_key_name(str, Optional):
The name of the encryption key that is stored in Google Cloud KMS. For example: "kmsKeyName": "projects/kms_project_id/locations/region/keyRings/ key_region/cryptoKeys/key
* raw_key(str, Optional):
Specifies a 256-bit customer-supplied encryption key, encoded in RFC 4648 base64 to either encrypt or decrypt this resource. You can provide either the rawKey or the rsaEncryptedKey. For example: "rawKey": "SGVsbG8gZnJvbSBHb29nbGUgQ2xvdWQgUGxhdGZvcm0="
source_disk(str, Optional):
The source disk used to create this snapshot. Defaults to None.
chain_name(str, Optional):
Creates the new snapshot in the snapshot chain labeled with the specified name. The chain name must be 1-63 characters long and comply with RFC1035. This is an uncommon option only for advanced service owners who needs to create separate snapshot chains, for example, for chargeback tracking. When you describe your snapshot resource, this field is visible only if it has a non-empty value. Defaults to None.
snapshot_type(str, Optional):
Indicates the type of the snapshot.
Enum type. Allowed values:
"ARCHIVE"
"STANDARD". Defaults to None.
snapshot_encryption_key(Dict[str, Any], Optional):
Encrypts the snapshot using a customer-supplied encryption key. After you encrypt a snapshot using a customer-supplied key, you must provide the same key if you use the snapshot later. For example, you must provide the encryption key when you create a disk from the encrypted snapshot in a future request. Customer-supplied encryption keys do not protect access to metadata of the snapshot. If you do not provide an encryption key when creating the snapshot, then the snapshot will be encrypted using an automatically generated key and you do not need to provide a key to use the snapshot later. Defaults to None.
* kms_key_service_account(str, Optional):
The service account being used for the encryption request for the given KMS key. If absent, the Compute Engine default service account is used. For example: "kmsKeyServiceAccount": "name@project_id.iam.gserviceaccount.com/"
* sha256 (str, Optional):
[Output only] The RFC 4648 base64 encoded SHA-256 hash of the customer-supplied encryption key that protects this resource.
* rsa_encrypted_key(str, Optional):
Specifies an RFC 4648 base64 encoded, RSA-wrapped 2048-bit customer-supplied encryption key to either encrypt or decrypt this resource. You can provide either the rawKey or the rsaEncryptedKey. For example: "rsaEncryptedKey": "ieCx/NcW06PcT7Ep1X6LUTc/hLvUDYyzSZPPVCVPTVEohpeHASqC8uw5TzyO9U+Fka9JFH z0mBibXUInrC/jEk014kCK/NPjYgEMOyssZ4ZINPKxlUh2zn1bV+MCaTICrdmuSBTWlUUiFoD D6PYznLwh8ZNdaheCeZ8ewEXgFQ8V+sDroLaN3Xs3MDTXQEMMoNUXMCZEIpg9Vtp9x2oe==" The key must meet the following requirements before you can provide it to Compute Engine: 1. The key is wrapped using a RSA public key certificate provided by Google. 2. After being wrapped, the key must be encoded in RFC 4648 base64 encoding. Gets the RSA public key certificate provided by Google at: https://cloud-certs.storage.googleapis.com/google-cloud-csek-ingress.pem
* kms_key_name(str, Optional):
The name of the encryption key that is stored in Google Cloud KMS. For example: "kmsKeyName": "projects/kms_project_id/locations/region/keyRings/ key_region/cryptoKeys/key
* raw_key(str, Optional):
Specifies a 256-bit customer-supplied encryption key, encoded in RFC 4648 base64 to either encrypt or decrypt this resource. You can provide either the rawKey or the rsaEncryptedKey. For example: "rawKey": "SGVsbG8gZnJvbSBHb29nbGUgQ2xvdWQgUGxhdGZvcm0="
request_id(str, Optional):
An optional request ID to identify requests. Specify a unique request ID so that if you must retry your request, the server will know to ignore the request if it has already been completed. For example, consider a situation where you make an initial request and the request times out. If you make the request again with the same request ID, the server can check if original operation with the same request ID was received, and if so, will ignore the second request. This prevents clients from accidentally creating duplicate commitments. The request ID must be a valid UUID with the exception that zero UUID is not supported ( 00000000-0000-0000-0000-000000000000). Defaults to None.
project(str):
Project ID for this request.
name(str):
Name of the Snapshot resource to return.
resource_id(str, Optional):
An identifier of the resource in the provider. Defaults to None.
Returns:
Dict[str, Any]
Examples:
.. code-block:: sls
resource_is_present:
gcp.compute.snapshot.present:
- name: value
- project: value
- snapshot: value
"""
result = {
"result": True,
"old_state": None,
"new_state": None,
"name": name,
"comment": [],
}
project = hub.tool.gcp.utils.get_project_from_account(ctx, project)
if ctx.get("wrapper_result"):
result = ctx.get("wrapper_result")
# to be autogenerated by pop-create based on insert/update props in properties.yaml
resource_body = {
"name": name,
"storage_locations": storage_locations,
"location_hint": location_hint,
"label_fingerprint": label_fingerprint,
"description": description,
"labels": labels,
"source_disk_encryption_key": source_disk_encryption_key,
"source_disk": source_disk,
"chain_name": chain_name,
"snapshot_type": snapshot_type,
"snapshot_encryption_key": snapshot_encryption_key,
}
resource_body = {k: v for (k, v) in resource_body.items() if v is not None}
operation = None
if result["old_state"]:
resource_body["label_fingerprint"] = label_fingerprint or result[
"old_state"
].get("label_fingerprint")
changes = hub.tool.gcp.utils.compare_states(
result["old_state"],
{"resource_id": resource_id, **resource_body},
"compute.snapshot",
)
# This check is necessary because "changes" does not detect label removal
are_labels_changed = (
labels is not None
and not (labels == {} and result["old_state"].get("labels") is None)
and result["old_state"].get("labels") != labels
)
if not changes and not are_labels_changed:
result["comment"].append(
hub.tool.gcp.comment_utils.up_to_date_comment(
"gcp.compute.snapshot", name
)
)
result["new_state"] = result["old_state"]
return result
changed_non_updatable_properties = (
hub.tool.gcp.resource_prop_utils.get_changed_non_updatable_properties(
"compute.snapshot", changes
)
)
if changed_non_updatable_properties:
result["result"] = False
result["comment"].append(
hub.tool.gcp.comment_utils.non_updatable_properties_comment(
"gcp.compute.snapshot",
name,
changed_non_updatable_properties,
)
)
result["new_state"] = result["old_state"]
return result
if ctx.get("test", False):
result["comment"].append(
hub.tool.gcp.comment_utils.would_update_comment(
"gcp.compute.snapshot", name
)
)
result["new_state"] = hub.tool.gcp.sanitizers.sanitize_resource_urls(
resource_body
)
result["new_state"]["resource_id"] = resource_id
return result
set_labels_body = {
"labels": labels,
}
if label_fingerprint:
set_labels_body.update(
{
"label_fingerprint": label_fingerprint,
}
)
elif result["old_state"].get("label_fingerprint"):
set_labels_body.update(
{
"label_fingerprint": result["old_state"].get("label_fingerprint"),
}
)
# the setLabels() method accepts "resource" as the name of the snapshot, so
# the resource_id is split into its properties ("project" and "snapshot") here to map "resource" to "snapshot
resource_id_props = (
hub.tool.gcp.resource_prop_utils.get_elements_from_resource_id(
"compute.snapshot", resource_id
)
)
update_ret = await hub.exec.gcp_api.client.compute.snapshot.setLabels(
ctx,
project=resource_id_props.get("project"),
resource=resource_id_props.get("snapshot"),
body=set_labels_body,
)
if not update_ret["result"] or not update_ret["ret"]:
result["result"] = False
result["comment"] += update_ret["comment"]
return result
if hub.tool.gcp.operation_utils.is_operation(update_ret["ret"]):
operation = update_ret["ret"]
else:
if ctx["test"]:
result["comment"].append(
hub.tool.gcp.comment_utils.would_create_comment(
"gcp.compute.snapshot", name
)
)
result["new_state"] = hub.tool.gcp.sanitizers.sanitize_resource_urls(
resource_body
)
result["new_state"]["resource_id"] = resource_id
return result
# Create
create_ret = await hub.exec.gcp_api.client.compute.snapshot.insert(
ctx,
name=name,
project=project,
request_id=request_id,
body=resource_body,
)
if not create_ret["result"]:
result["result"] = False
if create_ret["comment"] and next(
(
comment
for comment in create_ret["comment"]
if "alreadyExists" in comment
),
None,
):
result["comment"].append(
hub.tool.gcp.comment_utils.already_exists_comment(
"gcp.compute.snapshot", name
)
)
return result
else:
result["comment"] += create_ret["comment"]
if create_ret["ret"]:
if hub.tool.gcp.operation_utils.is_operation(create_ret["ret"]):
operation = create_ret["ret"]
if operation:
operation_id = hub.tool.gcp.resource_prop_utils.parse_link_to_resource_id(
operation.get("selfLink"), "compute.global_operation"
)
result["rerun_data"] = {
"operation_id": operation_id,
"old_state": result["old_state"],
}
return result
[docs]async def describe(hub, ctx) -> Dict[str, Dict[str, Any]]:
r"""Describe the resource in a way that can be recreated/managed with the corresponding "present" function.
Retrieves the list of Snapshot resources contained within the specified project.
Returns:
Dict[str, Any]
Examples:
.. code-block:: bash
$ idem describe gcp.compute.snapshot
"""
result = {}
describe_ret = await hub.exec.gcp.compute.snapshot.list(
ctx, project=ctx.acct.project_id
)
if not describe_ret["result"]:
hub.log.debug(f"Could not describe snapshots {describe_ret['comment']}")
return {}
for resource in describe_ret["ret"]:
resource_id = resource.get("resource_id")
result[resource_id] = {
"gcp.compute.snapshot.present": [
{parameter_key: parameter_value}
for parameter_key, parameter_value in resource.items()
]
}
return result
[docs]def is_pending(hub, ret: dict, state: str = None, **pending_kwargs) -> bool:
"""Default implemented for each module."""
return hub.tool.gcp.utils.is_pending(ret=ret, state=state, **pending_kwargs)