feat: implement grpc client

Signed-off-by: Alexandre Gomez <[email protected]>

controller/lib/src/grpc_client/interface.rs

@@ -0,0 +1,153 @@
+use log::info;
+use proto::scheduler::instance_service_client::InstanceServiceClient;
+use proto::scheduler::node_service_client::NodeServiceClient;
+use proto::scheduler::{
+    Instance, InstanceIdentifier, InstanceStatus, NodeRegisterRequest, NodeStatus,
+    NodeUnregisterRequest, NodeUnregisterResponse,
+};
+use tonic::transport::{Channel, Error};
+use tonic::{Request, Response, Status, Streaming};
+
+pub enum SchedulerClientInterfaceError {
+    ConnectionError(Error),
+    RequestFailed(Status),
+}
+
+pub struct SchedulerClientInterface {
+    node_client: NodeServiceClient<Channel>,
+    instance_client: InstanceServiceClient<Channel>,
+}
+
+impl SchedulerClientInterface {
+    pub async fn new(
+        node_client_address: String,
+        instance_client_address: String,
+    ) -> Result<Self, SchedulerClientInterfaceError> {
+        info!(
+            "Starting gRPC client for scheduler Node Service on {}",
+            node_client_address,
+        );
+
+        let node_client = match NodeServiceClient::connect(node_client_address).await {
+            Ok(client) => client,
+            Err(e) => return Err(SchedulerClientInterfaceError::ConnectionError(e)),
+        };
+
+        info!(
+            "Starting gRPC client for scheduler Instance Service on {}",
+            instance_client_address,
+        );
+
+        let instance_client = match InstanceServiceClient::connect(instance_client_address).await {
+            Ok(client) => client,
+            Err(e) => return Err(SchedulerClientInterfaceError::ConnectionError(e)),
+        };
+
+        Ok(Self {
+            node_client,
+            instance_client,
+        })
+    }
+
+    pub async fn register_node(
+        &mut self,
+        request: Request<NodeRegisterRequest>,
+    ) -> Result<Response<Streaming<NodeStatus>>, SchedulerClientInterfaceError> {
+        let remote_address = request.remote_addr().unwrap();
+
+        info!(
+            "Calling gRPC procedure \"register_node\" to {}",
+            remote_address
+        );
+
+        match self.node_client.register(request).await {
+            Ok(response) => Ok(response),
+            Err(e) => Err(SchedulerClientInterfaceError::RequestFailed(e)),
+        }
+    }
+
+    pub async fn unregister_node(
+        &mut self,
+        request: Request<NodeUnregisterRequest>,
+    ) -> Result<Response<NodeUnregisterResponse>, SchedulerClientInterfaceError> {
+        let remote_address = request.remote_addr().unwrap();
+
+        info!(
+            "Calling gRPC procedure \"unregister_node\" to {}",
+            remote_address
+        );
+
+        match self.node_client.unregister(request).await {
+            Ok(response) => Ok(response),
+            Err(e) => Err(SchedulerClientInterfaceError::RequestFailed(e)),
+        }
+    }
+
+    pub async fn create_instance(
+        &mut self,
+        request: Request<Instance>,
+    ) -> Result<Response<Streaming<InstanceStatus>>, SchedulerClientInterfaceError> {
+        let remote_address = request.remote_addr().unwrap();
+
+        info!(
+            "Calling gRPC procedure \"create_instance\" to {}",
+            remote_address
+        );
+
+        match self.instance_client.create(request).await {
+            Ok(response) => Ok(response),
+            Err(e) => Err(SchedulerClientInterfaceError::RequestFailed(e)),
+        }
+    }
+
+    pub async fn destroy_instance(
+        &mut self,
+        request: Request<InstanceIdentifier>,
+    ) -> Result<Response<()>, SchedulerClientInterfaceError> {
+        let remote_address = request.remote_addr().unwrap();
+
+        info!(
+            "Calling gRPC procedure \"destroy_instance\" to {}",
+            remote_address
+        );
+
+        match self.instance_client.destroy(request).await {
+            Ok(response) => Ok(response),
+            Err(e) => Err(SchedulerClientInterfaceError::RequestFailed(e)),
+        }
+    }
+
+    pub async fn start_instance(
+        &mut self,
+        request: Request<InstanceIdentifier>,
+    ) -> Result<Response<()>, SchedulerClientInterfaceError> {
+        let remote_address = request.remote_addr().unwrap();
+
+        info!(
+            "Calling gRPC procedure \"start_instance\" to {}",
+            remote_address
+        );
+
+        match self.instance_client.start(request).await {
+            Ok(response) => Ok(response),
+            Err(e) => Err(SchedulerClientInterfaceError::RequestFailed(e)),
+        }
+    }
+
+    pub async fn stop_instance(
+        &mut self,
+        request: Request<InstanceIdentifier>,
+    ) -> Result<Response<()>, SchedulerClientInterfaceError> {
+        let remote_address = request.remote_addr().unwrap();
+
+        info!(
+            "Calling gRPC procedure \"stop_instance\" to {}",
+            remote_address
+        );
+
+        match self.instance_client.stop(request).await {
+            Ok(response) => Ok(response),
+            Err(e) => Err(SchedulerClientInterfaceError::RequestFailed(e)),
+        }
+    }
+}

controller/lib/src/grpc_client/mod.rs

@@ -0,0 +1 @@
+pub mod interface;