Feature: Node topology plugin (#325)

* add basic implementation for status update plugins and intitial webhook plugin

* introduce webhook plugin

* Introduce Node grouping plugin for status update & scheduler 

* automatically generate p2p id on worker

Author: JannikSt

Cargo.lock

@@ -78,6 +78,10 @@ watch-worker:
 	set -a; source ${ENV_FILE}; set +a; \
 	cargo watch -w crates/worker/src -x "run --bin worker -- run --port 8091 --external-ip $${WORKER_EXTERNAL_IP:-localhost} --compute-pool-id $$WORKER_COMPUTE_POOL_ID --skip-system-checks $${LOKI_URL:+--loki-url $${LOKI_URL}} --log-level $${LOG_LEVEL:-info}"
 
+watch-worker-two:
+	set -a; source ${ENV_FILE}; set +a; \
+	cargo watch -w crates/worker/src -x "run --bin worker -- run --port 8092 --private-key-node $${PRIVATE_KEY_NODE_2} --private-key-provider $${PRIVATE_KEY_PROVIDER} --external-ip $${WORKER_EXTERNAL_IP:-localhost} --compute-pool-id $$WORKER_COMPUTE_POOL_ID --skip-system-checks $${LOKI_URL:+--loki-url $${LOKI_URL}} --log-level $${LOG_LEVEL:-info}"
+
 watch-check:
 	cargo watch -w crates/worker/src -x "run --bin worker -- check"	
 
@@ -87,7 +91,7 @@ watch-validator:
 
 watch-orchestrator:
 	set -a; source ${ENV_FILE}; set +a; \
-	cargo watch -w crates/orchestrator/src -x "run --bin orchestrator -- -r $$RPC_URL -k $$POOL_OWNER_PRIVATE_KEY -d 0  -p 8090 -i 10 -u http://localhost:8090 --s3-credentials $$S3_CREDENTIALS --compute-pool-id $$WORKER_COMPUTE_POOL_ID --bucket-name $$BUCKET_NAME -l $${LOG_LEVEL:-info} --hourly-s3-upload-limit $${HOURLY_S3_LIMIT:-3}"
+	cargo watch -w crates/orchestrator/src -x "run --bin orchestrator -- -r $$RPC_URL -k $$POOL_OWNER_PRIVATE_KEY -d 0  -p 8090 -i 10 -u http://localhost:8090 --s3-credentials $$S3_CREDENTIALS --compute-pool-id $$WORKER_COMPUTE_POOL_ID --bucket-name $$BUCKET_NAME -l $${LOG_LEVEL:-info} --hourly-s3-upload-limit $${HOURLY_S3_LIMIT:-3} --with-basic-group-plugin --group-size 1"
 
 build-worker:
 	cargo build --release --bin worker
@@ -155,6 +159,23 @@ watch-worker-remote: setup-remote setup-tunnel sync-remote
 				--auto-accept \
 				2>&1 | tee worker.log\"'"
 
+.PHONY: watch-worker-remote-two
+watch-worker-remote-two: setup-remote setup-tunnel sync-remote
+	$(SSH_CONNECTION) -t "cd ~/$(notdir $(CURDIR)) && \
+		export PATH=\"\$$HOME/.cargo/bin:\$$PATH\" && \
+		. \"\$$HOME/.cargo/env\" && \
+		export TERM=xterm-256color && \
+		bash --login -i -c '\
+			set -a && source .env && set +a && \
+			export EXTERNAL_IP=$(EXTERNAL_IP) && \
+			clear && \
+			RUST_BACKTRACE=1 RUST_LOG=debug cargo watch -w crates/worker/src -x \"run --bin worker -- run \
+				--port $(PORT) \
+				--compute-pool-id \$$WORKER_COMPUTE_POOL_ID \
+				--auto-accept \
+				--private-key-node \$$PRIVATE_KEY_NODE_2 \
+				2>&1 | tee worker.log\"'"
+
 # Kill SSH tunnel
 .PHONY: kill-tunnel
 kill-tunnel:
@@ -167,6 +188,12 @@ remote-worker:
 	@trap 'make kill-tunnel' EXIT; \
 	make watch-worker-remote
 
+.PHONY: remote-worker-two
+remote-worker-two:
+	@trap 'make kill-tunnel' EXIT; \
+	make watch-worker-remote-two
+
+
 # testing:
 eject-node:
 	set -a; source ${ENV_FILE}; set +a; \

Makefile

@@ -144,6 +144,7 @@ mod tests {
             node_store: Arc::new(NodeStore::new(RedisStore::new_test())),
             contracts: None,
             last_chain_sync: Arc::new(Mutex::new(None::<SystemTime>)),
+            only_one_node_per_ip: true,
         };
         let app = test::init_service(
             App::new()
@@ -182,6 +183,7 @@ mod tests {
             node_store: Arc::new(NodeStore::new(RedisStore::new_test())),
             contracts: None,
             last_chain_sync: Arc::new(Mutex::new(None::<SystemTime>)),
+            only_one_node_per_ip: true,
         };
         let app = test::init_service(
             App::new()

crates/discovery/src/api/routes/get_nodes.rs

@@ -92,18 +92,20 @@ pub async fn register_node(
     let existing_node_by_ip = data
         .node_store
         .get_active_node_by_ip(update_node.ip_address.clone());
-    if let Ok(Some(existing_node)) = existing_node_by_ip {
-        if existing_node.id != update_node.id {
-            warn!(
+    if data.only_one_node_per_ip {
+        if let Ok(Some(existing_node)) = existing_node_by_ip {
+            if existing_node.id != update_node.id {
+                warn!(
                 "Node {} tried to change discovery but another active node is already registered to this IP address",
                 update_node.id
             );
-            debug!("Existing node: {:?}", existing_node);
-            debug!("Update node: {:?}", update_node);
-            return HttpResponse::BadRequest().json(ApiResponse::new(
-                false,
-                "Another active Node is already registered to this IP address",
-            ));
+                debug!("Existing node: {:?}", existing_node);
+                debug!("Update node: {:?}", update_node);
+                return HttpResponse::BadRequest().json(ApiResponse::new(
+                    false,
+                    "Another active Node is already registered to this IP address",
+                ));
+            }
         }
     }
 
@@ -215,6 +217,7 @@ mod tests {
             node_store: Arc::new(NodeStore::new(RedisStore::new_test())),
             contracts: None,
             last_chain_sync: Arc::new(Mutex::new(None::<SystemTime>)),
+            only_one_node_per_ip: true,
         };
 
         let app = test::init_service(
@@ -270,6 +273,7 @@ mod tests {
             node_store: Arc::new(NodeStore::new(RedisStore::new_test())),
             contracts: None,
             last_chain_sync: Arc::new(Mutex::new(None::<SystemTime>)),
+            only_one_node_per_ip: true,
         };
 
         let validate_signatures =
@@ -394,6 +398,7 @@ mod tests {
             node_store: Arc::new(NodeStore::new(RedisStore::new_test())),
             contracts: None,
             last_chain_sync: Arc::new(Mutex::new(None::<SystemTime>)),
+            only_one_node_per_ip: true,
         };
 
         let validate_signatures =
@@ -458,6 +463,7 @@ mod tests {
             node_store: Arc::new(NodeStore::new(RedisStore::new_test())),
             contracts: None,
             last_chain_sync: Arc::new(Mutex::new(None::<SystemTime>)),
+            only_one_node_per_ip: true,
         };
 
         let validate_signatures =
@@ -520,6 +526,7 @@ mod tests {
             node_store: Arc::new(NodeStore::new(RedisStore::new_test())),
             contracts: None,
             last_chain_sync: Arc::new(Mutex::new(None::<SystemTime>)),
+            only_one_node_per_ip: true,
         };
 
         app_state.node_store.register_node(node.clone()).unwrap();

crates/discovery/src/api/routes/node.rs

@@ -23,6 +23,7 @@ pub struct AppState {
     pub node_store: Arc<NodeStore>,
     pub contracts: Option<Arc<Contracts>>,
     pub last_chain_sync: Arc<Mutex<Option<SystemTime>>>,
+    pub only_one_node_per_ip: bool,
 }
 
 async fn health_check(app_state: web::Data<AppState>) -> HttpResponse {
@@ -76,6 +77,7 @@ pub async fn start_server(
     contracts: Arc<Contracts>,
     platform_api_key: String,
     last_chain_sync: Arc<Mutex<Option<SystemTime>>>,
+    only_one_node_per_ip: bool,
 ) -> std::io::Result<()> {
     info!("Starting server at http://{}:{}", host, port);
 
@@ -91,6 +93,7 @@ pub async fn start_server(
         node_store,
         contracts: Some(contracts),
         last_chain_sync,
+        only_one_node_per_ip,
     };
 
     // it seems we have a validator for the validator

crates/discovery/src/api/server.rs

@@ -32,6 +32,10 @@ struct Args {
     /// Port
     #[arg(short = 'P', long, default_value = "8089")]
     port: u16,
+
+    /// Only one node per IP address
+    #[arg(long)]
+    only_one_node_per_ip: Option<bool>,
 }
 
 #[tokio::main]
@@ -86,6 +90,7 @@ async fn main() -> Result<()> {
         contracts_clone,
         args.platform_api_key,
         heartbeat_server_clone,
+        args.only_one_node_per_ip.unwrap_or(false),
     )
     .await
     {

crates/discovery/src/main.rs

@@ -23,6 +23,8 @@ ENV LOG_LEVEL=""
 ENV HOURLY_S3_UPLOAD_LIMIT="2"
 ENV WEBHOOK_URLS=""
 ENV MODE="full"
+ENV WITH_BASIC_GROUP_PLUGIN="false"
+ENV GROUP_SIZE="4"
 
 RUN echo '#!/bin/sh\n\
 exec /usr/local/bin/orchestrator \
@@ -44,6 +46,8 @@ $([ ! -z "$BUCKET_NAME" ] && echo "--bucket-name $BUCKET_NAME") \
 $([ ! -z "$LOG_LEVEL" ] && echo "--log-level $LOG_LEVEL") \
 $([ ! -z "$HOURLY_S3_UPLOAD_LIMIT" ] && echo "--hourly-s3-upload-limit $HOURLY_S3_UPLOAD_LIMIT") \
 $([ ! -z "$WEBHOOK_URLS" ] && echo "--webhook-urls $WEBHOOK_URLS") \
+$([ "$WITH_BASIC_GROUP_PLUGIN" = "true" ] && echo "--with-basic-group-plugin") \
+$([ ! -z "$GROUP_SIZE" ] && echo "--group-size $GROUP_SIZE") \
 "$@"' > /entrypoint.sh && \
 chmod +x /entrypoint.sh
 

crates/orchestrator/Dockerfile

@@ -32,6 +32,14 @@ async fn heartbeat(
         task_info.task_id,
         task_info.task_state,
     );
+
+    if let Some(p2p_id) = &heartbeat.p2p_id {
+        app_state
+            .store_context
+            .node_store
+            .update_node_p2p_id(&node_address, p2p_id);
+    }
+
     app_state.store_context.heartbeat_store.beat(&heartbeat);
     app_state
         .store_context
@@ -111,6 +119,7 @@ mod tests {
                 metrics: None,
                 version: None,
                 timestamp: None,
+                p2p_id: None,
             })
         );
     }
@@ -167,6 +176,7 @@ mod tests {
             metrics: None,
             version: None,
             timestamp: None,
+            p2p_id: None,
         };
         assert_eq!(value, heartbeat);
     }

crates/orchestrator/src/api/routes/heartbeat.rs

@@ -270,6 +270,7 @@ mod tests {
             task_state: None,
             version: None,
             last_status_change: None,
+            p2p_id: None,
         };
         app_state.store_context.node_store.add_node(node.clone());
 

crates/orchestrator/src/api/routes/nodes.rs

@@ -34,7 +34,7 @@ pub async fn create_test_app_state() -> Data<AppState> {
 
     let store_context = Arc::new(StoreContext::new(store.clone()));
     let mode = ServerMode::Full;
-    let scheduler = Scheduler::new(store_context.clone());
+    let scheduler = Scheduler::new(store_context.clone(), vec![]);
     Data::new(AppState {
         store_context: store_context.clone(),
         contracts: None,