You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by yh...@apache.org on 2026/06/10 19:47:37 UTC
(beam) 01/01: Revert "[Python] Honor disableCounterMetrics, disableStringSetMetrics, and di…"
This is an automated email from the ASF dual-hosted git repository.
Abacn pushed a commit to branch revert-38749-py-disable-user-metrics-38746
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 56fd97f4f9a778e4e7bf4a9ec452190ab8d87352
Author: Yi Hu <hu...@gmail.com>
AuthorDate: Wed Jun 10 15:47:25 2026 -0400
Revert "[Python] Honor disableCounterMetrics, disableStringSetMetrics, and di…"
This reverts commit c01ceeab24c7cae7167dbc08b21529339659094f.
---
CHANGES.md | 1 -
sdks/python/apache_beam/metrics/metric.py | 56 +----------
sdks/python/apache_beam/metrics/metric_test.py | 104 ---------------------
sdks/python/apache_beam/pipeline.py | 2 -
.../apache_beam/runners/worker/sdk_worker_main.py | 2 -
5 files changed, 4 insertions(+), 161 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index ac8215f3549..698d88b01fa 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -69,7 +69,6 @@
## New Features / Improvements
-* X feature added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)).
* (Java) Enabled state tag encoding v2 by default for new Dataflow Streaming Engine jobs. It can be disabled by passing `--experiments=disable_streaming_engine_state_tag_encoding_v2` or `--updateCompatibilityVersion=2.74.0` pipeline option. Note that the tag encoding version cannot change during a job update. Jobs using tag encoding v2 (enabled by default for new jobs on 2.75.0+) cannot be downgraded to Beam versions prior to 2.73.0, as only versions 2.73.0 and later support tag encoding [...]
## Breaking Changes
diff --git a/sdks/python/apache_beam/metrics/metric.py b/sdks/python/apache_beam/metrics/metric.py
index a66eed640be..b15237cae02 100644
--- a/sdks/python/apache_beam/metrics/metric.py
+++ b/sdks/python/apache_beam/metrics/metric.py
@@ -46,13 +46,11 @@ from apache_beam.metrics.metricbase import Gauge
from apache_beam.metrics.metricbase import Histogram
from apache_beam.metrics.metricbase import MetricName
from apache_beam.metrics.metricbase import StringSet
-from apache_beam.options.pipeline_options import DebugOptions
if TYPE_CHECKING:
from apache_beam.internal.metrics.metric import MetricLogger
from apache_beam.metrics.execution import MetricKey
from apache_beam.metrics.metricbase import Metric
- from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.utils.histogram import BucketType
__all__ = ['Metrics', 'MetricsFilter', 'Lineage']
@@ -60,37 +58,6 @@ __all__ = ['Metrics', 'MetricsFilter', 'Lineage']
_LOGGER = logging.getLogger(__name__)
-class MetricsFlag(object):
- """Process-wide flags controlling which user metric kinds are emitted."""
- counter_disabled = False
- string_set_disabled = False
- bounded_trie_disabled = False
- _initialized = False
-
- @classmethod
- def set_default_pipeline_options(cls, options: 'PipelineOptions') -> None:
- if cls._initialized:
- return
- debug_options = options.view_as(DebugOptions)
- if debug_options.lookup_experiment('disableCounterMetrics'):
- cls.counter_disabled = True
- _LOGGER.info('Counter metrics are disabled.')
- if debug_options.lookup_experiment('disableStringSetMetrics'):
- cls.string_set_disabled = True
- _LOGGER.info('StringSet metrics are disabled.')
- if debug_options.lookup_experiment('disableBoundedTrieMetrics'):
- cls.bounded_trie_disabled = True
- _LOGGER.info('BoundedTrie metrics are disabled.')
- cls._initialized = True
-
- @classmethod
- def reset(cls) -> None:
- cls.counter_disabled = False
- cls.string_set_disabled = False
- cls.bounded_trie_disabled = False
- cls._initialized = False
-
-
class Metrics(object):
"""Lets users create/access metric objects during pipeline execution."""
@staticmethod
@@ -237,17 +204,12 @@ class Metrics(object):
def __init__(
self, metric_name: MetricName, process_wide: bool = False) -> None:
super().__init__(metric_name)
- self._updater = MetricUpdater(
+ self.inc = MetricUpdater( # type: ignore[method-assign]
cells.CounterCell,
metric_name,
default_value=1,
process_wide=process_wide)
- def inc(self, n: int = 1) -> None:
- if MetricsFlag.counter_disabled:
- return
- self._updater(n)
-
class DelegatingDistribution(Distribution):
"""Metrics Distribution Delegates functionality to MetricsEnvironment."""
def __init__(
@@ -269,23 +231,13 @@ class Metrics(object):
"""Metrics StringSet that Delegates functionality to MetricsEnvironment."""
def __init__(self, metric_name: MetricName) -> None:
super().__init__(metric_name)
- self._updater = MetricUpdater(cells.StringSetCell, metric_name)
-
- def add(self, value: str) -> None:
- if MetricsFlag.string_set_disabled:
- return
- self._updater(value)
+ self.add = MetricUpdater(cells.StringSetCell, metric_name) # type: ignore[method-assign]
class DelegatingBoundedTrie(BoundedTrie):
- """Metrics BoundedTrie that Delegates functionality to MetricsEnvironment."""
+ """Metrics StringSet that Delegates functionality to MetricsEnvironment."""
def __init__(self, metric_name: MetricName) -> None:
super().__init__(metric_name)
- self._updater = MetricUpdater(cells.BoundedTrieCell, metric_name)
-
- def add(self, value) -> None:
- if MetricsFlag.bounded_trie_disabled:
- return
- self._updater(value)
+ self.add = MetricUpdater(cells.BoundedTrieCell, metric_name) # type: ignore[method-assign]
class MetricResults(object):
diff --git a/sdks/python/apache_beam/metrics/metric_test.py b/sdks/python/apache_beam/metrics/metric_test.py
index 6937236a8aa..ae66200737b 100644
--- a/sdks/python/apache_beam/metrics/metric_test.py
+++ b/sdks/python/apache_beam/metrics/metric_test.py
@@ -32,9 +32,7 @@ from apache_beam.metrics.metric import Lineage
from apache_beam.metrics.metric import MetricResults
from apache_beam.metrics.metric import Metrics
from apache_beam.metrics.metric import MetricsFilter
-from apache_beam.metrics.metric import MetricsFlag
from apache_beam.metrics.metricbase import MetricName
-from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.runners.direct.direct_runner import BundleBasedDirectRunner
from apache_beam.runners.worker import statesampler
from apache_beam.testing.metric_result_matchers import DistributionMatcher
@@ -123,108 +121,6 @@ class MetricsTest(unittest.TestCase):
with self.assertRaises(ValueError):
Metrics.get_namespace(object())
- def test_metrics_flag(self):
- MetricsFlag.reset()
- try:
- self.assertFalse(MetricsFlag.counter_disabled)
- self.assertFalse(MetricsFlag.string_set_disabled)
- self.assertFalse(MetricsFlag.bounded_trie_disabled)
-
- options = PipelineOptions(['--experiments=disableCounterMetrics'])
- MetricsFlag.set_default_pipeline_options(options)
- self.assertTrue(MetricsFlag.counter_disabled)
- self.assertFalse(MetricsFlag.string_set_disabled)
- self.assertFalse(MetricsFlag.bounded_trie_disabled)
-
- MetricsFlag.reset()
- options = PipelineOptions(['--experiments=disableStringSetMetrics'])
- MetricsFlag.set_default_pipeline_options(options)
- self.assertFalse(MetricsFlag.counter_disabled)
- self.assertTrue(MetricsFlag.string_set_disabled)
- self.assertFalse(MetricsFlag.bounded_trie_disabled)
-
- MetricsFlag.reset()
- options = PipelineOptions(['--experiments=disableBoundedTrieMetrics'])
- MetricsFlag.set_default_pipeline_options(options)
- self.assertFalse(MetricsFlag.counter_disabled)
- self.assertFalse(MetricsFlag.string_set_disabled)
- self.assertTrue(MetricsFlag.bounded_trie_disabled)
-
- MetricsFlag.reset()
- options = PipelineOptions([
- '--experiments=disableCounterMetrics',
- '--experiments=disableStringSetMetrics',
- '--experiments=disableBoundedTrieMetrics',
- ])
- MetricsFlag.set_default_pipeline_options(options)
- self.assertTrue(MetricsFlag.counter_disabled)
- self.assertTrue(MetricsFlag.string_set_disabled)
- self.assertTrue(MetricsFlag.bounded_trie_disabled)
- finally:
- MetricsFlag.reset()
-
- def test_disabled_counter_is_noop(self):
- sampler = statesampler.StateSampler('', counters.CounterFactory())
- statesampler.set_current_tracker(sampler)
- state = sampler.scoped_state(
- 'mystep', 'myState', metrics_container=MetricsContainer('mystep'))
- MetricsFlag.reset()
- try:
- sampler.start()
- with state:
- container = MetricsEnvironment.current_container()
- Metrics.counter('ns', 'baseline').inc()
- self.assertEqual(len(container.metrics), 1)
- options = PipelineOptions(['--experiments=disableCounterMetrics'])
- MetricsFlag.set_default_pipeline_options(options)
- Metrics.counter('ns', 'after_disable').inc()
- Metrics.counter('ns', 'after_disable').inc(5)
- Metrics.counter('ns', 'after_disable').dec()
- self.assertEqual(len(container.metrics), 1)
- finally:
- sampler.stop()
- MetricsFlag.reset()
-
- def test_disabled_string_set_is_noop(self):
- sampler = statesampler.StateSampler('', counters.CounterFactory())
- statesampler.set_current_tracker(sampler)
- state = sampler.scoped_state(
- 'mystep', 'myState', metrics_container=MetricsContainer('mystep'))
- MetricsFlag.reset()
- try:
- sampler.start()
- with state:
- container = MetricsEnvironment.current_container()
- Metrics.string_set('ns', 'baseline').add('seed')
- self.assertEqual(len(container.metrics), 1)
- options = PipelineOptions(['--experiments=disableStringSetMetrics'])
- MetricsFlag.set_default_pipeline_options(options)
- Metrics.string_set('ns', 'after_disable').add('value')
- self.assertEqual(len(container.metrics), 1)
- finally:
- sampler.stop()
- MetricsFlag.reset()
-
- def test_disabled_bounded_trie_is_noop(self):
- sampler = statesampler.StateSampler('', counters.CounterFactory())
- statesampler.set_current_tracker(sampler)
- state = sampler.scoped_state(
- 'mystep', 'myState', metrics_container=MetricsContainer('mystep'))
- MetricsFlag.reset()
- try:
- sampler.start()
- with state:
- container = MetricsEnvironment.current_container()
- Metrics.bounded_trie('ns', 'baseline').add(['a'])
- self.assertEqual(len(container.metrics), 1)
- options = PipelineOptions(['--experiments=disableBoundedTrieMetrics'])
- MetricsFlag.set_default_pipeline_options(options)
- Metrics.bounded_trie('ns', 'after_disable').add(['a', 'b'])
- self.assertEqual(len(container.metrics), 1)
- finally:
- sampler.stop()
- MetricsFlag.reset()
-
def test_counter_empty_name(self):
with self.assertRaises(ValueError):
Metrics.counter("namespace", "")
diff --git a/sdks/python/apache_beam/pipeline.py b/sdks/python/apache_beam/pipeline.py
index 594660d9bea..750868f7443 100644
--- a/sdks/python/apache_beam/pipeline.py
+++ b/sdks/python/apache_beam/pipeline.py
@@ -73,7 +73,6 @@ from apache_beam import pvalue
from apache_beam.coders import typecoders
from apache_beam.internal import pickler
from apache_beam.io.filesystems import FileSystems
-from apache_beam.metrics.metric import MetricsFlag
from apache_beam.options.pipeline_options import CrossLanguageOptions
from apache_beam.options.pipeline_options import DebugOptions
from apache_beam.options.pipeline_options import PipelineOptions
@@ -193,7 +192,6 @@ class Pipeline(HasDisplayData):
self._options = PipelineOptions([])
FileSystems.set_options(self._options)
- MetricsFlag.set_default_pipeline_options(self._options)
if runner is None:
runner = self._options.view_as(StandardOptions).runner
diff --git a/sdks/python/apache_beam/runners/worker/sdk_worker_main.py b/sdks/python/apache_beam/runners/worker/sdk_worker_main.py
index 58beda96d63..754a631eaf3 100644
--- a/sdks/python/apache_beam/runners/worker/sdk_worker_main.py
+++ b/sdks/python/apache_beam/runners/worker/sdk_worker_main.py
@@ -33,7 +33,6 @@ from google.protobuf import text_format
from apache_beam.internal import pickler
from apache_beam.io import filesystems
-from apache_beam.metrics import metric
from apache_beam.options.pipeline_options import DebugOptions
from apache_beam.options.pipeline_options import GoogleCloudOptions
from apache_beam.options.pipeline_options import PipelineOptions
@@ -124,7 +123,6 @@ def create_harness(environment, dry_run=False):
RuntimeValueProvider.set_runtime_options(pipeline_options_dict)
sdk_pipeline_options = PipelineOptions.from_dictionary(pipeline_options_dict)
filesystems.FileSystems.set_options(sdk_pipeline_options)
- metric.MetricsFlag.set_default_pipeline_options(sdk_pipeline_options)
pickle_library = sdk_pipeline_options.view_as(SetupOptions).pickle_library
pickler.set_library(pickle_library)