From b588b401201ac974b78fbb2036dbe95ca7b4e695 Mon Sep 17 00:00:00 2001
From: Loakesh Indiran <66478092+Loaki07@users.noreply.github.com>
Date: Fri, 11 Jul 2025 20:04:14 +0530
Subject: [PATCH] fix: f32 deserialization nats (#7543) (#7544)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

### **User description**
fix #7543


___

### **PR Type**
Bug fix


___

### **Description**
- Add JSON parse error handling in cluster module

- Log and skip invalid node payloads

- Prevent infinite or NaN CPU usage

- Warn and reset invalid CPU usage to zero


___

### **Changes diagram**

```mermaid
flowchart LR
  A["cluster::watch_node_list"] -- "on JSON parse error" --> B["log error and continue"]
  C["sysinfo::get_node_metrics"] -- "on CPU infinite or NaN" --> D["log warning and set zero"]
```


___



### **Changes walkthrough** 📝
<table><thead><tr><th></th><th align="left">Relevant
files</th></tr></thead><tbody><tr><td><strong>Error
handling</strong></td><td><table>
<tr>
  <td>
    <details>
<summary><strong>mod.rs</strong><dd><code>Add JSON parse error
handling</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
</dd></summary>
<hr>

src/common/infra/cluster/mod.rs

<li>Replace <code>unwrap</code> with <code>match</code> on JSON
deserialization<br> <li> Log detailed error and payload on parse
failure<br> <li> Continue loop instead of panicking on invalid node


</details>


  </td>
<td><a
href="https://github.com/openobserve/openobserve/pull/7544/files#diff-d15434a27e2b8406811f5a707b605203b603dad4a262c0642326be860bd2b568">+12/-1</a>&nbsp;
&nbsp; </td>

</tr>

<tr>
  <td>
    <details>
<summary><strong>mod.rs</strong><dd><code>Sanitize CPU usage
values</code>&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;
&nbsp; &nbsp; &nbsp; &nbsp; </dd></summary>
<hr>

src/config/src/utils/sysinfo/mod.rs

<li>Introduce sanity check for CPU usage value<br> <li> Log warning on
infinite or NaN CPU usage<br> <li> Reset <code>cpu_usage</code> to zero
when invalid


</details>


  </td>
<td><a
href="https://github.com/openobserve/openobserve/pull/7544/files#diff-48f8beee2099460df806583ba87a0aff864fe38a4fe74224f06f98c95b7b0e42">+8/-1</a>&nbsp;
&nbsp; &nbsp; </td>

</tr>
</table></td></tr></tr></tbody></table>

___

> <details> <summary> Need help?</summary><li>Type <code>/help how to
...</code> in the comments thread for any questions about PR-Agent
usage.</li><li>Check out the <a
href="https://qodo-merge-docs.qodo.ai/usage-guide/">documentation</a>
for more information.</li></details>
---
 src/common/infra/cluster/mod.rs     | 13 ++++++++++++-
 src/config/src/utils/sysinfo/mod.rs |  9 ++++++++-
 2 files changed, 20 insertions(+), 2 deletions(-)

diff --git a/src/common/infra/cluster/mod.rs b/src/common/infra/cluster/mod.rs
index cff9ae3fb6..ccd1ae40d3 100644
--- a/src/common/infra/cluster/mod.rs
+++ b/src/common/infra/cluster/mod.rs
@@ -326,7 +326,18 @@ async fn watch_node_list() -> Result<()> {
         match ev {
             Event::Put(ev) => {
                 let item_key = ev.key.strip_prefix(key).unwrap();
-                let mut item_value: Node = json::from_slice(&ev.value.unwrap()).unwrap();
+                let mut item_value: Node = match json::from_slice(ev.value.as_ref().unwrap()) {
+                    Ok(v) => v,
+                    Err(e) => {
+                        let value_str = String::from_utf8_lossy(ev.value.as_ref().unwrap());
+                        log::error!(
+                            "[CLUSTER] watch_node_list: error parsing node: {}, payload: {}",
+                            e,
+                            value_str
+                        );
+                        continue;
+                    }
+                };
                 let (_broadcasted, exist) = match NODES.read().await.get(item_key) {
                     Some(v) => (v.broadcasted, item_value.is_same(v)),
                     None => (false, false),
diff --git a/src/config/src/utils/sysinfo/mod.rs b/src/config/src/utils/sysinfo/mod.rs
index 2bf32d6fb5..bc1c92ce6b 100644
--- a/src/config/src/utils/sysinfo/mod.rs
+++ b/src/config/src/utils/sysinfo/mod.rs
@@ -41,7 +41,14 @@ pub struct NodeMetrics {
 /// Get the node running metrics
 pub fn get_node_metrics() -> NodeMetrics {
     let cpu_total = get_cpu_limit();
-    let cpu_usage = get_cpu_usage() / cpu_total as f32;
+    let mut cpu_usage = get_cpu_usage() / cpu_total as f32;
+    if cpu_usage.is_infinite() || cpu_usage.is_nan() {
+        log::warn!(
+            "CPU usage is infinite or NaN, setting to 0. cpu_usage: {}, cpu_total: {cpu_total}",
+            get_cpu_usage()
+        );
+        cpu_usage = 0.0;
+    }
     let memory_total = get_memory_limit();
     let memory_usage = get_memory_usage();
     let tcp_conns = get_tcp_connections();