MetadataManager init

This introduces `MetadataManager`, a node-level service that provides uniform access to different metadata kinds. This PR focuses on the nodes configuration only.

`Metadata` is a cheaply cloneable handle that components can use to access metadata. `MetadataWriter` is only shared with components that can propose updates to the cached metadata.

Additionally, this changes the node startup procedure to be one step closer to the original design. Worker attachment is only concerned with partition assignment and node IDs are directly fetched from NodesConfiguration.

In this, nodes configuration will be statically created on startup if the bootstrap_cluster option is set (set by default at the moment). The server will fail to start if this is unset until we implement fetching nodes configuration from metadata store.

The admin_address is not used anymore on startup but will be replaced with metadata address later for remote nodes config fetching.

crates/network/Cargo.toml

@@ -16,9 +16,12 @@ anyhow = { workspace = true }
 bytes = { workspace = true }
 drain = { workspace = true }
 futures = { workspace = true }
+http = { workspace = true }
 pin-project = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
+tonic = { workspace = true }
+tower = { workspace = true }
 tracing = { workspace = true }
 
 [dev-dependencies]

crates/network/src/lib.rs

@@ -16,6 +16,7 @@ use tokio::sync::mpsc;
 
 mod routing;
 mod unbounded_handle;
+pub mod utils;
 
 pub use routing::{Network, PartitionProcessorSender, RoutingError};
 pub use unbounded_handle::UnboundedNetworkHandle;

crates/network/src/utils.rs

@@ -0,0 +1,39 @@
+// Copyright (c) 2024 -  Restate Software, Inc., Restate GmbH.
+// All rights reserved.
+//
+// Use of this software is governed by the Business Source License
+// included in the LICENSE file.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0.
+
+use std::time::Duration;
+
+use http::Uri;
+use restate_types::nodes_config::AdvertisedAddress;
+use tokio::net::UnixStream;
+use tonic::transport::{Channel, Endpoint};
+use tower::service_fn;
+
+pub fn create_grpc_channel_from_network_address(
+    address: AdvertisedAddress,
+) -> Result<Channel, http::Error> {
+    let channel = match address {
+        AdvertisedAddress::Uds(uds_path) => {
+            // dummy endpoint required to specify an uds connector, it is not used anywhere
+            Endpoint::try_from("/")
+                .expect("/ should be a valid Uri")
+                .connect_with_connector_lazy(service_fn(move |_: Uri| {
+                    UnixStream::connect(uds_path.clone())
+                }))
+        }
+        AdvertisedAddress::Http(uri) => {
+            // todo: Make the channel settings configurable
+            Channel::builder(uri)
+                .connect_timeout(Duration::from_secs(5))
+                .connect_lazy()
+        }
+    };
+    Ok(channel)
+}

crates/node-services/proto/cluster_controller.proto

@@ -14,13 +14,12 @@ import "dev/restate/common/common.proto";
 package dev.restate.cluster_controller;
 
 service ClusterControllerSvc {
-  // Attach node at cluster controller
+  // Attach worker at cluster controller
   rpc AttachNode(AttachmentRequest) returns (AttachmentResponse);
 }
 
 message AttachmentRequest {
-  string node_name = 1;
-  optional dev.restate.common.NodeId node_id = 2;
+  optional dev.restate.common.NodeId node_id = 1;
 }
 
 message AttachmentResponse {}

crates/node/Cargo.toml

@@ -17,12 +17,12 @@ options_schema = [
     "restate-cluster-controller/options_schema"]
 
 [dependencies]
-
 restate-admin = { workspace = true }
 restate-bifrost = { workspace = true }
 restate-cluster-controller = { workspace = true }
 restate-errors = { workspace = true }
 restate-meta = { workspace = true }
+restate-network = { workspace = true }
 restate-node-services = { workspace = true }
 restate-schema-api = { workspace = true }
 restate-schema-impl = { workspace = true }
@@ -34,13 +34,16 @@ restate-worker = { workspace = true }
 restate-worker-api = { workspace = true }
 
 anyhow = { workspace = true }
+arc-swap = { workspace = true }
 arrow-flight = { workspace = true }
 async-trait = { workspace = true }
 axum = { workspace = true }
+bincode = { workspace = true }
 codederror = { workspace = true }
 datafusion = { workspace = true }
 derive_builder = { workspace = true }
 drain = { workspace = true }
+enum-map = { workspace = true }
 enumset = { workspace = true }
 futures = { workspace = true }
 hostname = { version = "0.3.1" }
@@ -54,9 +57,10 @@ metrics-util = { version = "0.16.0" }
 rocksdb = { workspace = true }
 schemars = { workspace = true, optional = true }
 serde = { workspace = true }
-bincode = { workspace = true }
 serde_json = { workspace = true }
 serde_with = { workspace = true }
+strum = { workspace = true }
+strum_macros = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tokio-util = { workspace = true }