Function update_task_enabled

src/python/bot/startup/run_bot.py:257–295

It uses the GCE VM metadata server `update_task_enabled` flag. This flag will be used to rollout the update_task deprecation by disabling it progressively for each instance group through the instance template metadata

()

def update_task_enabled() -> bool:
  """ It uses the GCE VM metadata server `update_task_enabled` flag.

      This flag will be used to rollout the update_task deprecation
      by disabling it progressively for each instance group through
      the instance template metadata 
  """
  metadata_url = ("http://metadata.google.internal/computeMetadata/v1/" +
                  "instance/attributes/")
  metadata_header = {"Metadata-Flavor": "Google"}
  metadata_key = "update_task_enabled"

  running_on_batch = bool(environment.is_uworker())

  try:
    # Construct the full URL for your specific metadata key
    response = requests.get(
        f"{metadata_url}{metadata_key}", headers=metadata_header, timeout=10)

    # Raise an exception for bad status codes (4xx or 5xx)
    response.raise_for_status()

    # The metadata value is in the response text
    metadata_value = response.text
    logs.info(f"The value for '{metadata_key}' is: {metadata_value}")
    is_update_task_enabled = metadata_value.lower() != 'false'

    # The flag is_uworker is true for Batch environment
    # The update task should run if it's not a Batch environment
    # and the flag is enabled on the VM template metadata
    return not running_on_batch and is_update_task_enabled

  except requests.exceptions.HTTPError as http_error:
    logs.warning(f"Http error fetching metadata: {http_error}")

  except Exception as ex:
    logs.error(f"Unknown exception fetching metadata: {ex}")

  return not running_on_batch


def main():