Cloud FunctionsでVMインスタンスを自動停止、起動する手順

import json
import sys
import base64
import googleapiclient.discovery
from typing import Any


from google.api_core.extended_operation import ExtendedOperation
from google.cloud import compute_v1


compute = googleapiclient.discovery.build('compute', 'v1')
project = "プロジェクト名"
zone = "asia-northeast1-b"




def wait_for_extended_operation(
    operation: ExtendedOperation, verbose_name: str = "operation", timeout: int = 300
) -> Any:
    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




def get_data(event):
    if 'data' in event:
        data = json.loads(base64.b64decode(event['data']).decode('utf-8'))
    else:
        raise RuntimeError('No data in event')
    return data




def get_label(data):
    if 'label' in data:
        label = data['label']
    else:
        raise RuntimeError('No label in data')
    return label




def list_instance_by_label(label_name, label_value):
    label_filter = "labels.{0}={1}".format(label_name, label_value)
    response = compute.instances().list(project=project, zone=zone, filter=label_filter).execute()
    print('Instances in project "%s" and zone "%s" with label "%s" are:' % (project, project, label_name))
    inst_name_list = []
    if (response.get('items')):
        for instance in response['items']:
            print(' - Id:           ' + instance['id'])
            print('   Name:         ' + instance['name'])
            print('   Status:       ' + instance['status'])
            print('   Machine type: ' + instance['machineType'])
            inst_name_list.append(instance['name'])
    else:
        print('NO instances matched')
    return inst_name_list




def start_vm_pubsub(event, context):
    print("EVENT", event)
    data = get_data(event)
    label = get_label(data)
    key = label.split('=')[0]
    val = label.split('=')[1]
    inst_name_list = list_instance_by_label(key, val)
    if len(inst_name_list) == 0:
        print("No matched instance, skip triggering start/stop.")
        return
    for instance_name in inst_name_list:
        print('Instance name is: ' + instance_name)
        instance_client = compute_v1.InstancesClient()
        operation = instance_client.start(project=project, zone=zone, instance=instance_name)
        wait_for_extended_operation(operation, "instance "+ instance_name+" starting")

# Function dependencies, for example:
# package>=version
cachetools==5.2.0
certifi==2022.6.15
charset-normalizer==2.1.0
google-api-core==2.8.2
google-api-python-client==2.56.0
google-auth==2.10.0
google-auth-httplib2==0.1.0
google-cloud-compute==1.5.0
googleapis-common-protos==1.56.4
grpcio==1.47.0
grpcio-status==1.47.0
httplib2==0.20.4
idna==3.3
proto-plus==1.20.6
protobuf==3.20.1
pyasn1==0.4.8
pyasn1-modules==0.2.8
pyparsing==3.0.9
requests==2.28.1
rsa==4.9
six==1.16.0
uritemplate==4.1.1
urllib3==1.26.11

import googleapiclient.discovery
from typing import Any


from google.api_core.extended_operation import ExtendedOperation
from google.cloud import compute_v1


compute = googleapiclient.discovery.build('compute', 'v1')
project = "プロジェクト名"
zone = "asia-northeast1-b"




def wait_for_extended_operation(
    operation: ExtendedOperation, verbose_name: str = "operation", timeout: int = 300
) -> Any:
    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




def get_data(event):
    if 'data' in event:
        data = json.loads(base64.b64decode(event['data']).decode('utf-8'))
    else:
        raise RuntimeError('No data in event')
    return data




def get_label(data):
    if 'label' in data:
        label = data['label']
    else:
        raise RuntimeError('No label in data')
    return label




def list_instance_by_label(label_name, label_value):
    label_filter = "labels.{0}={1}".format(label_name, label_value)
    response = compute.instances().list(project=project, zone=zone, filter=label_filter).execute()
    print('Instances in project "%s" and zone "%s" with label "%s" are:' % (project, project, label_name))
    inst_name_list = []
    if (response.get('items')):
        for instance in response['items']:
            print(' - Id:           ' + instance['id'])
            print('   Name:         ' + instance['name'])
            print('   Status:       ' + instance['status'])
            print('   Machine type: ' + instance['machineType'])
            inst_name_list.append(instance['name'])
    else:
        print('NO instances matched')
    return inst_name_list




def stop_vm_pubsub(event, context):
    print("EVENT", event)
    data = get_data(event)
    label = get_label(data)
    key = label.split('=')[0]
    val = label.split('=')[1]
    inst_name_list = list_instance_by_label(key, val)
    if len(inst_name_list) == 0:
        print("No matched instance, skip triggering stop.")
        return
    for instance_name in inst_name_list:
        print('Instance name is: ' + instance_name)
        instance_client = compute_v1.InstancesClient()
        operation = instance_client.stop(project=project, zone=zone, instance=instance_name)
        wait_for_extended_operation(operation, "instance "+ instance_name+" stopping")

# Function dependencies, for example:
# package>=version
cachetools==5.2.0
certifi==2022.6.15
charset-normalizer==2.1.0
google-api-core==2.8.2
google-api-python-client==2.56.0
google-auth==2.10.0
google-auth-httplib2==0.1.0
google-cloud-compute==1.5.0
googleapis-common-protos==1.56.4
grpcio==1.47.0
grpcio-status==1.47.0
httplib2==0.20.4
idna==3.3
proto-plus==1.20.6
protobuf==3.20.1
pyasn1==0.4.8
pyasn1-modules==0.2.8
pyparsing==3.0.9
requests==2.28.1
rsa==4.9
six==1.16.0
uritemplate==4.1.1
urllib3==1.26.11