Merge pull request #19 from input-output-hk/sg/monitor

Monitor

Author: SupernaviX

modules/clock/Cargo.toml

@@ -2,14 +2,14 @@
 
 [package]
 name = "caryatid_module_clock"
-version = "0.12.0"
+version = "0.13.0"
 edition = "2021"
 authors = ["Paul Clark <[email protected]>"]
 description = "Clock module for Caryatid"
 license = "Apache-2.0"
 
 [dependencies]
-caryatid_sdk = { version="0.12", path = "../../sdk" }
+caryatid_sdk = { version="0.13", path = "../../sdk" }
 anyhow = "1.0"
 tokio = { version = "1", features = ["full"] }
 serde_json = "1.0"

modules/clock/src/clock.rs

@@ -2,7 +2,7 @@
 //! Generates regular clock.tick events
 
 use anyhow::Result;
-use caryatid_sdk::{module, Context, MessageBounds, Module};
+use caryatid_sdk::{module, Context, MessageBounds};
 use chrono::{DateTime, Utc};
 use config::Config;
 use std::sync::Arc;
@@ -70,6 +70,7 @@ impl<M: From<ClockTickMessage> + MessageBounds> Clock<M> {
 mod tests {
     use super::*;
     use caryatid_sdk::mock_bus::MockBus;
+    use caryatid_sdk::Module;
     use config::{Config, FileFormat};
     use tokio::sync::{mpsc, watch::Sender, Notify};
     use tokio::time::{timeout, Duration};
@@ -99,6 +100,7 @@ mod tests {
     struct TestSetup {
         module: Arc<dyn Module<Message>>,
         context: Arc<Context<Message>>,
+        startup_watch: Sender<bool>,
     }
 
     impl TestSetup {
@@ -120,12 +122,9 @@ mod tests {
             // Create mock bus
             let bus = Arc::new(MockBus::<Message>::new(&config));
 
+            let startup_watch = Sender::new(false);
             // Create a context
-            let context = Arc::new(Context::new(
-                config.clone(),
-                bus.clone(),
-                Sender::<bool>::new(false),
-            ));
+            let context = Arc::new(Context::new(config.clone(), bus, startup_watch.subscribe()));
 
             // Create the clock
             let clock = Clock::<Message> {
@@ -136,11 +135,12 @@ mod tests {
             Self {
                 module: Arc::new(clock),
                 context,
+                startup_watch,
             }
         }
 
         fn start(&self) {
-            let _ = self.context.startup_watch.send(true);
+            let _ = self.startup_watch.send(true);
         }
     }
 

modules/playback/Cargo.toml

@@ -2,14 +2,14 @@
 
 [package]
 name = "caryatid_module_playback"
-version = "0.9.0"
+version = "0.10.0"
 edition = "2021"
 authors = ["Alex Woods <[email protected]>"]
 description = "Message playback module for Caryatid"
 license = "Apache-2.0"
 
 [dependencies]
-caryatid_sdk = { version="0.12", path = "../../sdk" }
+caryatid_sdk = { version="0.13", path = "../../sdk" }
 anyhow = "1.0"
 serde_json = "1.0"
 config = "0.15.11"

modules/record/Cargo.toml

@@ -2,14 +2,14 @@
 
 [package]
 name = "caryatid_module_record"
-version = "0.9.0"
+version = "0.10.0"
 edition = "2021"
 authors = ["Alex Woods <[email protected]>"]
 description = "Message recording module for Caryatid"
 license = "Apache-2.0"
 
 [dependencies]
-caryatid_sdk = { version="0.12", path = "../../sdk" }
+caryatid_sdk = { version="0.13", path = "../../sdk" }
 anyhow = "1.0"
 serde_json = "1.0"
 config = "0.15.11"

modules/rest_server/Cargo.toml

@@ -2,14 +2,14 @@
 
 [package]
 name = "caryatid_module_rest_server"
-version = "0.14.0"
+version = "0.15.0"
 edition = "2021"
 authors = ["Paul Clark <[email protected]>"]
 description = "REST server module for Caryatid"
 license = "Apache-2.0"
 
 [dependencies]
-caryatid_sdk = { version="0.12", path = "../../sdk" }
+caryatid_sdk = { version="0.13", path = "../../sdk" }
 anyhow = "1.0"
 serde_json = "1.0"
 config = "0.15.11"

modules/rest_server/src/rest_server.rs

@@ -2,7 +2,7 @@
 //! Provides a REST endpoint which integrates with the message bus
 
 use anyhow::Result;
-use caryatid_sdk::{module, Context, MessageBounds, Module};
+use caryatid_sdk::{module, Context, MessageBounds};
 use config::Config;
 use std::{collections::HashMap, sync::Arc};
 use tracing::{error, info};
@@ -176,6 +176,7 @@ impl<M: From<RESTRequest> + GetRESTResponse + MessageBounds> RESTServer<M> {
 mod tests {
     use super::*;
     use caryatid_sdk::mock_bus::MockBus;
+    use caryatid_sdk::Module;
     use config::{Config, FileFormat};
     use futures::future;
     use hyper::Client;
@@ -227,6 +228,7 @@ mod tests {
     struct TestSetup {
         module: Arc<dyn Module<Message>>,
         context: Arc<Context<Message>>,
+        startup_watch: Sender<bool>,
     }
 
     impl TestSetup {
@@ -249,10 +251,11 @@ mod tests {
             let mock_bus = Arc::new(MockBus::<Message>::new(&config));
 
             // Create a context
+            let startup_watch = Sender::new(false);
             let context = Arc::new(Context::new(
                 config.clone(),
-                mock_bus.clone(),
-                Sender::<bool>::new(false),
+                mock_bus,
+                startup_watch.subscribe(),
             ));
 
             // Create the server
@@ -264,11 +267,12 @@ mod tests {
             Self {
                 module: Arc::new(rest_server),
                 context,
+                startup_watch,
             }
         }
 
         fn start(&self) {
-            let _ = self.context.startup_watch.send(true);
+            let _ = self.startup_watch.send(true);
         }
     }
 

modules/spy/Cargo.toml

@@ -2,14 +2,14 @@
 
 [package]
 name = "caryatid_module_spy"
-version = "0.12.0"
+version = "0.13.0"
 edition = "2021"
 authors = ["Paul Clark <[email protected]>"]
 description = "Spy module for Caryatid"
 license = "Apache-2.0"
 
 [dependencies]
-caryatid_sdk = { version="0.12", path = "../../sdk" }
+caryatid_sdk = { version="0.13", path = "../../sdk" }
 anyhow = "1.0"
 tokio = { version = "1", features = ["full"] }
 serde_json = "1.0"

monitor.md

@@ -0,0 +1,23 @@
+# Module monitor
+
+Caryatid supports a "monitor" to see the activity of the different modules and topics in a process. It's implemented as an overlay to the message bus, which tracks reads/writes to each topic in an in-memory `DashMap`.
+
+This monitor is useful for debugging slowness/stoppages, or for understanding the interactions between modules. It's not intended to serve as a form of instrumentation.
+
+Below is how to enable it:
+```toml
+[monitor]
+output = "monitor.json" # Write to a file named monitor.json
+frequency_secs = 5.0    # every 5 seconds
+```
+
+Every module shows a list of "reads" (streams it reads from) and "writes" (streams it writes to). 
+
+We track this information for reads:
+ - `read`: how many messages this module has read from this topic.
+ - `unread`: how many messages are available for this module on this topic. This is based on how many messages were published by another module, it will be an underestimate if messages are sent over rabbitmq.
+ - `pending_for`: how long has this module been waiting to read a message on this topic.
+
+We track this information for writes:
+- `written`: how many messages this module has written to this topic.
+- `pending_for`: how long has this module been waiting to write a message to this topic. If this is set, the topic is congested; some module is subscribed to it but not reading from it.

process/Cargo.toml

@@ -1,19 +1,20 @@
 # Caraytid process - loads and runs modules
 [package]
 name = "caryatid_process"
-version = "0.12.2"
+version = "0.13.0"
 edition = "2021"
 authors = ["Paul Clark <[email protected]>"]
 description = "Library for building a Caryatid process"
 license = "Apache-2.0"
 
 [dependencies]
-caryatid_sdk = { version="0.12", path = "../sdk" }
+caryatid_sdk = { version="0.13", path = "../sdk" }
 futures = "0.3"
 anyhow = "1.0"
 tokio = { version = "1", features = ["full"] }
 config = "0.15.11"
 minicbor-serde = { version = "0.6", features = ["alloc"] }
+dashmap = "6"
 tracing = "0.1.40"
 serde = "1.0.210"
 serde_json = "1.0"