Add new telemetry for cross-cluster-replication feature

[rishabh6788]

Description

Adds a ccr-stats-v2 telemetry device that samples CCR follower stats at index level via the _remote_replication/secondary_stats API. Replication lag is reported directly by the API, so no leader stats are collected and no lag is calculated by OSB.

Prefer lbzip2 for parallel bz2 decompression.

For .bz2 archives, try lbzip2 before pbzip2, falling back to the standard library. Unlike pbzip2, lbzip2 parallelizes decompression of any bz2 file, including the single-stream files used by most published corpora (e.g. big5), so decompression is no longer pinned to a single core.

Generalizes _do_decompress_manually to accept an ordered list of candidate external commands, and adds lbzip2 to the Docker image.

Issues Resolved

[List any issues this PR will resolve]

Testing

• New functionality includes testing

[Describe how this change was tested]

---

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 3bf379a)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 Multiple PR themes Sub-PR theme: Add CcrStatsV2 telemetry device Relevant files: osbenchmark/telemetry.py osbenchmark/worker_coordinator/worker_coordinator.py tests/telemetry_test.py Sub-PR theme: Prefer lbzip2 for parallel bz2 decompression Relevant files: osbenchmark/utils/io.py tests/utils/io_test.py docker/Dockerfile

⚡ Recommended focus areas for review Possible KeyError In CcrStatsV2.on_benchmark_start, when self.indices_per_cluster is set but does not include an entry for every cluster in self.specified_cluster_names, self.indices_per_cluster[cluster_name] will raise a KeyError. Although specified_cluster_names is reset to indices_per_cluster.keys() when a filter is provided, if a user later relies on partial specification (as documented: "Not all clusters need to be specified"), only clusters explicitly listed will be sampled — contradicting the docstring. Consider using .get(cluster_name) to align behavior with the documented intent. def on_benchmark_start(self): for cluster_name in self.specified_cluster_names: recorder = CcrStatsV2Recorder(cluster_name, self.clients[cluster_name], self.metrics_store, self.sample_interval, self.indices_per_cluster[cluster_name] if self.indices_per_cluster else None) sampler = SamplerThread(recorder) self.samplers.append(sampler) sampler.daemon = True # we don't require starting recorders precisely at the same time sampler.start() Missing sample_interval sleep Unlike CcrStatsRecorder, CcrStatsV2Recorder.record() does not call time.sleep(self.sample_interval) between samples. If the SamplerThread expects the recorder to pace itself (as the v1 recorder does), the v2 recorder will hammer the endpoint continuously instead of sampling at the configured interval. Verify the SamplerThread's contract and add pacing if required. def record(self): """ Collect CCR follower stats for indices (optionally) specified in telemetry parameters and push to metrics store. """ try: stats = self.client.transport.perform_request( "GET", "/_remote_replication/secondary_stats", params={"level": "indices"}) except opensearchpy.TransportError: msg = "A transport error occurred while collecting CCR secondary stats on cluster [{}]".format(self.cluster_name) self.logger.exception(msg) raise exceptions.BenchmarkError(msg) for index_name, index_stats in stats.get("indices", {}).items(): if not self._match_list_or_pattern(index_name): continue self.record_stats_per_index(index_name, index_stats)

[github-actions]

PR Code Suggestions ✨

Latest suggestions up to 3bf379a

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Avoid KeyError when accessing per-cluster indices When ccr-stats-v2-indices is provided but does not include an entry for every cluster in specified_cluster_names, this will raise a KeyError. Use .get(cluster_name) to safely default to None when a cluster is not in the indices mapping. osbenchmark/telemetry.py [538-542] def on_benchmark_start(self): for cluster_name in self.specified_cluster_names: recorder = CcrStatsV2Recorder(cluster_name, self.clients[cluster_name], self.metrics_store, self.sample_interval, - self.indices_per_cluster[cluster_name] if self.indices_per_cluster else None) + self.indices_per_cluster.get(cluster_name) if self.indices_per_cluster else None) Suggestion importance[1-10]: 7 __ Why: Valid concern: if indices_per_cluster is set but specified_cluster_names is derived from it (which is the case in this code), then all keys should be present. However, using .get() is a defensive improvement, though the actual bug risk is low since specified_cluster_names becomes self.indices_per_cluster.keys() when the mapping is provided.	Medium
General	Verify recorder is periodically sampled CcrStatsV2 only calls on_benchmark_start/on_benchmark_stop but never actually invokes the recorder in a loop. Other telemetry devices rely on SamplerThread calling recorder.record() repeatedly; ensure CcrStatsV2Recorder is compatible with SamplerThread (i.e. that the loop sleeps for sample_interval), otherwise no periodic sampling will occur. osbenchmark/telemetry.py [486-488] -+class CcrStatsV2(TelemetryDevice): -+ internal = False -+ command = "ccr-stats-v2" +class CcrStatsV2(TelemetryDevice): + internal = False + command = "ccr-stats-v2" Suggestion importance[1-10]: 2 __ Why: This is a verification-only suggestion. The existing CcrStats class follows the same pattern using SamplerThread, so the mechanism is already established and works.	Low

---

Previous suggestions

Suggestions up to commit e6aaf20

Category	Suggestion	Impact
Possible issue	Avoid KeyError on missing cluster filter When ccr-stats-v2-indices is provided but does not include all clusters in specified_cluster_names (which happens if only some clusters are specified), self.indices_per_cluster[cluster_name] will raise KeyError. Use .get(cluster_name) to safely default to None for clusters without an explicit filter. osbenchmark/telemetry.py [538-542] def on_benchmark_start(self): for cluster_name in self.specified_cluster_names: recorder = CcrStatsV2Recorder(cluster_name, self.clients[cluster_name], self.metrics_store, self.sample_interval, - self.indices_per_cluster[cluster_name] if self.indices_per_cluster else None) + self.indices_per_cluster.get(cluster_name) if self.indices_per_cluster else None) Suggestion importance[1-10]: 5 __ Why: In practice, specified_cluster_names is reassigned to self.indices_per_cluster.keys() when the indices dict is provided, so a KeyError is unlikely. However, using .get() is a defensive improvement.	Low
General	Warn when all external decompressors fail If all external tools are available but each of them fails, the code silently falls back to the standard library without emitting any warning, which will hide runtime errors from users. Log a warning when the fallback occurs because every available external tool failed. osbenchmark/utils/io.py [371-378] available = [cmd for cmd in decompressor_commands if is_executable(cmd[0])] for decompressor_args in available: if _do_decompress_manually_external(target_directory, filename, base_path_without_extension, decompressor_args): return if not available: logging.getLogger(__name__).warning("%s not found in PATH. Using standard library, decompression will take longer.", " or ".join(cmd[0] for cmd in decompressor_commands)) +elif available: + logging.getLogger(__name__).warning("All available external decompressors failed for [%s]. Falling back to standard library.", + filename) Suggestion importance[1-10]: 4 __ Why: Adding a warning when all available external tools fail improves observability, though _do_decompress_manually_external itself already logs warnings on failure, making this a minor improvement.	Low

Suggestions up to commit 7c916cc

Category	Suggestion	Impact
General	Warn when external decompressors all fail If external tools are available but all fail, the code silently falls back to the stdlib without any warning. Log a warning in this case so operators can diagnose why decompression is slow. osbenchmark/utils/io.py [371-378] available = [cmd for cmd in decompressor_commands if is_executable(cmd[0])] for decompressor_args in available: if _do_decompress_manually_external(target_directory, filename, base_path_without_extension, decompressor_args): return if not available: logging.getLogger(__name__).warning("%s not found in PATH. Using standard library, decompression will take longer.", " or ".join(cmd[0] for cmd in decompressor_commands)) +else: + logging.getLogger(__name__).warning( + "All external decompressors [%s] failed. Falling back to the standard library.", + ", ".join(cmd[0] for cmd in available)) Suggestion importance[1-10]: 5 __ Why: A valid observability improvement: without this warning, silent fallback to slower stdlib decompression could go unnoticed. Impact is minor but useful.	Low
Possible issue	Avoid KeyError when indexing per-cluster indices When ccr-stats-v2-indices is provided but does not include an entry for every cluster in specified_cluster_names, self.indices_per_cluster[cluster_name] will raise a KeyError. Use .get(cluster_name) to safely default to no filter for clusters not listed. osbenchmark/telemetry.py [542] def on_benchmark_start(self): for cluster_name in self.specified_cluster_names: recorder = CcrStatsV2Recorder(cluster_name, self.clients[cluster_name], self.metrics_store, self.sample_interval, - self.indices_per_cluster[cluster_name] if self.indices_per_cluster else None) + self.indices_per_cluster.get(cluster_name) if self.indices_per_cluster else None) Suggestion importance[1-10]: 3 __ Why: When indices_per_cluster is set, specified_cluster_names is reassigned to its keys, so the KeyError scenario cannot actually occur. The suggestion is defensive but not addressing a real bug.	Low

[github-actions]

Persistent review updated to latest commit e6aaf20

[github-actions]

Persistent review updated to latest commit 3bf379a

[opensearch-ci-bot]

The backport to 1.x failed. Please backport manually. See failed workflow run: https://github.com/opensearch-project/opensearch-benchmark/actions/runs/31415413359