def determine_group_level_monitored_resource_and_labels(tag,
                                                            local_resource_id)
      resource = @resource.dup
      resource.labels = @resource.labels.dup
      common_labels = @common_labels.dup

      # Change the resource type and set matched_regexp_group if the tag matches
      # certain regexp.
      matched_regexp_group = nil # @tag_regexp_list can be an empty list.
      @tag_regexp_list.each do |derived_type, tag_regexp|
        matched_regexp_group = tag_regexp.match(tag)
        if matched_regexp_group
          resource.type = derived_type
          break
        end
      end

      # Determine the monitored resource based on the local_resource_id.
      # Different monitored resource types have unique ids in different format.
      # We will query Metadata Agent for the monitored resource. Return the
      # legacy monitored resource (either the instance resource or the resource
      # inferred from the tag) if failed to get a monitored resource from
      # Metadata Agent with this key.
      #
      # Examples:
      # "container.<container_id>" // Docker container.
      # "k8s_pod.<namespace_name>.<pod_name>" // GKE pod.
      if local_resource_id
        converted_resource = monitored_resource_from_local_resource_id(
          local_resource_id)
        resource = converted_resource if converted_resource
      end

      # Once the resource type is settled down, determine the labels.
      case resource.type
      # Cloud Functions.
      when CLOUDFUNCTIONS_CONSTANTS[:resource_type]
        resource.labels.merge!(
          'region' => @gcf_region,
          'function_name' => decode_cloudfunctions_function_name(
            matched_regexp_group['encoded_function_name'])
        )
        instance_id = resource.labels.delete('instance_id')
        common_labels.merge!(
          "#{GKE_CONSTANTS[:service]}/instance_id" => instance_id,
          "#{COMPUTE_CONSTANTS[:service]}/resource_id" => instance_id,
          "#{GKE_CONSTANTS[:service]}/cluster_name" =>
            resource.labels.delete('cluster_name'),
          "#{COMPUTE_CONSTANTS[:service]}/zone" =>
            resource.labels.delete('zone')
        )

      # GKE container.
      when GKE_CONSTANTS[:resource_type]
        if matched_regexp_group
          # We only expect one occurrence of each key in the match group.
          resource_labels_candidates =
            matched_regexp_group.names.zip(matched_regexp_group.captures).to_h
          common_labels_candidates = resource_labels_candidates.dup
          resource.labels.merge!(
            delete_and_extract_labels(
              resource_labels_candidates,
              # The kubernetes_tag_regexp is poorly named. 'namespace_name' is
              # in fact 'namespace_id'. 'pod_name' is in fact 'pod_id'.
              # TODO(qingling128): Figure out how to put this map into
              # constants like GKE_CONSTANTS[:extra_resource_labels].
              'container_name' => 'container_name',
              'namespace_name' => 'namespace_id',
              'pod_name' => 'pod_id'))

          common_labels.merge!(
            delete_and_extract_labels(
              common_labels_candidates,
              GKE_CONSTANTS[:extra_common_labels]
                .map { |l| [l, "#{GKE_CONSTANTS[:service]}/#{l}"] }.to_h))
        end

      # Docker container.
      # TODO(qingling128): Remove this logic once the resource is retrieved at a
      # proper time (b/65175256).
      when DOCKER_CONSTANTS[:resource_type]
        common_labels.delete("#{COMPUTE_CONSTANTS[:service]}/resource_name")

      # TODO(qingling128): Temporary fallback for metadata agent restarts.
      # K8s resources.
      when K8S_CONTAINER_CONSTANTS[:resource_type],
           K8S_POD_CONSTANTS[:resource_type],
           K8S_NODE_CONSTANTS[:resource_type]
        common_labels.delete("#{COMPUTE_CONSTANTS[:service]}/resource_name")

      end

      # Cloud Dataflow and Cloud ML.
      # These labels can be set via the 'labels' option.
      # Report them as monitored resource labels instead of common labels.
      # e.g. "dataflow.googleapis.com/job_id" => "job_id"
      [DATAFLOW_CONSTANTS, ML_CONSTANTS].each do |service_constants|
        next unless resource.type == service_constants[:resource_type]
        resource.labels.merge!(
          delete_and_extract_labels(
            common_labels, service_constants[:extra_resource_labels]
              .map { |l| ["#{service_constants[:service]}/#{l}", l] }.to_h))
      end

      resource.freeze
      resource.labels.freeze
      common_labels.freeze

      [resource, common_labels]
    end
def determine_group_level_monitored_resource_and_labels(tag,
                                                            local_resource_id)
      resource = @resource.dup
      resource.labels = @resource.labels.dup
      common_labels = @common_labels.dup

      # Change the resource type and set matched_regexp_group if the tag matches
      # certain regexp.
      matched_regexp_group = nil # @tag_regexp_list can be an empty list.
      @tag_regexp_list.each do |derived_type, tag_regexp|
        matched_regexp_group = tag_regexp.match(tag)
        if matched_regexp_group
          resource.type = derived_type
          break
        end
      end

      # Determine the monitored resource based on the local_resource_id.
      # Different monitored resource types have unique ids in different format.
      # We will query Metadata Agent for the monitored resource. Return the
      # legacy monitored resource (either the instance resource or the resource
      # inferred from the tag) if failed to get a monitored resource from
      # Metadata Agent with this key.
      #
      # Examples:
      # "container.<container_id>" // Docker container.
      # "k8s_pod.<namespace_name>.<pod_name>" // GKE pod.
      if local_resource_id
        converted_resource = monitored_resource_from_local_resource_id(
          local_resource_id)
        resource = converted_resource if converted_resource
      end

      # Once the resource type is settled down, determine the labels.
      case resource.type
      # Cloud Functions.
      when CLOUDFUNCTIONS_CONSTANTS[:resource_type]
        resource.labels.merge!(
          'region' => @gcf_region,
          'function_name' => decode_cloudfunctions_function_name(
            matched_regexp_group['encoded_function_name'])
        )
        instance_id = resource.labels.delete('instance_id')
        common_labels.merge!(
          "#{GKE_CONSTANTS[:service]}/instance_id" => instance_id,
          "#{COMPUTE_CONSTANTS[:service]}/resource_id" => instance_id,
          "#{GKE_CONSTANTS[:service]}/cluster_name" =>
            resource.labels.delete('cluster_name'),
          "#{COMPUTE_CONSTANTS[:service]}/zone" =>
            resource.labels.delete('zone')
        )

      # GKE container.
      when GKE_CONSTANTS[:resource_type]
        if matched_regexp_group
          # We only expect one occurrence of each key in the match group.
          resource_labels_candidates =
            matched_regexp_group.names.zip(matched_regexp_group.captures).to_h
          common_labels_candidates = resource_labels_candidates.dup
          resource.labels.merge!(
            delete_and_extract_labels(
              resource_labels_candidates,
              # The kubernetes_tag_regexp is poorly named. 'namespace_name' is
              # in fact 'namespace_id'. 'pod_name' is in fact 'pod_id'.
              # TODO(qingling128): Figure out how to put this map into
              # constants like GKE_CONSTANTS[:extra_resource_labels].
              'container_name' => 'container_name',
              'namespace_name' => 'namespace_id',
              'pod_name' => 'pod_id'))

          common_labels.merge!(
            delete_and_extract_labels(
              common_labels_candidates,
              GKE_CONSTANTS[:extra_common_labels]
                .map { |l| [l, "#{GKE_CONSTANTS[:service]}/#{l}"] }.to_h))
        end

      # Docker container.
      # TODO(qingling128): Remove this logic once the resource is retrieved at a
      # proper time (b/65175256).
      when DOCKER_CONSTANTS[:resource_type]
        common_labels.delete("#{COMPUTE_CONSTANTS[:service]}/resource_name")

      # TODO(qingling128): Temporary fallback for metadata agent restarts.
      # K8s resources.
      when K8S_CONTAINER_CONSTANTS[:resource_type],
           K8S_POD_CONSTANTS[:resource_type],
           K8S_NODE_CONSTANTS[:resource_type]
        common_labels.delete("#{COMPUTE_CONSTANTS[:service]}/resource_name")

      end

      # Cloud Dataflow and Cloud ML.
      # These labels can be set via the 'labels' option.
      # Report them as monitored resource labels instead of common labels.
      # e.g. "dataflow.googleapis.com/job_id" => "job_id"
      [DATAFLOW_CONSTANTS, ML_CONSTANTS].each do |service_constants|
        next unless resource.type == service_constants[:resource_type]
        resource.labels.merge!(
          delete_and_extract_labels(
            common_labels, service_constants[:extra_resource_labels]
              .map { |l| ["#{service_constants[:service]}/#{l}", l] }.to_h))
      end

      resource.freeze
      resource.labels.freeze
      common_labels.freeze

      [resource, common_labels]
    end
Determine the group level monitored resource and common labels shared by a
 collection of entries.