fix: node cache status (#3170)

src/handler/http/request/status/mod.rs

@@ -290,6 +290,15 @@ async fn get_stream_schema_status() -> (usize, usize, usize) {
             mem_size += schema.size();
         }
     }
+    drop(r);
+    let r = STREAM_SCHEMAS_LATEST.read().await;
+    for (key, schema) in r.iter() {
+        stream_num += 1;
+        stream_schema_num += 1;
+        mem_size += key.len();
+        mem_size += schema.size();
+    }
+    drop(r);
     (stream_num, stream_schema_num, mem_size)
 }
 

src/service/db/schema.rs

@@ -337,14 +337,14 @@ pub async fn update_metadata(
         db::NEED_WATCH,
         None,
         Box::new(move |value| {
-            let latest_schema = match value {
-                None => Schema::empty(),
+            let (latest_schema, not_exists) = match value {
+                None => (Schema::empty(), true),
                 Some(value) => {
                     let mut schemas: Vec<Schema> = json::from_slice(&value)?;
                     if schemas.is_empty() {
-                        Schema::empty()
+                        (Schema::empty(), false)
                     } else {
-                        schemas.remove(schemas.len() - 1)
+                        (schemas.remove(schemas.len() - 1), false)
                     }
                 }
             };
@@ -352,11 +352,32 @@ pub async fn update_metadata(
             for (k, v) in metadata.iter() {
                 schema_metadata.insert(k.clone(), v.clone());
             }
+            let start_dt = match schema_metadata.get("created_at") {
+                Some(v) => v.parse().unwrap(),
+                None => Utc::now().timestamp_micros(),
+            };
+            if !schema_metadata.contains_key("created_at") {
+                schema_metadata.insert("created_at".to_string(), start_dt.to_string());
+            }
+            if !schema_metadata.contains_key("start_dt") {
+                schema_metadata.insert("start_dt".to_string(), start_dt.to_string());
+            }
             let new_schema = vec![latest_schema.with_metadata(schema_metadata)];
-            Ok(Some((
-                Some(json::to_vec(&new_schema).unwrap().into()),
-                None,
-            )))
+            if not_exists {
+                Ok(Some((
+                    None,
+                    Some((
+                        key,
+                        json::to_vec(&new_schema).unwrap().into(),
+                        Some(start_dt),
+                    )),
+                )))
+            } else {
+                Ok(Some((
+                    Some(json::to_vec(&new_schema).unwrap().into()),
+                    None,
+                )))
+            }
         }),
     )
     .await?;