我有一个可以创建 Google Cloud Compute Engine 实例模板的代码,但我无法设法让它创建区域实例模板(例如在 europe-north1)。
创建_模板
from __future__ import annotations
from google.cloud import compute_v1
from src.routes.run_fw_engagements.c_instance_template.helpers.wait_for_extended_operation import wait_for_extended_operation
def create_template(project_id: str, template_name: str, network_tags: list, startup_script: str, instance_template_region: str) -> compute_v1.InstanceTemplate:
"""
Create a new instance template with the provided name and a specific
instance configuration.
Args:
project_id: project ID or project number of the Cloud project you use.
template_name: name of the new template to create.
network_tags: List of network tags to apply to the instance.
startup_script: The startup script to run on instance creation.
instance_template_region: The location (zone or region) where the instances will be launched.
Returns:
InstanceTemplate object that represents the new instance template.
"""
# The template describes the size and source image of the boot disk
# to attach to the instance.
disk = compute_v1.AttachedDisk()
initialize_params = compute_v1.AttachedDiskInitializeParams()
initialize_params.source_image = (
"projects/ubuntu-os-cloud/global/images/family/ubuntu-2204-lts"
)
initialize_params.disk_size_gb = 10
disk.initialize_params = initialize_params
disk.auto_delete = True
disk.boot = True
# The template connects the instance to the default network.
network_interface = compute_v1.NetworkInterface()
network_interface.name = "global/networks/default"
# The template lets the instance use an external IP address.
access_config = compute_v1.AccessConfig()
access_config.name = "External NAT"
access_config.type_ = "ONE_TO_ONE_NAT"
access_config.network_tier = "PREMIUM"
network_interface.access_configs = [access_config]
# Add template properties
template = compute_v1.InstanceTemplate()
template.name = template_name
template.properties.disks = [disk]
template.properties.machine_type = "e2-micro"
template.properties.network_interfaces = [network_interface]
template.properties.tags = compute_v1.Tags(items=network_tags)
# Add startup script, OS Login, and location to metadata
metadata = compute_v1.Metadata()
metadata.items = [
compute_v1.Items(key="startup-script", value=startup_script),
compute_v1.Items(key="enable-oslogin", value="true"),
compute_v1.Items(key="enable-oslogin-2fa", value="true"),
compute_v1.Items(key="instance-template-region", value=instance_template_region) # Add location to metadata for reference
]
template.properties.metadata = metadata
# Create client
template_client = compute_v1.InstanceTemplatesClient()
operation = template_client.insert(
project=project_id, instance_template_resource=template
)
wait_for_extended_operation(operation, "instance template creation")
response = template_client.get(project=project_id, instance_template=template_name)
# Return the response or useful template info
return response
if __name__ == '__main__':
startup_script = """#!/bin/bash
sudo apt update
sudo apt install -y apache2
systemctl start apache2
"""
create_template(
project_id="sindre-dev",
template_name="template-test-10",
network_tags=["mytag-test"],
startup_script=startup_script,
instance_template_region="europe-north1"
)
等待扩展操作
from __future__ import annotations
import sys
from typing import Any
from google.api_core.extended_operation import ExtendedOperation
def wait_for_extended_operation(
operation: ExtendedOperation, verbose_name: str = "operation", timeout: int = 300
) -> Any:
"""
Waits for the extended (long-running) operation to complete.
If the operation is successful, it will return its result.
If the operation ends with an error, an exception will be raised.
If there were any warnings during the execution of the operation
they will be printed to sys.stderr.
Args:
operation: a long-running operation you want to wait on.
verbose_name: (optional) a more verbose name of the operation,
used only during error and warning reporting.
timeout: how long (in seconds) to wait for operation to finish.
If None, wait indefinitely.
Returns:
Whatever the operation.result() returns.
Raises:
This method will raise the exception received from `operation.exception()`
or RuntimeError if there is no exception set, but there is an `error_code`
set for the `operation`.
In case of an operation taking longer than `timeout` seconds to complete,
a `concurrent.futures.TimeoutError` will be raised.
"""
result = operation.result(timeout=timeout)
if operation.error_code:
print(
f"Error during {verbose_name}: [Code: {operation.error_code}]: {operation.error_message}",
file=sys.stderr,
flush=True,
)
print(f"Operation ID: {operation.name}", file=sys.stderr, flush=True)
raise operation.exception() or RuntimeError(operation.error_message)
if operation.warnings:
print(f"Warnings during {verbose_name}:\n", file=sys.stderr, flush=True)
for warning in operation.warnings:
print(f" - {warning.code}: {warning.message}", file=sys.stderr, flush=True)
return result
如何确保实例模板是在
instance_template_region="europe-north1"
中创建的?
要创建区域模板实例,您需要将客户端从
InstanceTemplatesClient
更改为 RegionInstanceTemplatesClient
。
因此,您只需要更改
create_template()
中您创建并与 client
交互的部分:
# Create client
template_client = compute_v1.RegionInstanceTemplatesClient() # Change client type
operation = template_client.insert(
project=project_id, region=region, instance_template_resource=template
) # Add region
wait_for_extended_operation(operation, "instance template creation")
response = template_client.get(project=project_id, region=region, instance_template=template_name) # Add region