code: def run_job(
      self,
      run: Run,
      job: Job,
      instance_offer: InstanceOfferWithAvailability,
      project_ssh_public_key: str,
      project_ssh_private_key: str,
  ) -> LaunchedInstanceInfo:
      project_id = run.project_name
      instance_name = get_instance_name(run, job)
      gcp_resources.create_runner_firewall_rules(
          firewalls_client=self.firewalls_client,
          project_id=self.config.project_id,
      )
      disk_size = round(
          instance_offer.instance.resources.disk.size_mib / 1024
      )
      for zone in _get_instance_zones(instance_offer):
          request = compute_v1.InsertInstanceRequest()
          request.zone = zone
          request.project = self.config.project_id
          request.instance_resource = (
              gcp_resources.create_instance_struct(
                  disk_size=disk_size,
                  image_id=gcp_resources.get_image_id(
                      len(instance_offer.instance.resources.gpus) > 0,
                  ),
                  machine_type=instance_offer.instance.name,
                  accelerators=gcp_resources.get_accelerators(
                      project_id=self.config.project_id,
                      zone=zone,
                      gpus=instance_offer.instance.resources.gpus,
                  ),
                  spot=instance_offer.instance.resources.spot,
                  user_data=get_user_data(
                      backend=BackendType.GCP,
                      image_name=job.job_spec.image_name,
                      authorized_keys=[
                          run.run_spec.ssh_key_pub.strip(),
                          project_ssh_public_key.strip(),
                      ],
                      registry_auth_required=job.job_spec.registry_auth
                      is not None,
                  ),
                  labels={
                      "owner": "dstack",
                      "dstack_project": project_id,
                      "dstack_user": run.user,
                  },
                  tags=[gcp_resources.DSTACK_INSTANCE_TAG],
                  instance_name=instance_name,
                  zone=zone,
                  service_account=self.config.service_account_email,
              )
          )
          try:
              operation = self.instances_client.insert(request=request)
              gcp_resources.wait_for_extended_operation(
                  operation, "instance creation"
              )
          except (
              google.api_core.exceptions.ServiceUnavailable,
              google.api_core.exceptions.NotFound,
          ):
              continue
          instance = self.instances_client.get(
              project=self.config.project_id,
              zone=zone,
              instance=instance_name,
          )
          return LaunchedInstanceInfo(
              instance_id=instance_name,
              region=zone,
              ip_address=instance.network_interfaces[0]
              .access_configs[0]
              .nat_i_p,
              username="ubuntu",
              ssh_port=22,
              dockerized=True,
              backend_data=None,
          )
      raise NoCapacityError()
  
imports: from collections import defaultdict
  from typing import Callable, Dict, List, Optional, Tuple
  import google.api_core.exceptions
  import google.cloud.compute_v1
  import dstack._internal.core.backends.gcp.auth
  import dstack._internal.core.backends.gcp.resources
  from dstack._internal.core.backends.base.compute import Compute, get_gateway_user_data, get_instance_name, get_user_data
  from dstack._internal.core.backends.base.offers import get_catalog_offers
  from dstack._internal.core.backends.gcp.config import GCPConfig
  from dstack._internal.core.errors import NoCapacityError, ResourceNotFoundError
  from dstack._internal.core.models.backends.base import BackendType
  from dstack._internal.core.models.instances import InstanceAvailability, InstanceOffer, InstanceOfferWithAvailability, InstanceType, LaunchedGatewayInfo, LaunchedInstanceInfo, Resources
  from dstack._internal.core.models.runs import Job, Requirements, Run
label_desc: Use image node_id 4 for calling the GCPCompute obj's underlying member method code with example usage: obj.run_job(run, job, instance_offer, project_ssh_public_key, project_ssh_private_key) and returns: LaunchedInstanceInfo