[BEAM-12490] Python PreCommit flaking in SdkWorkerTest.test_harness_monitoring_infos_and_metadata

• Type: Bug
• Status: Resolved
• Priority: P1
• Resolution: Fixed
• Fix Version/s: Not applicable
• Component/s: test-failures
• Labels: flake
• Assignee: Mike Hernandez
• Reporter: Brian Hulette
• Created: 14/Jun/21 16:01
• Updated: 18/Jun/21 21:22
• Resolved: 18/Jun/21 16:21

Description

SdkWorkerTest.test_harness_monitoring_infos_and_metadata has been very flaky over the last few days.

Based on test history it looks like the first flake was in cron build #4303.

self = <apache_beam.runners.worker.sdk_worker_test.SdkWorkerTest testMethod=test_harness_monitoring_infos_and_metadata>

    def test_harness_monitoring_infos_and_metadata(self):
      # Clear the process wide metric container.
      MetricsEnvironment.process_wide_container().reset()
      # Create a process_wide metric.
      urn = 'my.custom.urn'
      labels = {'key': 'value'}
      InternalMetrics.counter(urn=urn, labels=labels, process_wide=True).inc(10)
    
      harness_monitoring_infos_request = beam_fn_api_pb2.InstructionRequest(
          instruction_id="monitoring_infos",
          harness_monitoring_infos=beam_fn_api_pb2.HarnessMonitoringInfosRequest(
          ))
    
>     responses = self.get_responses([harness_monitoring_infos_request])

apache_beam/runners/worker/sdk_worker_test.py:236: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
apache_beam/runners/worker/sdk_worker_test.py:220: in get_responses
    harness.run()
apache_beam/runners/worker/sdk_worker.py:257: in run
    work_request)
apache_beam/runners/worker/sdk_worker.py:330: in _request_harness_monitoring_infos
    ).to_runner_api_monitoring_infos(None).values()
apache_beam/metrics/execution.py:312: in to_runner_api_monitoring_infos
    cell in items
apache_beam/metrics/execution.py:312: in <listcomp>
    cell in items
apache_beam/metrics/cells.py:76: in to_runner_api_monitoring_info
    mi = self.to_runner_api_monitoring_info_impl(name, transform_id)
apache_beam/metrics/cells.py:158: in to_runner_api_monitoring_info_impl
    name.urn, self.get_cumulative(), labels=name.labels)
apache_beam/metrics/monitoring_infos.py:187: in int64_counter
    return create_monitoring_info(urn, SUM_INT64_TYPE, metric, labels)
apache_beam/metrics/monitoring_infos.py:286: in create_monitoring_info
    urn=urn, type=type_urn, labels=labels or dict(), payload=payload)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

args = [{'GCS_BUCKET': 'gcsio-test', 'GCS_PROJECT_ID': 1, 'METHOD': 'Objects.insert', 'RESOURCE': '//storage.googleapis.com/buckets/gcsio-test', ...}]
kwds = {}
self = {'GCS_BUCKET': 'gcsio-test', 'METHOD': 'Objects.insert', 'RESOURCE': '//storage.googleapis.com/buckets/gcsio-test', 'GCS_PROJECT_ID': '', 'SERVICE': 'Storage'}
other = {'GCS_BUCKET': 'gcsio-test', 'GCS_PROJECT_ID': 1, 'METHOD': 'Objects.insert', 'RESOURCE': '//storage.googleapis.com/buckets/gcsio-test', ...}
key = 'GCS_PROJECT_ID'

    def update(*args, **kwds):
        ''' D.update([E, ]**F) -> None.  Update D from mapping/iterable E and F.
            If E present and has a .keys() method, does:     for k in E: D[k] = E[k]
            If E present and lacks .keys() method, does:     for (k, v) in E: D[k] = v
            In either case, this is followed by: for k, v in F.items(): D[k] = v
        '''
        if not args:
            raise TypeError("descriptor 'update' of 'MutableMapping' object "
                            "needs an argument")
        self, *args = args
        if len(args) > 1:
            raise TypeError('update expected at most 1 arguments, got %d' %
                            len(args))
        if args:
            other = args[0]
            if isinstance(other, Mapping):
                for key in other:
>                   self[key] = other[key]
E                   TypeError: 1 has type int, but expected one of: bytes, unicode

/usr/lib/python3.7/_collections_abc.py:841: TypeError

Issue Links

• links to: GitHub Pull Request #15033