Add new telemetry for cross-cluster-replication feature (#1102)

[rishabh6788]

(cherry picked from commit 7a0e191)

Description

[Describe what this change achieves]

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 Code Analyzer ❗

AI-powered 'Code-Diff-Analyzer' found issues on commit 1a89588.

⛔ Hard block: Issues at High severity or above will block this PR from merging.

Path	Line	Severity	Description
docker/Dockerfile	10	high	New system package 'lbzip2' added to apt-get install in the build stage of the container. Per mandatory policy, all container file package installs must be flagged for maintainer verification regardless of apparent legitimacy.
docker/Dockerfile	40	high	New system package 'lbzip2' added to apt-get install in the runtime stage of the container. Per mandatory policy, all container file package installs must be flagged for maintainer verification regardless of apparent legitimacy.

The table above displays the top 10 most important findings.

Total: 2 | Critical: 0 | High: 2 | Medium: 0 | Low: 0

---

Pull Requests Author(s): Please update your Pull Request according to the report above.

Repository Maintainer(s): You can bypass diff analyzer by adding label skip-diff-analyzer after reviewing the changes carefully, then re-run failed actions. To re-enable the analyzer, remove the label, then re-run all actions.

---

⚠️ Note: The Code-Diff-Analyzer helps protect against potentially harmful code patterns. Please ensure you have thoroughly reviewed the changes beforehand.

Thanks.

[github-actions]

PR Reviewer Guide 🔍

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: Support lbzip2 and multiple decompressor fallbacks 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 ccr-stats-v2-indices is provided but does not include every cluster in --target-hosts, self.specified_cluster_names is set to self.indices_per_cluster.keys(), so this path is safe. However, when ccr-stats-v2-indices is not provided, self.specified_cluster_names equals all client keys and self.indices_per_cluster is None, which is handled. The risk is if indices_per_cluster is provided as an empty dict {}: self.indices_per_cluster is truthy check if self.indices_per_cluster: is False, so specified_cluster_names remains all clients, but then self.indices_per_cluster[cluster_name] in on_benchmark_start will raise TypeError since indices_per_cluster is {} (truthy would be False — actually falsy, so the ternary picks None). Verify behavior with empty dict input; consider validating that ccr-stats-v2-indices is non-empty if provided. if self.indices_per_cluster: for cluster_name in self.indices_per_cluster.keys(): if cluster_name not in clients: raise exceptions.SystemSetupError( "The telemetry parameter 'ccr-stats-v2-indices' must be a JSON Object with keys matching " "the cluster names [{}] specified in --target-hosts " "but it had [{}].".format(",".join(sorted(clients.keys())), cluster_name)) self.specified_cluster_names = self.indices_per_cluster.keys() self.metrics_store = metrics_store self.samplers = [] 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) Missing sampler_interval CcrStatsV2 creates SamplerThread(recorder) but CcrStatsV2Recorder is not shown to expose a sample_interval attribute usage compatible with SamplerThread — verify SamplerThread reads recorder.sample_interval (it is set in __init__, so this is likely fine). However, unlike CcrStatsRecorder, CcrStatsV2Recorder.record() raises BenchmarkError on transport errors. If SamplerThread does not catch exceptions, a single transient transport failure will terminate the sampler thread for the remainder of the benchmark, silently stopping CCR v2 telemetry collection. Consider logging and continuing instead of raising. 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)

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
Possible issue	Avoid KeyError for unspecified clusters When ccr-stats-v2-indices is provided but does not include an entry for every cluster in --target-hosts, self.indices_per_cluster[cluster_name] will raise KeyError for the missing clusters. Use .get(cluster_name) so that clusters without an explicit filter simply record all indices, matching the docstring which states "Not all clusters need to be specified". osbenchmark/telemetry.py [538-547] if self.indices_per_cluster: for cluster_name in self.indices_per_cluster.keys(): if cluster_name not in clients: raise exceptions.SystemSetupError( "The telemetry parameter 'ccr-stats-v2-indices' must be a JSON Object with keys matching " "the cluster names [{}] specified in --target-hosts " "but it had [{}].".format(",".join(sorted(clients.keys())), cluster_name)) self.specified_cluster_names = self.indices_per_cluster.keys() self.metrics_store = metrics_store self.samplers = [] 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]: 4 __ Why: The suggestion is technically wrong in claim: when indices_per_cluster is set, self.specified_cluster_names is reassigned to self.indices_per_cluster.keys(), so iterating won't hit a missing key. However, the docstring says "Not all clusters need to be specified", which contradicts current behavior—using .get() would align with the documented intent, so there's some minor merit.	Low