You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@teaclave.apache.org by ms...@apache.org on 2020/02/13 02:04:07 UTC
[incubator-teaclave] branch develop updated: [management] Implement
create/get task (#222)
This is an automated email from the ASF dual-hosted git repository.
mssun pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/incubator-teaclave.git
The following commit(s) were added to refs/heads/develop by this push:
new d5404a1 [management] Implement create/get task (#222)
d5404a1 is described below
commit d5404a16cb5c5d598383491ef49c9f60d204a94e
Author: TX <li...@gmail.com>
AuthorDate: Wed Feb 12 18:03:57 2020 -0800
[management] Implement create/get task (#222)
---
services/management/enclave/src/fusion_data.rs | 5 -
services/management/enclave/src/lib.rs | 2 +
services/management/enclave/src/service.rs | 177 ++++++++++-
services/management/enclave/src/task.rs | 122 ++++++++
services/proto/proto_gen/templates/proto.j2 | 1 +
.../src/proto/teaclave_frontend_service.proto | 54 ++++
.../src/proto/teaclave_management_service.proto | 3 +
services/proto/src/teaclave_frontend_service.rs | 346 ++++++++++++++++++---
.../enclave/src/teaclave_management_service.rs | 80 +++++
9 files changed, 735 insertions(+), 55 deletions(-)
diff --git a/services/management/enclave/src/fusion_data.rs b/services/management/enclave/src/fusion_data.rs
index 2734878..ad1216f 100644
--- a/services/management/enclave/src/fusion_data.rs
+++ b/services/management/enclave/src/fusion_data.rs
@@ -20,7 +20,6 @@ use serde_json;
use std::prelude::v1::*;
use teaclave_types::TeaclaveFileCryptoInfo;
use url::Url;
-#[cfg(any(test_mode, feature = "enclave_unit_test"))]
use uuid::Uuid;
const FUSION_DATA_PREFIX: &str = "fusion-data-";
@@ -33,7 +32,6 @@ pub(crate) struct FusionData {
pub(crate) data_id: String,
}
-#[cfg(any(test_mode, feature = "enclave_unit_test"))]
fn gen_url_for_fusion_data(data_id: &str) -> Result<Url> {
let url = format!("fusion://path/{}?token=fusion_token", data_id);
info!("{}", url);
@@ -41,7 +39,6 @@ fn gen_url_for_fusion_data(data_id: &str) -> Result<Url> {
}
impl FusionData {
- #[cfg(any(test_mode, feature = "enclave_unit_test"))]
pub fn new(data_owner_id_list: Vec<String>) -> Result<Self> {
let data_id = format!("{}{}", FUSION_DATA_PREFIX, Uuid::new_v4().to_string());
let url = gen_url_for_fusion_data(&data_id)?;
@@ -59,12 +56,10 @@ impl FusionData {
serde_json::from_slice(&bytes).map_err(|_| anyhow!("failed to Deserialize"))
}
- #[cfg(any(test_mode, feature = "enclave_unit_test"))]
pub(crate) fn to_vec(&self) -> Result<Vec<u8>> {
serde_json::to_vec(&self).map_err(|_| anyhow!("failed to Serialize"))
}
- #[cfg(any(test_mode, feature = "enclave_unit_test"))]
pub(crate) fn get_key_vec(&self) -> Vec<u8> {
self.data_id.as_bytes().to_vec()
}
diff --git a/services/management/enclave/src/lib.rs b/services/management/enclave/src/lib.rs
index 10fb682..5851c39 100644
--- a/services/management/enclave/src/lib.rs
+++ b/services/management/enclave/src/lib.rs
@@ -43,6 +43,7 @@ mod file;
mod function;
mod fusion_data;
mod service;
+mod task;
fn start_service(args: &StartServiceInput) -> anyhow::Result<()> {
let listen_address = args.config.internal_endpoints.management.listen_address;
@@ -154,6 +155,7 @@ pub mod tests {
service::tests::handle_output_file,
service::tests::handle_fusion_data,
service::tests::handle_function,
+ service::tests::handle_task,
)
}
}
diff --git a/services/management/enclave/src/service.rs b/services/management/enclave/src/service.rs
index b934b21..da6bef1 100644
--- a/services/management/enclave/src/service.rs
+++ b/services/management/enclave/src/service.rs
@@ -1,12 +1,14 @@
use crate::file::{InputFile, OutputFile};
use crate::function::Function;
use crate::fusion_data::FusionData;
+use crate::task::Task;
use anyhow::{anyhow, Result};
use std::prelude::v1::*;
use std::sync::{Arc, SgxMutex as Mutex};
use teaclave_proto::teaclave_frontend_service::{
- GetFunctionRequest, GetFunctionResponse, GetFusionDataRequest, GetFusionDataResponse,
- GetOutputFileRequest, GetOutputFileResponse, RegisterFunctionRequest, RegisterFunctionResponse,
+ CreateTaskRequest, CreateTaskResponse, GetFunctionRequest, GetFunctionResponse,
+ GetFusionDataRequest, GetFusionDataResponse, GetOutputFileRequest, GetOutputFileResponse,
+ GetTaskRequest, GetTaskResponse, RegisterFunctionRequest, RegisterFunctionResponse,
RegisterInputFileRequest, RegisterInputFileResponse, RegisterOutputFileRequest,
RegisterOutputFileResponse,
};
@@ -28,6 +30,8 @@ enum TeaclaveManagementError {
StorageError,
#[error("permission denied")]
PermissionDenied,
+ #[error("bad task")]
+ BadTask,
}
impl From<TeaclaveManagementError> for TeaclaveServiceResponseError {
@@ -188,16 +192,7 @@ impl TeaclaveManagement for TeaclaveManagementService {
.get("id")
.ok_or_else(|| TeaclaveManagementError::InvalidRequest)?
.to_string();
- let function_id = request.message.function_id;
- if !Function::is_function_id(&function_id) {
- return Err(TeaclaveManagementError::PermissionDenied.into());
- }
- let key: &[u8] = function_id.as_bytes();
- let value = self
- .read_from_storage(key)
- .map_err(|_| TeaclaveManagementError::StorageError)?;
- let function =
- Function::from_slice(&value).map_err(|_| TeaclaveManagementError::DataError)?;
+ let function = self.get_function_from_storage(&request.message.function_id)?;
if !(function.is_public || function.owner == user_id) {
return Err(TeaclaveManagementError::PermissionDenied.into());
}
@@ -213,6 +208,76 @@ impl TeaclaveManagement for TeaclaveManagementService {
};
Ok(response)
}
+
+ fn create_task(
+ &self,
+ request: Request<CreateTaskRequest>,
+ ) -> TeaclaveServiceResponseResult<CreateTaskResponse> {
+ let user_id = request
+ .metadata
+ .get("id")
+ .ok_or_else(|| TeaclaveManagementError::InvalidRequest)?
+ .to_string();
+ let request = request.message;
+ let function = self.get_function_from_storage(&request.function_id)?;
+ let mut task = Task::new(
+ function,
+ user_id,
+ request.arg_list,
+ request.input_data_owner_list,
+ request.output_data_owner_list,
+ )
+ .map_err(|_| TeaclaveManagementError::BadTask)?;
+ // register fusion data
+ for (output_name, data_owner_id_list) in task.output_data_owner_list.iter() {
+ if data_owner_id_list.user_id_list.len() > 1 {
+ let user_id_list: Vec<String> =
+ data_owner_id_list.user_id_list.iter().cloned().collect();
+ let fusion_data = self.alloc_fusion_data(user_id_list)?;
+ task.output_map
+ .insert(output_name.to_string(), fusion_data.data_id);
+ }
+ }
+ let key = task.get_key_vec();
+ let value = task
+ .to_vec()
+ .map_err(|_| TeaclaveManagementError::DataError)?;
+ self.write_to_storage(&key, &value)
+ .map_err(|_| TeaclaveManagementError::StorageError)?;
+ Ok(CreateTaskResponse {
+ task_id: task.task_id,
+ })
+ }
+
+ fn get_task(
+ &self,
+ request: Request<GetTaskRequest>,
+ ) -> TeaclaveServiceResponseResult<GetTaskResponse> {
+ let user_id = request
+ .metadata
+ .get("id")
+ .ok_or_else(|| TeaclaveManagementError::InvalidRequest)?
+ .to_string();
+ let task: Task = self.get_task_from_storage(&request.message.task_id)?;
+ if !task.participants.contains(&user_id) {
+ return Err(TeaclaveManagementError::PermissionDenied.into());
+ }
+ let response = GetTaskResponse {
+ task_id: task.task_id,
+ creator: task.creator,
+ function_id: task.function_id,
+ function_owner: task.function_owner,
+ arg_list: task.arg_list,
+ input_data_owner_list: task.input_data_owner_list,
+ output_data_owner_list: task.output_data_owner_list,
+ participants: task.participants,
+ approved_user_list: task.approved_user_list,
+ input_map: task.input_map,
+ output_map: task.output_map,
+ status: task.status,
+ };
+ Ok(response)
+ }
}
impl TeaclaveManagementService {
@@ -234,6 +299,14 @@ impl TeaclaveManagementService {
name: "output".to_string(),
description: "output_desc".to_string(),
};
+ let function_input2 = FunctionInput {
+ name: "input2".to_string(),
+ description: "input_desc".to_string(),
+ };
+ let function_output2 = FunctionOutput {
+ name: "output2".to_string(),
+ description: "output_desc".to_string(),
+ };
let native_function = Function {
function_id: "native-mock-native-func".to_string(),
@@ -241,9 +314,9 @@ impl TeaclaveManagementService {
description: "mock-desc".to_string(),
payload: b"mock-payload".to_vec(),
is_public: true,
- arg_list: vec!["arg".to_string()],
- input_list: vec![function_input],
- output_list: vec![function_output],
+ arg_list: vec!["arg1".to_string(), "arg2".to_string()],
+ input_list: vec![function_input, function_input2],
+ output_list: vec![function_output, function_output2],
owner: "teaclave".to_string(),
is_native: true,
};
@@ -285,11 +358,52 @@ impl TeaclaveManagementService {
.get(get_request)?;
Ok(get_response.value)
}
+
+ fn get_function_from_storage(
+ &self,
+ function_id: &str,
+ ) -> TeaclaveServiceResponseResult<Function> {
+ if !Function::is_function_id(function_id) {
+ return Err(TeaclaveManagementError::PermissionDenied.into());
+ }
+ let function_key = function_id.as_bytes();
+ let function_bytes = self
+ .read_from_storage(function_key)
+ .map_err(|_| TeaclaveManagementError::StorageError)?;
+ Function::from_slice(&function_bytes).map_err(|_| TeaclaveManagementError::DataError.into())
+ }
+
+ fn alloc_fusion_data(
+ &self,
+ user_id_list: Vec<String>,
+ ) -> TeaclaveServiceResponseResult<FusionData> {
+ let fusion_data =
+ FusionData::new(user_id_list).map_err(|_| TeaclaveManagementError::DataError)?;
+ let data_key = fusion_data.get_key_vec();
+ let value = fusion_data
+ .to_vec()
+ .map_err(|_| TeaclaveManagementError::DataError)?;
+ self.write_to_storage(&data_key, &value)
+ .map_err(|_| TeaclaveManagementError::StorageError)?;
+ Ok(fusion_data)
+ }
+
+ fn get_task_from_storage(&self, task_id: &str) -> TeaclaveServiceResponseResult<Task> {
+ if !Task::is_task_id(task_id) {
+ return Err(TeaclaveManagementError::PermissionDenied.into());
+ }
+ let task_key = task_id.as_bytes();
+ let task_bytes = self
+ .read_from_storage(task_key)
+ .map_err(|_| TeaclaveManagementError::StorageError)?;
+ Task::from_slice(&task_bytes).map_err(|_| TeaclaveManagementError::DataError.into())
+ }
}
#[cfg(feature = "enclave_unit_test")]
pub mod tests {
use super::*;
+ use std::collections::HashMap;
use teaclave_proto::teaclave_frontend_service::{FunctionInput, FunctionOutput};
use teaclave_types::{TeaclaveFileCryptoInfo, TeaclaveFileRootKey128};
use url::Url;
@@ -367,4 +481,37 @@ pub mod tests {
let deserialized_data = Function::from_slice(&value).unwrap();
info!("data: {:?}", deserialized_data);
}
+
+ pub fn handle_task() {
+ let function_request = RegisterFunctionRequest {
+ name: "mock_function".to_string(),
+ description: "mock function".to_string(),
+ payload: b"python script".to_vec(),
+ is_public: false,
+ arg_list: vec!["arg".to_string()],
+ input_list: vec![],
+ output_list: vec![],
+ };
+ let function =
+ Function::new_from_register_request(function_request, "mock_user".to_string());
+ let mut arg_list = HashMap::new();
+ arg_list.insert("arg".to_string(), "data".to_string());
+
+ let task = Task::new(
+ function,
+ "mock_user".to_string(),
+ arg_list,
+ HashMap::new(),
+ HashMap::new(),
+ )
+ .unwrap();
+
+ let key = task.get_key_vec();
+ let key_str = std::str::from_utf8(&key).unwrap();
+ info!("key: {}", key_str);
+ assert!(Task::is_task_id(key_str));
+ let value = task.to_vec().unwrap();
+ let deserialized_data = Task::from_slice(&value).unwrap();
+ info!("data: {:?}", deserialized_data);
+ }
}
diff --git a/services/management/enclave/src/task.rs b/services/management/enclave/src/task.rs
new file mode 100644
index 0000000..3d11085
--- /dev/null
+++ b/services/management/enclave/src/task.rs
@@ -0,0 +1,122 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+use crate::function::Function;
+use anyhow::{anyhow, ensure, Result};
+use serde::{Deserialize, Serialize};
+use serde_json;
+use std::collections::HashMap;
+use std::collections::HashSet;
+use std::prelude::v1::*;
+use teaclave_proto::teaclave_frontend_service::{DataOwnerList, TaskStatus};
+use uuid::Uuid;
+
+const TASK_PREFIX: &str = "task-";
+
+#[derive(Debug, Deserialize, Serialize)]
+pub(crate) struct Task {
+ pub(crate) task_id: String,
+ pub(crate) creator: String,
+ pub(crate) function_id: String,
+ pub(crate) function_owner: String,
+ pub(crate) arg_list: HashMap<String, String>,
+ pub(crate) input_data_owner_list: HashMap<String, DataOwnerList>,
+ pub(crate) output_data_owner_list: HashMap<String, DataOwnerList>,
+ pub(crate) participants: HashSet<String>,
+ pub(crate) approved_user_list: HashSet<String>,
+ pub(crate) input_map: HashMap<String, String>,
+ pub(crate) output_map: HashMap<String, String>,
+ pub(crate) status: TaskStatus,
+}
+
+impl Task {
+ pub(crate) fn new(
+ function: Function,
+ creator: String,
+ arg_list: HashMap<String, String>,
+ input_data_owner_list: HashMap<String, DataOwnerList>,
+ output_data_owner_list: HashMap<String, DataOwnerList>,
+ ) -> Result<Self> {
+ let task_id = format!("{}{}", TASK_PREFIX, Uuid::new_v4().to_string());
+ let mut participants = HashSet::new();
+ if !function.is_public {
+ participants.insert(function.owner.clone());
+ }
+ participants.insert(creator.clone());
+ for (_, data_owner_list) in input_data_owner_list.iter() {
+ for user_id in data_owner_list.user_id_list.iter() {
+ participants.insert(user_id.clone());
+ }
+ }
+ for (_, data_owner_list) in output_data_owner_list.iter() {
+ for user_id in data_owner_list.user_id_list.iter() {
+ participants.insert(user_id.clone());
+ }
+ }
+ let task = Self {
+ task_id,
+ creator,
+ function_id: function.function_id,
+ function_owner: function.owner,
+ arg_list,
+ input_data_owner_list,
+ output_data_owner_list,
+ participants,
+ approved_user_list: HashSet::new(),
+ input_map: HashMap::new(),
+ output_map: HashMap::new(),
+ status: TaskStatus::Created,
+ };
+ // check arguments
+ let function_args: HashSet<String> = function.arg_list.into_iter().collect();
+ let provide_args: HashSet<String> = task.arg_list.keys().cloned().collect();
+ let diff: HashSet<_> = function_args.difference(&provide_args).collect();
+ ensure!(diff.is_empty(), "bad arguments");
+
+ // check input
+ let input_args: HashSet<String> = function.input_list.into_iter().map(|f| f.name).collect();
+ let provide_args: HashSet<String> = task.input_data_owner_list.keys().cloned().collect();
+ let diff: HashSet<_> = input_args.difference(&provide_args).collect();
+ ensure!(diff.is_empty(), "bad input");
+
+ // check output
+ let output_args: HashSet<String> =
+ function.output_list.into_iter().map(|f| f.name).collect();
+ let provide_args: HashSet<String> = task.output_data_owner_list.keys().cloned().collect();
+ let diff: HashSet<_> = output_args.difference(&provide_args).collect();
+ ensure!(diff.is_empty(), "bad output");
+
+ Ok(task)
+ }
+
+ pub(crate) fn from_slice(bytes: &[u8]) -> Result<Self> {
+ let ret: Task =
+ serde_json::from_slice(&bytes).map_err(|_| anyhow!("failed to Deserialize"))?;
+ Ok(ret)
+ }
+
+ pub(crate) fn to_vec(&self) -> Result<Vec<u8>> {
+ serde_json::to_vec(&self).map_err(|_| anyhow!("failed to Serialize"))
+ }
+
+ pub(crate) fn get_key_vec(&self) -> Vec<u8> {
+ self.task_id.as_bytes().to_vec()
+ }
+
+ pub(crate) fn is_task_id(id: &str) -> bool {
+ id.starts_with(TASK_PREFIX)
+ }
+}
diff --git a/services/proto/proto_gen/templates/proto.j2 b/services/proto/proto_gen/templates/proto.j2
index ee4edb1..70c16fd 100644
--- a/services/proto/proto_gen/templates/proto.j2
+++ b/services/proto/proto_gen/templates/proto.j2
@@ -6,6 +6,7 @@ pub enum {{ service.proto_name }}Request {
{%- endfor %}
}
+#[allow(clippy::large_enum_variant)]
#[derive(Clone, serde::Serialize, serde::Deserialize, Debug)]
#[serde(tag = "response", rename_all = "snake_case")]
pub enum {{ service.proto_name }}Response {
diff --git a/services/proto/src/proto/teaclave_frontend_service.proto b/services/proto/src/proto/teaclave_frontend_service.proto
index c41e585..82a37c4 100644
--- a/services/proto/src/proto/teaclave_frontend_service.proto
+++ b/services/proto/src/proto/teaclave_frontend_service.proto
@@ -79,6 +79,60 @@ message GetFunctionResponse {
repeated FunctionOutput output_list = 8;
}
+message DataOwnerList {
+ string data_name = 1;
+ repeated string user_id_list = 2;
+}
+
+message DataMap {
+ string data_name = 1;
+ string data_id = 2;
+}
+
+message Argument {
+ string arg_name = 1;
+ string arg_value = 2;
+}
+
+message CreateTaskRequest {
+ string function_id = 1;
+ repeated Argument arg_list = 2;
+ repeated DataOwnerList input_data_owner_list = 3;
+ repeated DataOwnerList output_data_owner_list = 4;
+}
+
+message CreateTaskResponse {
+ string task_id = 1;
+}
+
+message GetTaskRequest {
+ string task_id = 1;
+}
+
+enum TaskStatus {
+ Created = 0;
+ Ready = 1;
+ Approved = 2;
+ Running = 3;
+ Failed = 4;
+ Finished = 5;
+}
+
+message GetTaskResponse {
+ string task_id = 1;
+ string creator = 2;
+ string function_id = 3;
+ string function_owner = 4;
+ repeated Argument arg_list = 5;
+ repeated DataOwnerList input_data_owner_list = 6;
+ repeated DataOwnerList output_data_owner_list = 7;
+ repeated string participants = 8;
+ repeated string approved_user_list = 9;
+ repeated DataMap input_map = 10;
+ repeated DataMap output_map = 11;
+ TaskStatus status = 12;
+}
+
service TeaclaveFrontend {
rpc RegisterInputFile (RegisterInputFileRequest) returns (RegisterInputFileResponse);
rpc RegisterOutputFile (RegisterOutputFileRequest) returns (RegisterOutputFileResponse);
diff --git a/services/proto/src/proto/teaclave_management_service.proto b/services/proto/src/proto/teaclave_management_service.proto
index c0b154f..1264c91 100644
--- a/services/proto/src/proto/teaclave_management_service.proto
+++ b/services/proto/src/proto/teaclave_management_service.proto
@@ -11,4 +11,7 @@ service TeaclaveManagement {
rpc GetFusionData (teaclave_frontend_service_proto.GetFusionDataRequest) returns (teaclave_frontend_service_proto.GetFusionDataResponse);
rpc RegisterFunction (teaclave_frontend_service_proto.RegisterFunctionRequest) returns (teaclave_frontend_service_proto.RegisterFunctionResponse);
rpc GetFunction (teaclave_frontend_service_proto.GetFunctionRequest) returns (teaclave_frontend_service_proto.GetFunctionResponse);
+ rpc CreateTask (teaclave_frontend_service_proto.CreateTaskRequest) returns (teaclave_frontend_service_proto.CreateTaskResponse);
+ rpc GetTask (teaclave_frontend_service_proto.GetTaskRequest) returns (teaclave_frontend_service_proto.GetTaskResponse);
+
}
\ No newline at end of file
diff --git a/services/proto/src/teaclave_frontend_service.rs b/services/proto/src/teaclave_frontend_service.rs
index e7f9865..762f24f 100644
--- a/services/proto/src/teaclave_frontend_service.rs
+++ b/services/proto/src/teaclave_frontend_service.rs
@@ -2,7 +2,10 @@ use crate::teaclave_frontend_service_proto as proto;
use anyhow::anyhow;
use anyhow::{Error, Result};
use core::convert::TryInto;
+use core::iter::FromIterator;
use serde::{Deserialize, Serialize};
+use std::collections::{HashMap, HashSet};
+use std::prelude::v1::*;
use teaclave_types::TeaclaveFileCryptoInfo;
use url::Url;
@@ -14,13 +17,13 @@ pub use proto::TeaclaveFrontendResponse;
#[derive(Debug)]
pub struct RegisterInputFileRequest {
pub url: Url,
- pub hash: std::string::String,
+ pub hash: String,
pub crypto_info: TeaclaveFileCryptoInfo,
}
#[derive(Debug)]
pub struct RegisterInputFileResponse {
- pub data_id: std::string::String,
+ pub data_id: String,
}
#[derive(Debug)]
@@ -31,75 +34,191 @@ pub struct RegisterOutputFileRequest {
#[derive(Debug)]
pub struct RegisterOutputFileResponse {
- pub data_id: std::string::String,
+ pub data_id: String,
}
#[derive(Debug)]
pub struct GetOutputFileRequest {
- pub data_id: std::string::String,
+ pub data_id: String,
}
#[derive(Debug)]
pub struct GetOutputFileResponse {
- pub hash: std::string::String,
+ pub hash: String,
}
#[derive(Debug)]
pub struct GetFusionDataRequest {
- pub data_id: std::string::String,
+ pub data_id: String,
}
#[derive(Debug)]
pub struct GetFusionDataResponse {
- pub hash: std::string::String,
- pub data_owner_id_list: std::vec::Vec<std::string::String>,
+ pub hash: String,
+ pub data_owner_id_list: Vec<String>,
}
#[derive(Debug, Deserialize, Serialize)]
pub struct FunctionInput {
- pub name: std::string::String,
- pub description: std::string::String,
+ pub name: String,
+ pub description: String,
}
#[derive(Debug, Deserialize, Serialize)]
pub struct FunctionOutput {
- pub name: std::string::String,
- pub description: std::string::String,
+ pub name: String,
+ pub description: String,
}
#[derive(Debug)]
pub struct RegisterFunctionRequest {
- pub name: std::string::String,
- pub description: std::string::String,
- pub payload: std::vec::Vec<u8>,
+ pub name: String,
+ pub description: String,
+ pub payload: Vec<u8>,
pub is_public: bool,
- pub arg_list: std::vec::Vec<std::string::String>,
- pub input_list: std::vec::Vec<FunctionInput>,
- pub output_list: std::vec::Vec<FunctionOutput>,
+ pub arg_list: Vec<String>,
+ pub input_list: Vec<FunctionInput>,
+ pub output_list: Vec<FunctionOutput>,
}
#[derive(Debug)]
pub struct RegisterFunctionResponse {
- pub function_id: std::string::String,
+ pub function_id: String,
}
#[derive(Debug)]
pub struct GetFunctionRequest {
- pub function_id: std::string::String,
+ pub function_id: String,
}
#[derive(Debug)]
pub struct GetFunctionResponse {
- pub name: std::string::String,
- pub description: std::string::String,
- pub owner: std::string::String,
- pub payload: std::vec::Vec<u8>,
+ pub name: String,
+ pub description: String,
+ pub owner: String,
+ pub payload: Vec<u8>,
pub is_public: bool,
- pub arg_list: std::vec::Vec<std::string::String>,
- pub input_list: std::vec::Vec<FunctionInput>,
- pub output_list: std::vec::Vec<FunctionOutput>,
+ pub arg_list: Vec<String>,
+ pub input_list: Vec<FunctionInput>,
+ pub output_list: Vec<FunctionOutput>,
}
+#[derive(Debug, Deserialize, Serialize, Clone)]
+pub struct DataOwnerList {
+ pub user_id_list: HashSet<String>,
+}
+
+#[derive(Debug)]
+pub struct CreateTaskRequest {
+ pub function_id: String,
+ pub arg_list: HashMap<String, String>,
+ pub input_data_owner_list: HashMap<String, DataOwnerList>,
+ pub output_data_owner_list: HashMap<String, DataOwnerList>,
+}
+
+#[derive(Debug)]
+pub struct CreateTaskResponse {
+ pub task_id: String,
+}
+
+#[derive(Debug)]
+pub struct DataMap {
+ pub data_name: String,
+ pub data_id: String,
+}
+
+#[derive(Debug, Deserialize, Serialize)]
+pub enum TaskStatus {
+ Created,
+ Ready,
+ Approved,
+ Running,
+ Failed,
+ Finished,
+}
+
+#[derive(Debug)]
+pub struct GetTaskRequest {
+ pub task_id: String,
+}
+
+#[derive(Debug)]
+pub struct GetTaskResponse {
+ pub task_id: String,
+ pub creator: String,
+ pub function_id: String,
+ pub function_owner: String,
+ pub arg_list: HashMap<String, String>,
+ pub input_data_owner_list: HashMap<String, DataOwnerList>,
+ pub output_data_owner_list: HashMap<String, DataOwnerList>,
+ pub participants: HashSet<String>,
+ pub approved_user_list: HashSet<String>,
+ pub input_map: HashMap<String, String>,
+ pub output_map: HashMap<String, String>,
+ pub status: TaskStatus,
+}
+
+fn arg_list_from_proto(vector: Vec<proto::Argument>) -> Result<HashMap<String, String>> {
+ let mut ret = HashMap::with_capacity(vector.len());
+ for item in vector.into_iter() {
+ ret.insert(item.arg_name, item.arg_value);
+ }
+ Ok(ret)
+}
+
+fn arg_list_to_proto(map: HashMap<String, String>) -> Vec<proto::Argument> {
+ let mut ret = Vec::with_capacity(map.len());
+ for (arg_name, arg_value) in map.into_iter() {
+ let argument = proto::Argument {
+ arg_name,
+ arg_value,
+ };
+ ret.push(argument);
+ }
+ ret
+}
+
+fn data_map_to_proto(map: HashMap<String, String>) -> Vec<proto::DataMap> {
+ let mut ret = Vec::with_capacity(map.len());
+ for (data_name, data_id) in map.into_iter() {
+ let data_map = proto::DataMap { data_name, data_id };
+ ret.push(data_map);
+ }
+ ret
+}
+
+fn data_map_from_proto(vector: Vec<proto::DataMap>) -> Result<HashMap<String, String>> {
+ let mut ret = HashMap::with_capacity(vector.len());
+ for item in vector.into_iter() {
+ ret.insert(item.data_name, item.data_id);
+ }
+ Ok(ret)
+}
+
+fn data_owner_list_from_proto(
+ vector: Vec<proto::DataOwnerList>,
+) -> Result<HashMap<String, DataOwnerList>> {
+ let mut ret = HashMap::with_capacity(vector.len());
+ for item in vector.into_iter() {
+ let data_owner_list = DataOwnerList {
+ user_id_list: HashSet::from_iter(item.user_id_list.into_iter()),
+ };
+ ret.insert(item.data_name, data_owner_list);
+ }
+ Ok(ret)
+}
+
+fn data_owner_list_to_proto(map: HashMap<String, DataOwnerList>) -> Vec<proto::DataOwnerList> {
+ let mut ret = Vec::with_capacity(map.len());
+ for (data_name, data_owner_list) in map.into_iter() {
+ let data_owner_list = proto::DataOwnerList {
+ data_name,
+ user_id_list: data_owner_list.user_id_list.into_iter().collect(),
+ };
+ ret.push(data_owner_list);
+ }
+ ret
+}
impl std::convert::TryFrom<proto::RegisterInputFileRequest> for RegisterInputFileRequest {
type Error = Error;
@@ -312,12 +431,12 @@ impl std::convert::TryFrom<proto::RegisterFunctionRequest> for RegisterFunctionR
type Error = Error;
fn try_from(proto: proto::RegisterFunctionRequest) -> Result<Self> {
- let input_list: Result<std::vec::Vec<FunctionInput>> = proto
+ let input_list: Result<Vec<FunctionInput>> = proto
.input_list
.into_iter()
.map(FunctionInput::try_from)
.collect();
- let output_list: Result<std::vec::Vec<FunctionOutput>> = proto
+ let output_list: Result<Vec<FunctionOutput>> = proto
.output_list
.into_iter()
.map(FunctionOutput::try_from)
@@ -338,12 +457,12 @@ impl std::convert::TryFrom<proto::RegisterFunctionRequest> for RegisterFunctionR
impl From<RegisterFunctionRequest> for proto::RegisterFunctionRequest {
fn from(request: RegisterFunctionRequest) -> Self {
- let input_list: std::vec::Vec<proto::FunctionInput> = request
+ let input_list: Vec<proto::FunctionInput> = request
.input_list
.into_iter()
.map(proto::FunctionInput::from)
.collect();
- let output_list: std::vec::Vec<proto::FunctionOutput> = request
+ let output_list: Vec<proto::FunctionOutput> = request
.output_list
.into_iter()
.map(proto::FunctionOutput::from)
@@ -405,12 +524,12 @@ impl std::convert::TryFrom<proto::GetFunctionResponse> for GetFunctionResponse {
type Error = Error;
fn try_from(proto: proto::GetFunctionResponse) -> Result<Self> {
- let input_list: Result<std::vec::Vec<FunctionInput>> = proto
+ let input_list: Result<Vec<FunctionInput>> = proto
.input_list
.into_iter()
.map(FunctionInput::try_from)
.collect();
- let output_list: Result<std::vec::Vec<FunctionOutput>> = proto
+ let output_list: Result<Vec<FunctionOutput>> = proto
.output_list
.into_iter()
.map(FunctionOutput::try_from)
@@ -433,12 +552,12 @@ impl std::convert::TryFrom<proto::GetFunctionResponse> for GetFunctionResponse {
impl From<GetFunctionResponse> for proto::GetFunctionResponse {
fn from(response: GetFunctionResponse) -> Self {
- let input_list: std::vec::Vec<proto::FunctionInput> = response
+ let input_list: Vec<proto::FunctionInput> = response
.input_list
.into_iter()
.map(proto::FunctionInput::from)
.collect();
- let output_list: std::vec::Vec<proto::FunctionOutput> = response
+ let output_list: Vec<proto::FunctionOutput> = response
.output_list
.into_iter()
.map(proto::FunctionOutput::from)
@@ -456,3 +575,160 @@ impl From<GetFunctionResponse> for proto::GetFunctionResponse {
}
}
}
+
+impl std::convert::TryFrom<proto::CreateTaskRequest> for CreateTaskRequest {
+ type Error = Error;
+
+ fn try_from(proto: proto::CreateTaskRequest) -> Result<Self> {
+ let arg_list = arg_list_from_proto(proto.arg_list)?;
+ let input_data_owner_list = data_owner_list_from_proto(proto.input_data_owner_list)?;
+ let output_data_owner_list = data_owner_list_from_proto(proto.output_data_owner_list)?;
+ let ret = Self {
+ function_id: proto.function_id,
+ arg_list,
+ input_data_owner_list,
+ output_data_owner_list,
+ };
+ Ok(ret)
+ }
+}
+
+impl From<CreateTaskRequest> for proto::CreateTaskRequest {
+ fn from(request: CreateTaskRequest) -> Self {
+ let arg_list = arg_list_to_proto(request.arg_list);
+ let input_data_owner_list = data_owner_list_to_proto(request.input_data_owner_list);
+ let output_data_owner_list = data_owner_list_to_proto(request.output_data_owner_list);
+
+ Self {
+ function_id: request.function_id,
+ arg_list,
+ input_data_owner_list,
+ output_data_owner_list,
+ }
+ }
+}
+
+impl std::convert::TryFrom<proto::CreateTaskResponse> for CreateTaskResponse {
+ type Error = Error;
+
+ fn try_from(proto: proto::CreateTaskResponse) -> Result<Self> {
+ let ret = Self {
+ task_id: proto.task_id,
+ };
+
+ Ok(ret)
+ }
+}
+
+impl From<CreateTaskResponse> for proto::CreateTaskResponse {
+ fn from(response: CreateTaskResponse) -> Self {
+ Self {
+ task_id: response.task_id,
+ }
+ }
+}
+
+impl std::convert::TryFrom<i32> for TaskStatus {
+ type Error = Error;
+
+ fn try_from(status: i32) -> Result<Self> {
+ let ret = match proto::TaskStatus::from_i32(status) {
+ Some(proto::TaskStatus::Created) => TaskStatus::Created,
+ Some(proto::TaskStatus::Ready) => TaskStatus::Ready,
+ Some(proto::TaskStatus::Approved) => TaskStatus::Approved,
+ Some(proto::TaskStatus::Running) => TaskStatus::Running,
+ Some(proto::TaskStatus::Failed) => TaskStatus::Failed,
+ Some(proto::TaskStatus::Finished) => TaskStatus::Finished,
+ None => return Err(anyhow!("invalid task status")),
+ };
+ Ok(ret)
+ }
+}
+
+impl From<TaskStatus> for i32 {
+ fn from(status: TaskStatus) -> i32 {
+ match status {
+ TaskStatus::Created => proto::TaskStatus::Created as i32,
+ TaskStatus::Ready => proto::TaskStatus::Ready as i32,
+ TaskStatus::Approved => proto::TaskStatus::Approved as i32,
+ TaskStatus::Running => proto::TaskStatus::Running as i32,
+ TaskStatus::Failed => proto::TaskStatus::Failed as i32,
+ TaskStatus::Finished => proto::TaskStatus::Finished as i32,
+ }
+ }
+}
+
+impl std::convert::TryFrom<proto::GetTaskRequest> for GetTaskRequest {
+ type Error = Error;
+
+ fn try_from(proto: proto::GetTaskRequest) -> Result<Self> {
+ let ret = Self {
+ task_id: proto.task_id,
+ };
+
+ Ok(ret)
+ }
+}
+
+impl From<GetTaskRequest> for proto::GetTaskRequest {
+ fn from(request: GetTaskRequest) -> Self {
+ Self {
+ task_id: request.task_id,
+ }
+ }
+}
+
+impl std::convert::TryFrom<proto::GetTaskResponse> for GetTaskResponse {
+ type Error = Error;
+
+ fn try_from(proto: proto::GetTaskResponse) -> Result<Self> {
+ let arg_list = arg_list_from_proto(proto.arg_list)?;
+ let input_data_owner_list = data_owner_list_from_proto(proto.input_data_owner_list)?;
+ let output_data_owner_list = data_owner_list_from_proto(proto.output_data_owner_list)?;
+ let input_map = data_map_from_proto(proto.input_map)?;
+ let output_map = data_map_from_proto(proto.output_map)?;
+ let status = TaskStatus::try_from(proto.status)?;
+
+ let ret = Self {
+ task_id: proto.task_id,
+ creator: proto.creator,
+ function_id: proto.function_id,
+ function_owner: proto.function_owner,
+ arg_list,
+ input_data_owner_list,
+ output_data_owner_list,
+ participants: proto.participants.into_iter().collect(),
+ approved_user_list: proto.approved_user_list.into_iter().collect(),
+ input_map,
+ output_map,
+ status,
+ };
+
+ Ok(ret)
+ }
+}
+
+impl From<GetTaskResponse> for proto::GetTaskResponse {
+ fn from(response: GetTaskResponse) -> Self {
+ let arg_list = arg_list_to_proto(response.arg_list);
+ let input_data_owner_list = data_owner_list_to_proto(response.input_data_owner_list);
+ let output_data_owner_list = data_owner_list_to_proto(response.output_data_owner_list);
+ let input_map = data_map_to_proto(response.input_map);
+ let output_map = data_map_to_proto(response.output_map);
+ let status = i32::from(response.status);
+ Self {
+ task_id: response.task_id,
+ creator: response.creator,
+ function_id: response.function_id,
+ function_owner: response.function_owner,
+ arg_list,
+ input_data_owner_list,
+ output_data_owner_list,
+ participants: response.participants.into_iter().collect(),
+ approved_user_list: response.approved_user_list.into_iter().collect(),
+ input_map,
+ output_map,
+ status,
+ }
+ }
+}
diff --git a/tests/functional_tests/enclave/src/teaclave_management_service.rs b/tests/functional_tests/enclave/src/teaclave_management_service.rs
index 21e25d3..949b705 100644
--- a/tests/functional_tests/enclave/src/teaclave_management_service.rs
+++ b/tests/functional_tests/enclave/src/teaclave_management_service.rs
@@ -17,9 +17,11 @@ pub fn run_tests() -> bool {
test_register_input_file,
test_register_output_file,
test_register_function,
+ test_create_task,
test_get_output_file,
test_get_fusion_data,
test_get_function,
+ test_get_task,
)
}
@@ -200,3 +202,81 @@ fn test_get_function() {
assert!(response.is_ok());
info!("{:?}", response.unwrap());
}
+
+fn get_correct_create_task() -> CreateTaskRequest {
+ let mut arg_list = HashMap::new();
+ arg_list.insert("arg1".to_string(), "data1".to_string());
+ arg_list.insert("arg2".to_string(), "data2".to_string());
+ let data_owner_id_list = DataOwnerList {
+ user_id_list: vec!["mock_user1".to_string()].into_iter().collect(),
+ };
+ let data_owner_id_list2 = DataOwnerList {
+ user_id_list: vec!["mock_user2".to_string(), "mock_user3".to_string()]
+ .into_iter()
+ .collect(),
+ };
+ let mut input_data_owner_list = HashMap::new();
+ input_data_owner_list.insert("input".to_string(), data_owner_id_list.clone());
+ input_data_owner_list.insert("input2".to_string(), data_owner_id_list2.clone());
+ let mut output_data_owner_list = HashMap::new();
+ output_data_owner_list.insert("output".to_string(), data_owner_id_list);
+ output_data_owner_list.insert("output2".to_string(), data_owner_id_list2);
+
+ CreateTaskRequest {
+ function_id: "native-mock-native-func".to_string(),
+ arg_list,
+ input_data_owner_list,
+ output_data_owner_list,
+ }
+}
+
+fn test_create_task() {
+ let request = CreateTaskRequest {
+ function_id: "invalid_function".to_string(),
+ arg_list: HashMap::new(),
+ input_data_owner_list: HashMap::new(),
+ output_data_owner_list: HashMap::new(),
+ };
+ let mut client = get_client("mock_user");
+ let response = client.create_task(request);
+ assert!(response.is_err());
+
+ let request = get_correct_create_task();
+ let response = client.create_task(request);
+ assert!(response.is_ok());
+
+ let mut request = get_correct_create_task();
+ request.arg_list.remove("arg1");
+ let response = client.create_task(request);
+ assert!(response.is_err());
+
+ let mut request = get_correct_create_task();
+ request.input_data_owner_list.remove("input");
+ let response = client.create_task(request);
+ assert!(response.is_err());
+
+ let mut request = get_correct_create_task();
+ request.output_data_owner_list.remove("output");
+ let response = client.create_task(request);
+ assert!(response.is_err());
+}
+
+fn test_get_task() {
+ let mut client = get_client("mock_user");
+ let request = get_correct_create_task();
+ let response = client.create_task(request);
+ assert!(response.is_ok());
+ let task_id = response.unwrap().task_id;
+
+ let request = GetTaskRequest { task_id };
+ let response = client.get_task(request);
+ assert!(response.is_ok());
+ let response = response.unwrap();
+ info!("{:?}", response);
+ let participants = vec!["mock_user1", "mock_user3", "mock_user2", "mock_user"];
+ for name in participants {
+ assert!(response.participants.contains(&name.to_string()));
+ }
+ assert!(response.participants.len() == 4);
+ assert!(response.output_map.len() == 1);
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@teaclave.apache.org
For additional commands, e-mail: commits-help@teaclave.apache.org