Migrate to Google Cloud Dataflow Client (#37639)

* [WIP] Migrate to Google Cloud Dataflow Client

rebase

* Trigger relevant postcommits

* base image update

* fix camel case

* update dataflow runner + tests

* slide import to avoid triggering unit tests

* yapf stuff

* remove extra print

* further spec structs, fix incorrect piplineUrl option, remove old client code

* suppress line-too-longs

* formatting

* linting, tweak metrics tests

* Proto-specific changes to metric processing tests

* try to dump logging

* handle more straightforward metrics values

* add skips since the unit tests now depend on the proto library

* testing if there's a disconnect between proto behavior locally and in tests

* correct scalar access

* clean up dist accesses

* linting, various fixes

* fix unit test setup for direct accesses

* linting

* more linting

* re-enable histograms

* Bump dataflow client version, restore pausing/paused concept

* formatting

* fix enum selection

* handle the proto hash PR

* remove unnecessary try/except block

* re-delete old messsages

* fix disk_provisioned_iops/throughput_mibps tests

* code bot suggestions

* revert pipeline options interaction

* leftover update

* yapf

* swap credential loading

* Fix message importance parsing

* fix apiclient_test case w/ cloud deps and no auth

* Local runner SSL fix

* last local change

* apitools test fix

* yapf

* Apply suggestions from code review

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

* gemini suggestions

* fix message importance

* fix local runner case

* remove legacy translation maps

* revert container requirements changes

* missing import

* fix bad trigger file merge

* Update sdks/python/apache_beam/runners/dataflow/dataflow_metrics.py

Co-authored-by: Danny McCormick <dannymccormick@google.com>

* fix merge problem with runnerv2 disabled tests

* fix state breakage

---------

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>
Co-authored-by: Danny McCormick <dannymccormick@google.com>

Author: 3 people

.github/trigger_files/beam_PostCommit_Python.json

@@ -1,5 +1,5 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run.",
   "pr": "38069",
-  "modification": 42
+  "modification": 43
 }

.github/trigger_files/beam_PostCommit_Python_ValidatesRunner_Dataflow.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "modification": 2
+  "modification": 3
 }

sdks/python/apache_beam/runners/dataflow/dataflow_metrics.py

@@ -57,7 +57,6 @@ def _get_match(proto, filter_fn):
 
 
 # V1b3 MetricStructuredName keys to accept and copy to the MetricKey labels.
-STEP_LABEL = 'step'
 STRUCTURED_NAME_LABELS = set(
     ['execution_step', 'original_name', 'output_user_name'])
 
@@ -112,9 +111,7 @@ def _translate_step_name(self, internal_name):
       try:
         step = _get_match(
             self._job_graph.proto.steps, lambda x: x.name == internal_name)
-        user_step_name = _get_match(
-            step.properties.additionalProperties,
-            lambda x: x.key == 'user_name').value.string_value
+        user_step_name = step.properties.get('user_name')
       except ValueError:
         pass  # Exception is handled below.
     if not user_step_name:
@@ -135,24 +132,17 @@ def _get_metric_key(self, metric):
       #   step name (only happens for unstructured-named metrics).
       # 2. Unable to unpack [step] or [namespace]; which should only happen
       #   for unstructured names.
-      step = _get_match(
-          metric.name.context.additionalProperties,
-          lambda x: x.key == STEP_LABEL).value
+      step = metric.name.context['step']
       step = self._translate_step_name(step)
-    except ValueError:
+    except (KeyError, ValueError):
       pass
 
     namespace = "dataflow/v1b3"  # Try to extract namespace or add a default.
-    try:
-      namespace = _get_match(
-          metric.name.context.additionalProperties,
-          lambda x: x.key == 'namespace').value
-    except ValueError:
-      pass
+    namespace = metric.name.context.get('namespace', 'dataflow/v1b3')
 
-    for kv in metric.name.context.additionalProperties:
-      if kv.key in STRUCTURED_NAME_LABELS:
-        labels[kv.key] = kv.value
+    for key in metric.name.context:
+      if key in STRUCTURED_NAME_LABELS:
+        labels[key] = metric.name.context[key]
     # Package everything besides namespace and name the labels as well,
     # including unmodified step names to assist in integration the exact
     # unmodified values which come from dataflow.
@@ -185,10 +175,7 @@ def _populate_metrics(self, response, result, user_metrics=False):
         #  in the service.
         # The second way is only useful for the UI, and should be ignored.
         continue
-      is_tentative = [
-          prop for prop in metric.name.context.additionalProperties
-          if prop.key == 'tentative' and prop.value == 'true'
-      ]
+      is_tentative = metric.name.context.get('tentative') == 'true'
       tentative_or_committed = 'tentative' if is_tentative else 'committed'
 
       metric_key = self._get_metric_key(metric)
@@ -209,32 +196,16 @@ def _get_metric_value(self, metric):
       return None
 
     if metric.scalar is not None:
-      return metric.scalar.integer_value
+      # This will always be a single value if there is any data in the field.
+      val = metric.scalar
+      if isinstance(val, float) and val.is_integer():
+        return int(val)
+      return val
     elif metric.distribution is not None:
-      dist_count = _get_match(
-          metric.distribution.object_value.properties,
-          lambda x: x.key == 'count').value.integer_value
-      dist_min = _get_match(
-          metric.distribution.object_value.properties,
-          lambda x: x.key == 'min').value.integer_value
-      dist_max = _get_match(
-          metric.distribution.object_value.properties,
-          lambda x: x.key == 'max').value.integer_value
-      dist_sum = _get_match(
-          metric.distribution.object_value.properties,
-          lambda x: x.key == 'sum').value.integer_value
-      if dist_sum is None:
-        # distribution metric is not meant to use on large values, but in case
-        # it is, the value can overflow and become double_value, the correctness
-        # of the value may not be guaranteed.
-        _LOGGER.info(
-            "Distribution metric sum value seems to have "
-            "overflowed integer_value range, the correctness of sum or mean "
-            "value may not be guaranteed: %s" % metric.distribution)
-        dist_sum = int(
-            _get_match(
-                metric.distribution.object_value.properties,
-                lambda x: x.key == 'sum').value.double_value)
+      dist_count = int(metric.distribution['count'])
+      dist_min = int(metric.distribution['min'])
+      dist_max = int(metric.distribution['max'])
+      dist_sum = int(metric.distribution['sum'])
       return DistributionResult(
           DistributionData(dist_sum, dist_count, dist_min, dist_max))
       #TODO(https://github.com/apache/beam/issues/31788) support StringSet after