def annotate_container(project_id, container, parent_pod):
    """Annotate the given container with Heapster GCM metric information.

  Args:
    project_id: the project ID
    container: the container object to annotate.
    parent_pod: the parent pod of 'container'.

  Raises:
    AssertionError: if the input arguments are invalid or if
    'parent_pod' is not the parent of 'container'
  """
    assert utilities.valid_string(project_id)
    assert utilities.is_wrapped_object(container, "Container")
    assert utilities.is_wrapped_object(parent_pod, "Pod")
    parent_name = utilities.get_attribute(container, ["properties", "Config", "Hostname"])
    assert utilities.valid_string(parent_name)
    pod_name = utilities.get_attribute(parent_pod, ["properties", "metadata", "name"])
    assert utilities.valid_string(pod_name)

    # The 'parent_name' value is truncated to the first 64 characters.
    # Thus it must be the prefix of the full pod name.
    assert pod_name.startswith(parent_name)

    m = _make_gcm_metrics(project_id, _get_container_labels(container, parent_pod))
    if m is None:
        return
    if container.get("annotations") is None:
        container["annotations"] = {}
    container["annotations"]["metrics"] = m

def _container_in_pod(gs, container, pod):
  """Returns True when 'container' is a part of 'pod'.

  Args:
    gs: global state.
    container: a wrapped container object.
    pod: a wrapped pod object.

  Raises:
    CollectorError: if the 'container' or the 'pod' are missing essential
    attributes.

  Returns:
  True iff container 'container' is a part of 'pod'.
  """
  assert isinstance(gs, global_state.GlobalState)
  assert utilities.is_wrapped_object(container, 'Container')
  assert utilities.is_wrapped_object(pod, 'Pod')

  parent_pod_id = utilities.get_parent_pod_id(container)
  if not utilities.valid_string(parent_pod_id):
    msg = 'could not find parent pod ID in container %s' % container['id']
    gs.logger_error(msg)
    raise collector_error.CollectorError(msg)

  return parent_pod_id == pod['id']

def _get_container_labels(container, parent_pod):
    """Returns key/value pairs identifying all metrics of this container.

  Args:
    container: the container object to annotate.
    parent_pod: the parent pod of 'container'.

  Returns:
  A dictionary of key/value pairs.
  If any error was detected, returns None.
  """
    if not utilities.is_wrapped_object(container, "Container"):
        return None
    if not utilities.is_wrapped_object(parent_pod, "Pod"):
        return None

    pod_id = utilities.get_attribute(parent_pod, ["properties", "metadata", "uid"])
    if not utilities.valid_string(pod_id):
        return None

    hostname = utilities.get_attribute(parent_pod, ["properties", "spec", "host"])
    if not utilities.valid_string(hostname):
        return None

    short_container_name = utilities.get_short_container_name(container, parent_pod)

    if not utilities.valid_string(short_container_name):
        return None

    return {"pod_id": pod_id, "hostname": hostname, "container_name": short_container_name}

def _do_compute_other_nodes(gs, cluster_guid, nodes_list, oldest_timestamp, g):
  """Adds nodes not in the node list but running pods to the graph.

  This handles the case when there are pods running on the master node,
  in which case we add a dummy node representing the master to the graph.
  The nodes list does not include the master.

  Args:
    gs: the global state.
    cluster_guid: the cluster's ID.
    nodes_list: a list of wrapped Node objects.
    oldest_timestamp: the timestamp of the oldest Node object.
    g: the context graph under construction.
  """
  assert isinstance(gs, global_state.GlobalState)
  assert utilities.valid_string(cluster_guid)
  assert isinstance(nodes_list, list)
  assert utilities.valid_string(oldest_timestamp)
  assert isinstance(g, ContextGraph)

  # Compute the set of known Node names.
  known_node_ids = set()
  for node in nodes_list:
    assert utilities.is_wrapped_object(node, 'Node')
    known_node_ids.add(node['id'])

  # Compute the set of Nodes referenced by pods but not in the known set.
  # The set of unknown node names may be empty.
  missing_node_ids = set()
  for pod in kubernetes.get_pods(gs):
    assert utilities.is_wrapped_object(pod, 'Pod')
    # pod.properties.spec.nodeName may be missing if the pod is waiting.
    parent_node_id = utilities.get_attribute(
        pod, ['properties', 'spec', 'nodeName'])
    if not utilities.valid_string(parent_node_id):
      continue

    if parent_node_id in known_node_ids:
      continue

    # Found a pod that does not belong to any of the known nodes.
    missing_node_ids.add(parent_node_id)

  # Process the missing nodes.
  for node_id in missing_node_ids:
    # Create a dummy node object just as a placeholder for metric
    # annotations.
    node = utilities.wrap_object({}, 'Node', node_id, time.time())

    metrics.annotate_node(node)
    node_guid = 'Node:' + node_id
    g.add_resource(node_guid, node['annotations'], 'Node', oldest_timestamp, {})
    g.add_relation(cluster_guid, node_guid, 'contains')  # Cluster contains Node

  def test_is_wrapped_object(self):
    """Tests is_wrapped_object()."""
    self.assertTrue(utilities.is_wrapped_object(CONTAINER, 'Container'))
    self.assertTrue(utilities.is_wrapped_object(CONTAINER))
    self.assertFalse(utilities.is_wrapped_object(CONTAINER, 'Pod'))

    self.assertTrue(utilities.is_wrapped_object(PARENT_POD, 'Pod'))
    self.assertTrue(utilities.is_wrapped_object(PARENT_POD))
    self.assertFalse(utilities.is_wrapped_object(PARENT_POD, 'Container'))

    self.assertFalse(utilities.is_wrapped_object({}))
    self.assertFalse(utilities.is_wrapped_object({}, 'Pod'))
    self.assertFalse(utilities.is_wrapped_object(None))
    self.assertFalse(utilities.is_wrapped_object('hello, world'))

def _do_compute_container(parent_guid, container, g):
  assert utilities.valid_string(parent_guid)
  assert utilities.is_wrapped_object(container, 'Container')
  assert isinstance(g, ContextGraph)

  container_id = container['id']
  container_guid = 'Container:' + container_id
  # TODO(vasbala): container_id is too verbose?
  g.add_resource(container_guid, container['annotations'],
                 'Container', container['timestamp'],
                 container['properties'])

  # The parent Pod contains Container.
  g.add_relation(parent_guid, container_guid, 'contains')

  image = kubernetes.get_image_from_container(container)
  image_guid = 'Image:' + image['id']

  # Add the image to the graph only if we have not added it before.
  #
  # Different containers might reference the same image using different
  # names. Unfortunately, only the first name encountered is recorded.
  # TODO(rimey): Record the other names as well, and choose the primary
  # name deterministically.
  g.add_resource(image_guid, image['annotations'], 'Image',
                 image['timestamp'], image['properties'])

  # Container createdFrom Image
  g.add_relation(container_guid, image_guid, 'createdFrom')

def _do_compute_service(gs, cluster_guid, service, g):
  assert isinstance(gs, global_state.GlobalState)
  assert utilities.valid_string(cluster_guid)
  assert utilities.is_wrapped_object(service, 'Service')
  assert isinstance(g, ContextGraph)

  service_id = service['id']
  service_guid = 'Service:' + service_id
  g.add_resource(service_guid, service['annotations'], 'Service',
                 service['timestamp'], service['properties'])

  # Cluster contains Service.
  g.add_relation(cluster_guid, service_guid, 'contains')

  # Pods load balanced by this service (use the service['spec', 'selector']
  # key/value pairs to find matching Pods)
  selector = utilities.get_attribute(
      service, ['properties', 'spec', 'selector'])
  if selector:
    if not isinstance(selector, types.DictType):
      msg = 'Service id=%s has an invalid "selector" value' % service_id
      gs.logger_error(msg)
      raise collector_error.CollectorError(msg)

    for pod in kubernetes.get_selected_pods(gs, selector):
      pod_guid = 'Pod:' + pod['id']
      # Service loadBalances Pod
      g.add_relation(service_guid, pod_guid, 'loadBalances')
  else:
    gs.logger_error('Service id=%s has no "selector" attribute', service_id)

def _do_compute_pod(gs, cluster_guid, node_guid, pod, g):
  assert isinstance(gs, global_state.GlobalState)
  assert utilities.valid_string(cluster_guid)
  assert utilities.valid_string(node_guid)
  assert utilities.is_wrapped_object(pod, 'Pod')
  assert isinstance(g, ContextGraph)

  pod_id = pod['id']
  pod_guid = 'Pod:' + pod_id
  g.add_resource(pod_guid, pod['annotations'], 'Pod', pod['timestamp'],
                 pod['properties'])

  # pod.properties.spec.nodeName may be missing if the pod is waiting
  # (not running yet).
  docker_host = utilities.get_attribute(
      pod, ['properties', 'spec', 'nodeName'])
  if utilities.valid_string(docker_host):
    # Pod is running.
    if node_guid == ('Node:' + docker_host):
      g.add_relation(node_guid, pod_guid, 'runs')  # Node runs Pod
    else:
      msg = ('Docker host (pod.properties.spec.nodeName)=%s '
             'not matching node ID=%s' % (docker_host, node_guid))
      gs.logger_error(msg)
      raise collector_error.CollectorError(msg)
  else:
    # Pod is not running.
    g.add_relation(cluster_guid, pod_guid, 'contains')  # Cluster contains Pod

def _do_compute_node(gs, input_queue, cluster_guid, node, g):
  assert isinstance(gs, global_state.GlobalState)
  assert isinstance(input_queue, Queue.PriorityQueue)
  assert utilities.valid_string(cluster_guid)
  assert utilities.is_wrapped_object(node, 'Node')
  assert isinstance(g, ContextGraph)

  node_id = node['id']
  node_guid = 'Node:' + node_id
  g.add_resource(node_guid, node['annotations'], 'Node', node['timestamp'],
                 node['properties'])
  g.add_relation(cluster_guid, node_guid, 'contains')  # Cluster contains Node
  # Pods in a Node
  # Do not compute the pods by worker threads in test mode because the order
  # of the output will be different than the golden files due to the effects
  # of queuing the work.
  for pod in kubernetes.get_pods(gs, node_id):
    if gs.get_testing():
      _do_compute_pod(gs, input_queue, node_guid, pod, g)
    else:
      input_queue.put((
          gs.get_random_priority(),
          _do_compute_pod,
          {'gs': gs, 'input_queue': input_queue, 'node_guid': node_guid,
           'pod': pod, 'g': g}))

def _do_compute_rcontroller(gs, cluster_guid, rcontroller, g):
  assert isinstance(gs, global_state.GlobalState)
  assert utilities.valid_string(cluster_guid)
  assert utilities.is_wrapped_object(rcontroller, 'ReplicationController')
  assert isinstance(g, ContextGraph)

  rcontroller_id = rcontroller['id']
  rcontroller_guid = 'ReplicationController:' + rcontroller_id
  g.add_resource(rcontroller_guid, rcontroller['annotations'],
                 'ReplicationController',
                 rcontroller['timestamp'], rcontroller['properties'])

  # Cluster contains Rcontroller
  g.add_relation(cluster_guid, rcontroller_guid, 'contains')

  # Pods that are monitored by this replication controller.
  # Use the rcontroller['spec']['selector'] key/value pairs to find matching
  # pods.
  selector = utilities.get_attribute(
      rcontroller, ['properties', 'spec', 'selector'])
  if selector:
    if not isinstance(selector, types.DictType):
      msg = ('Rcontroller id=%s has an invalid "replicaSelector" value' %
             rcontroller_id)
      gs.logger_error(msg)
      raise collector_error.CollectorError(msg)

    for pod in kubernetes.get_selected_pods(gs, selector):
      pod_guid = 'Pod:' + pod['id']
      # Rcontroller monitors Pod
      g.add_relation(rcontroller_guid, pod_guid, 'monitors')
  else:
    gs.logger_error('Rcontroller id=%s has no "spec.selector" attribute',
                    rcontroller_id)

def get_nodes_with_metrics(gs):
  """Gets the list of all nodes in the current cluster with their metrics.

  Args:
    gs: global state.

  Returns:
    list of wrapped node objects.
    Each element in the list is the result of
    utilities.wrap_object(node, 'Node', ...)

  Raises:
    CollectorError in case of failure to fetch data from Kubernetes.
    Other exceptions may be raised due to exectution errors.
  """
  nodes_list = get_nodes(gs)

  for node in nodes_list:
    assert utilities.is_wrapped_object(node, 'Node')
    project_id = utilities.node_id_to_project_id(node['id'])
    # The project_id may be '_unknown_'. This is not a big
    # deal, since the aggregator knows the project ID.
    metrics.annotate_node(project_id, node)

  return nodes_list

def get_containers_from_pod(pod):
    """Extracts synthesized container resources from a pod.

  Only containers for which status is available are included. (The pod may
  still be pending.)
  """
    assert utilities.is_wrapped_object(pod, "Pod")
    specs = utilities.get_attribute(pod, ["properties", "spec", "containers"])
    statuses = utilities.get_attribute(pod, ["properties", "status", "containerStatuses"])
    timestamp = pod["timestamp"]

    spec_dict = {}
    for spec in specs or []:
        spec = spec.copy()
        spec_dict[spec.pop("name")] = spec

    containers = []
    for status in statuses or []:
        status = status.copy()
        name = status.pop("name")
        unique_id = status.get("containerID", name)
        obj = {"metadata": {"name": name}, "spec": spec_dict.get(name, {}), "status": status}
        container = utilities.wrap_object(obj, "Container", unique_id, timestamp, label=name)
        containers.append(container)

    return containers

def get_image_from_container(container):
    """Extracts a synthesized image resource from a container."""
    assert utilities.is_wrapped_object(container, "Container")
    timestamp = container["timestamp"]
    image_name = container["properties"]["status"]["image"]
    image_id = container["properties"]["status"]["imageID"]
    obj = {"metadata": {"name": image_name}}
    return utilities.wrap_object(obj, "Image", image_id, timestamp, label=image_name)

def _do_compute_node(gs, input_queue, cluster_guid, node, g):
  assert isinstance(gs, global_state.GlobalState)
  assert isinstance(input_queue, Queue.PriorityQueue)
  assert utilities.valid_string(cluster_guid)
  assert utilities.is_wrapped_object(node, 'Node')
  assert isinstance(g, ContextGraph)

  node_id = node['id']
  node_guid = 'Node:' + node_id
  g.add_resource(node_guid, node['annotations'], 'Node', node['timestamp'],
                 node['properties'])
  g.add_relation(cluster_guid, node_guid, 'contains')  # Cluster contains Node
  # Pods in a Node
  pod_ids = set()
  docker_hosts = set()

  # Process pods sequentially because calls to _do_compute_pod() do not call
  # lower-level services or wait.
  for pod in kubernetes.get_pods(gs, node_id):
    _do_compute_pod(gs, cluster_guid, node_guid, pod, g)
    pod_ids.add(pod['id'])
    # pod.properties.spec.nodeName may be missing if the pod is waiting.
    docker_host = utilities.get_attribute(
        pod, ['properties', 'spec', 'nodeName'])
    if utilities.valid_string(docker_host):
      docker_hosts.add(docker_host)

  # 'docker_hosts' should contain a single Docker host, because all of
  # the pods run in the same Node. However, if it is not the case, we
  # cannot fix the situation, so we just log an error message and continue.
  if len(docker_hosts) != 1:
    gs.logger_error(
        'corrupt pod data in node=%s: '
        '"docker_hosts" is empty or contains more than one entry: %s',
        node_guid, str(docker_hosts))

  # Process containers concurrently.
  for docker_host in docker_hosts:
    for container in docker.get_containers_with_metrics(gs, docker_host):
      parent_pod_id = utilities.get_parent_pod_id(container)
      if utilities.valid_string(parent_pod_id) and (parent_pod_id in pod_ids):
        # This container is contained in a pod.
        parent_guid = 'Pod:' + parent_pod_id
      else:
        # This container is not contained in a pod.
        parent_guid = node_guid

      # Do not compute the containers by worker threads in test mode
      # because the order of the output will be different than the golden
      # files due to the effects of queuing the work.
      if gs.get_testing():
        _do_compute_container(gs, docker_host, parent_guid, container, g)
      else:
        input_queue.put((
            gs.get_random_priority(),
            _do_compute_container,
            {'gs': gs, 'docker_host': docker_host, 'parent_guid': parent_guid,
             'container': container, 'g': g}))

def annotate_container(container, parent_pod):
  """Annotate the given container with Heapster GCM metric information.

  Args:
    container: the container object to annotate.
    parent_pod: the parent pod of 'container'.

  Raises:
    AssertionError: if the input arguments are invalid
  """
  assert utilities.is_wrapped_object(container, 'Container')
  assert utilities.is_wrapped_object(parent_pod, 'Pod')

  m = _make_gcm_metrics(_get_container_labels(container, parent_pod))
  if m is not None:
    if 'annotations' not in container:
      container['annotations'] = {}
    container['annotations']['metrics'] = m

def _do_compute_node(cluster_guid, node, g):
  assert utilities.valid_string(cluster_guid)
  assert utilities.is_wrapped_object(node, 'Node')
  assert isinstance(g, ContextGraph)

  node_id = node['id']
  node_guid = 'Node:' + node_id
  g.add_resource(node_guid, node['annotations'], 'Node', node['timestamp'],
                 node['properties'])
  g.add_relation(cluster_guid, node_guid, 'contains')  # Cluster contains Node

  def count_resources(self, output, type_name):
    assert isinstance(output, types.DictType)
    assert isinstance(type_name, types.StringTypes)
    if not isinstance(output.get('resources'), types.ListType):
      return 0

    n = 0
    for r in output.get('resources'):
      assert utilities.is_wrapped_object(r)
      if r.get('type') == type_name:
        n += 1

    return n

def _get_container_labels(container, parent_pod):
  """Returns key/value pairs identifying all metrics of this container.

  Args:
    container: the container object to annotate.
    parent_pod: the parent pod of 'container'.

  Returns:
  A dictionary of key/value pairs.
  If any error was detected, returns None.
  """
  if not utilities.is_wrapped_object(container, 'Container'):
    return None
  if not utilities.is_wrapped_object(parent_pod, 'Pod'):
    return None

  pod_id = utilities.get_attribute(
      parent_pod, ['properties', 'metadata', 'uid'])
  if not utilities.valid_string(pod_id):
    return None

  hostname = utilities.get_attribute(
      parent_pod, ['properties', 'spec', 'host'])
  if not utilities.valid_string(hostname):
    return None

  short_container_name = utilities.get_short_container_name(
      container, parent_pod)

  if not utilities.valid_string(short_container_name):
    return None

  return {
      'pod_id': pod_id,
      'hostname': hostname,
      'container_name': short_container_name
  }

def annotate_node(node):
  """Annotate the given node with Heapster GCM metric information.

  Args:
    node: the node object to annotate.

  Raises:
    AssertionError: if the input argument is invalid.
  """
  assert utilities.is_wrapped_object(node, 'Node')

  m = _make_gcm_metrics(_get_node_labels(node))
  if m is not None:
    if 'annotations' not in node:
      node['annotations'] = {}
    node['annotations']['metrics'] = m

def annotate_container_error(container, message):
    """Annotate the given container with an error message in lieu of metrics.

  Args:
    container: the container object to annotate.
    message: a message explaining why this container was not annotated with
      metrics.

  Raises:
    AssertionError: if the input arguments are invalid.
  """
    assert utilities.is_wrapped_object(container, "Container")
    assert utilities.valid_string(message)
    if container.get("annotations") is None:
        container["annotations"] = {}
    container["annotations"]["metrics"] = {"gcmError": message}