code: def emit_metric(
      metric_name: str,
      metrics: Dict[str, Any],
  ) -> None:
      """
      Upload a metric to DynamoDB (and from there, Rockset).
  
      Even if EMIT_METRICS is set to False, this function will still run the code to
      validate and shape the metrics, skipping just the upload.
  
      Parameters:
          metric_name:
              Name of the metric. Every unique metric should have a different name
              and be emitted just once per run attempt.
              Metrics are namespaced by their module and the function that emitted them.
          metrics: The actual data to record.
  
      Some default values are populated from environment variables, which must be set
      for metrics to be emitted. (If they're not set, this function becomes a noop):
      """
  
      if metrics is None:
          raise ValueError("You didn't ask to upload any metrics!")
  
      # Merge the given metrics with the global metrics, overwriting any duplicates
      # with the given metrics.
      metrics = {**global_metrics, **metrics}
  
      # We use these env vars that to determine basic info about the workflow run.
      # By using env vars, we don't have to pass this info around to every function.
      # It also helps ensure that we only emit metrics during CI
      env_var_metrics = [
          EnvVarMetric("repo", "GITHUB_REPOSITORY"),
          EnvVarMetric("workflow", "GITHUB_WORKFLOW"),
          EnvVarMetric("build_environment", "BUILD_ENVIRONMENT"),
          EnvVarMetric("job", "GITHUB_JOB"),
          EnvVarMetric("test_config", "TEST_CONFIG", required=False),
          EnvVarMetric(
              "pr_number",
              "PR_NUMBER",
              required=False,
              type_conversion_fn=int,
          ),
          EnvVarMetric(
              "run_id", "GITHUB_RUN_ID", type_conversion_fn=int
          ),
          EnvVarMetric(
              "run_number", "GITHUB_RUN_NUMBER", type_conversion_fn=int
          ),
          EnvVarMetric(
              "run_attempt",
              "GITHUB_RUN_ATTEMPT",
              type_conversion_fn=int,
          ),
          EnvVarMetric("job_id", "JOB_ID", type_conversion_fn=int),
          EnvVarMetric("job_name", "JOB_NAME"),
      ]
  
      # Use info about the function that invoked this one as a namespace and a way to filter metrics.
      calling_frame = inspect.currentframe().f_back  # type: ignore[union-attr]
      calling_frame_info = inspect.getframeinfo(calling_frame)  # type: ignore[arg-type]
      calling_file = os.path.basename(calling_frame_info.filename)
      calling_module = inspect.getmodule(calling_frame).__name__  # type: ignore[union-attr]
      calling_function = calling_frame_info.function
  
      try:
          reserved_metrics = {
              "metric_name": metric_name,
              "calling_file": calling_file,
              "calling_module": calling_module,
              "calling_function": calling_function,
              "timestamp": datetime.datetime.utcnow().strftime(
                  "%Y-%m-%d %H:%M:%S.%f"
              ),
              **{
                  m.name: m.value()
                  for m in env_var_metrics
                  if m.value()
              },
          }
      except ValueError as e:
          warn(f"Not emitting metrics for {metric_name}. {e}")
          return
  
      # Prefix key with metric name and timestamp to derisk chance of a uuid1 name collision
      reserved_metrics[
          "dynamo_key"
      ] = f"{metric_name}_{int(time.time())}_{uuid.uuid1().hex}"
  
      # Ensure the metrics dict doesn't contain any reserved keys
      for key in reserved_metrics.keys():
          used_reserved_keys = [k for k in metrics.keys() if k == key]
          if used_reserved_keys:
              raise ValueError(
                  f"Metrics dict contains reserved keys: [{', '.join(key)}]"
              )
  
      # boto3 doesn't support uploading float values to DynamoDB, so convert them all to decimals.
      metrics = _convert_float_values_to_decimals(metrics)
  
      if EMIT_METRICS:
          try:
              session = boto3.Session(region_name="us-east-1")
              session.resource("dynamodb").Table(
                  "torchci-metrics"
              ).put_item(
                  Item={
                      **reserved_metrics,
                      **metrics,
                  }
              )
          except Exception as e:
              # We don't want to fail the job if we can't upload the metric.
              # We still raise the ValueErrors outside this try block since those indicate improperly configured metrics
              warn(
                  f"Error uploading metric {metric_name} to DynamoDB: {e}"
              )
              return
      else:
          print(
              f"Not emitting metrics for {metric_name}. Boto wasn't imported."
          )
  
file_path: pytorch/tools/stats/upload_metrics.py
imports: import datetime
  import inspect
  import os
  import time
  import uuid
  from decimal import Decimal
  from typing import Any, Dict
  from warnings import warn
label_desc: Use image node_id 2 for calling a global function with example usage: emit_metric(metric_name, metrics) without return types