From ee7159a40655ff0b2ab71f3aab67433c45c47863 Mon Sep 17 00:00:00 2001 From: Sammy Kerata Oina <44265300+SammyOina@users.noreply.github.com> Date: Mon, 8 Jan 2024 19:51:46 +0300 Subject: [PATCH] COCOS-35 - Refactor RunRequest to use structured Computation (#38) * Refactor RunRequest to use structured Computation The protobuf and associated service implementations for the RunRequest message were refactored to replace the raw Computation byte slice with a structured ComputationReq object. This allows clearer and more type-safe manipulation of computation requests. The grpc, http, and agent service layers were updated to build and parse ComputationReq accordingly. The ComputationReq structure includes details like IDs, names, time stamps, and metadata, forming a well-defined contract for computation tasks. This change aligns with efforts to standardize request formats and improve clarity in inter-service communication. It impacts all systems interfacing with the RunRequest service and thus requires coordinated updates to the entire stack. Signed-off-by: SammyOina * Initialize metadata maps and handle nil values Improved the robustness of metadata handling in gRPC endpoints and SDK by initializing metadata maps and explicitly checking for nil values before converting them. This ensures that both the agent's gRPC endpoint and the SDK properly handle cases where metadata fields may be uninitialized or contain nil values, preventing potential null pointer exceptions. Signed-off-by: SammyOina * Refactor computation request handling Refactored the endpoint to construct Computation object from gRPC request, incorporating structpb for metadata handling and timestamppb for StartTime and EndTime fields. The management service and API requests are also updated to align with these changes, improving type safety and ensuring data is correctly marshalled when making service calls. Resolves data marshalling issues for computation requests. Signed-off-by: SammyOina * use singular Signed-off-by: SammyOina * remove unuse fields Signed-off-by: SammyOina * remove unused fields Signed-off-by: SammyOina --------- Signed-off-by: SammyOina --- agent/agent.pb.go | 412 +++++++++++++++++++++++++++++------ agent/agent.proto | 22 +- agent/api/grpc/endpoint.go | 24 +- agent/api/grpc/requests.go | 9 +- agent/computations.go | 8 - agent/tracing/tracing.go | 3 - api/manager.yml | 87 +++----- manager/api/grpc/requests.go | 11 +- manager/api/http/endpoint.go | 17 +- manager/api/http/requests.go | 14 +- manager/api/logging.go | 2 +- manager/api/metrics.go | 2 +- manager/manager.pb.go | 340 ++++++++++++++++++++++++++--- manager/manager.proto | 22 +- manager/service.go | 23 +- manager/tracing/tracing.go | 2 +- pkg/sdk/agent.go | 22 +- 17 files changed, 810 insertions(+), 210 deletions(-) diff --git a/agent/agent.pb.go b/agent/agent.pb.go index 382f63d6..6ce76ce8 100644 --- a/agent/agent.pb.go +++ b/agent/agent.pb.go @@ -28,7 +28,7 @@ type RunRequest struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - Computation []byte `protobuf:"bytes,1,opt,name=computation,proto3" json:"computation,omitempty"` + Computation *ComputationReq `protobuf:"bytes,1,opt,name=computation,proto3" json:"computation,omitempty"` } func (x *RunRequest) Reset() { @@ -63,7 +63,7 @@ func (*RunRequest) Descriptor() ([]byte, []int) { return file_agent_agent_proto_rawDescGZIP(), []int{0} } -func (x *RunRequest) GetComputation() []byte { +func (x *RunRequest) GetComputation() *ComputationReq { if x != nil { return x.Computation } @@ -516,63 +516,293 @@ func (x *AttestationResponse) GetFile() []byte { return nil } +type ComputationReq struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"` + Description string `protobuf:"bytes,3,opt,name=description,proto3" json:"description,omitempty"` + Datasets []*DatasetReq `protobuf:"bytes,4,rep,name=datasets,proto3" json:"datasets,omitempty"` + Algorithms []*AlgorithmReq `protobuf:"bytes,5,rep,name=algorithms,proto3" json:"algorithms,omitempty"` + ResultConsumers []string `protobuf:"bytes,6,rep,name=result_consumers,json=resultConsumers,proto3" json:"result_consumers,omitempty"` + Timeout string `protobuf:"bytes,7,opt,name=timeout,proto3" json:"timeout,omitempty"` +} + +func (x *ComputationReq) Reset() { + *x = ComputationReq{} + if protoimpl.UnsafeEnabled { + mi := &file_agent_agent_proto_msgTypes[10] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *ComputationReq) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ComputationReq) ProtoMessage() {} + +func (x *ComputationReq) ProtoReflect() protoreflect.Message { + mi := &file_agent_agent_proto_msgTypes[10] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ComputationReq.ProtoReflect.Descriptor instead. +func (*ComputationReq) Descriptor() ([]byte, []int) { + return file_agent_agent_proto_rawDescGZIP(), []int{10} +} + +func (x *ComputationReq) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +func (x *ComputationReq) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +func (x *ComputationReq) GetDescription() string { + if x != nil { + return x.Description + } + return "" +} + +func (x *ComputationReq) GetDatasets() []*DatasetReq { + if x != nil { + return x.Datasets + } + return nil +} + +func (x *ComputationReq) GetAlgorithms() []*AlgorithmReq { + if x != nil { + return x.Algorithms + } + return nil +} + +func (x *ComputationReq) GetResultConsumers() []string { + if x != nil { + return x.ResultConsumers + } + return nil +} + +func (x *ComputationReq) GetTimeout() string { + if x != nil { + return x.Timeout + } + return "" +} + +type DatasetReq struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Provider string `protobuf:"bytes,1,opt,name=provider,proto3" json:"provider,omitempty"` + Id string `protobuf:"bytes,2,opt,name=id,proto3" json:"id,omitempty"` +} + +func (x *DatasetReq) Reset() { + *x = DatasetReq{} + if protoimpl.UnsafeEnabled { + mi := &file_agent_agent_proto_msgTypes[11] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *DatasetReq) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*DatasetReq) ProtoMessage() {} + +func (x *DatasetReq) ProtoReflect() protoreflect.Message { + mi := &file_agent_agent_proto_msgTypes[11] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use DatasetReq.ProtoReflect.Descriptor instead. +func (*DatasetReq) Descriptor() ([]byte, []int) { + return file_agent_agent_proto_rawDescGZIP(), []int{11} +} + +func (x *DatasetReq) GetProvider() string { + if x != nil { + return x.Provider + } + return "" +} + +func (x *DatasetReq) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +type AlgorithmReq struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Provider string `protobuf:"bytes,1,opt,name=provider,proto3" json:"provider,omitempty"` + Id string `protobuf:"bytes,2,opt,name=id,proto3" json:"id,omitempty"` +} + +func (x *AlgorithmReq) Reset() { + *x = AlgorithmReq{} + if protoimpl.UnsafeEnabled { + mi := &file_agent_agent_proto_msgTypes[12] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *AlgorithmReq) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*AlgorithmReq) ProtoMessage() {} + +func (x *AlgorithmReq) ProtoReflect() protoreflect.Message { + mi := &file_agent_agent_proto_msgTypes[12] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use AlgorithmReq.ProtoReflect.Descriptor instead. +func (*AlgorithmReq) Descriptor() ([]byte, []int) { + return file_agent_agent_proto_rawDescGZIP(), []int{12} +} + +func (x *AlgorithmReq) GetProvider() string { + if x != nil { + return x.Provider + } + return "" +} + +func (x *AlgorithmReq) GetId() string { + if x != nil { + return x.Id + } + return "" +} + var File_agent_agent_proto protoreflect.FileDescriptor var file_agent_agent_proto_rawDesc = []byte{ 0x0a, 0x11, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2f, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x70, 0x72, - 0x6f, 0x74, 0x6f, 0x12, 0x05, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x22, 0x2e, 0x0a, 0x0a, 0x52, 0x75, - 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x20, 0x0a, 0x0b, 0x63, 0x6f, 0x6d, 0x70, - 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0b, 0x63, - 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x2f, 0x0a, 0x0b, 0x52, 0x75, - 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x20, 0x0a, 0x0b, 0x43, 0x6f, 0x6d, - 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, - 0x43, 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x57, 0x0a, 0x0b, 0x41, - 0x6c, 0x67, 0x6f, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1c, 0x0a, 0x09, 0x61, 0x6c, - 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x09, 0x61, - 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x76, - 0x69, 0x64, 0x65, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x76, - 0x69, 0x64, 0x65, 0x72, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, - 0x52, 0x02, 0x69, 0x64, 0x22, 0x30, 0x0a, 0x0c, 0x41, 0x6c, 0x67, 0x6f, 0x52, 0x65, 0x73, 0x70, - 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x20, 0x0a, 0x0b, 0x61, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, - 0x6d, 0x49, 0x44, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x61, 0x6c, 0x67, 0x6f, 0x72, - 0x69, 0x74, 0x68, 0x6d, 0x49, 0x44, 0x22, 0x53, 0x0a, 0x0b, 0x44, 0x61, 0x74, 0x61, 0x52, 0x65, - 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x18, 0x0a, 0x07, 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, - 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x12, + 0x6f, 0x74, 0x6f, 0x12, 0x05, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x22, 0x45, 0x0a, 0x0a, 0x52, 0x75, + 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x37, 0x0a, 0x0b, 0x63, 0x6f, 0x6d, 0x70, + 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0b, 0x32, 0x15, 0x2e, + 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x43, 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, + 0x6e, 0x52, 0x65, 0x71, 0x52, 0x0b, 0x63, 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, + 0x6e, 0x22, 0x2f, 0x0a, 0x0b, 0x52, 0x75, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, + 0x12, 0x20, 0x0a, 0x0b, 0x43, 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, + 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x43, 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, + 0x6f, 0x6e, 0x22, 0x57, 0x0a, 0x0b, 0x41, 0x6c, 0x67, 0x6f, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, + 0x74, 0x12, 0x1c, 0x0a, 0x09, 0x61, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, 0x18, 0x01, + 0x20, 0x01, 0x28, 0x0c, 0x52, 0x09, 0x61, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, 0x65, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, 0x65, 0x72, 0x12, 0x0e, 0x0a, 0x02, 0x69, - 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x22, 0x2c, 0x0a, 0x0c, 0x44, - 0x61, 0x74, 0x61, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x1c, 0x0a, 0x09, 0x64, - 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x49, 0x44, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, - 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x49, 0x44, 0x22, 0x2b, 0x0a, 0x0d, 0x52, 0x65, 0x73, - 0x75, 0x6c, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x63, 0x6f, - 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x63, 0x6f, - 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x22, 0x24, 0x0a, 0x0e, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, - 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x66, 0x69, 0x6c, 0x65, - 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x66, 0x69, 0x6c, 0x65, 0x22, 0x14, 0x0a, 0x12, - 0x41, 0x74, 0x74, 0x65, 0x73, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, - 0x73, 0x74, 0x22, 0x29, 0x0a, 0x13, 0x41, 0x74, 0x74, 0x65, 0x73, 0x74, 0x61, 0x74, 0x69, 0x6f, - 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x66, 0x69, 0x6c, - 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x66, 0x69, 0x6c, 0x65, 0x32, 0xa5, 0x02, - 0x0a, 0x0c, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x2e, - 0x0a, 0x03, 0x52, 0x75, 0x6e, 0x12, 0x11, 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x52, 0x75, - 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x12, 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, - 0x2e, 0x52, 0x75, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x31, - 0x0a, 0x04, 0x41, 0x6c, 0x67, 0x6f, 0x12, 0x12, 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x41, - 0x6c, 0x67, 0x6f, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x13, 0x2e, 0x61, 0x67, 0x65, - 0x6e, 0x74, 0x2e, 0x41, 0x6c, 0x67, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, - 0x00, 0x12, 0x31, 0x0a, 0x04, 0x44, 0x61, 0x74, 0x61, 0x12, 0x12, 0x2e, 0x61, 0x67, 0x65, 0x6e, - 0x74, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x13, 0x2e, - 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, - 0x73, 0x65, 0x22, 0x00, 0x12, 0x37, 0x0a, 0x06, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, 0x14, - 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x52, 0x65, 0x71, - 0x75, 0x65, 0x73, 0x74, 0x1a, 0x15, 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x52, 0x65, 0x73, - 0x75, 0x6c, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x46, 0x0a, - 0x0b, 0x41, 0x74, 0x74, 0x65, 0x73, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x19, 0x2e, 0x61, - 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x41, 0x74, 0x74, 0x65, 0x73, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, - 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1a, 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, - 0x41, 0x74, 0x74, 0x65, 0x73, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, - 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x09, 0x5a, 0x07, 0x2e, 0x2f, 0x61, 0x67, 0x65, 0x6e, 0x74, - 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x22, 0x30, 0x0a, 0x0c, 0x41, + 0x6c, 0x67, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x20, 0x0a, 0x0b, 0x61, + 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, 0x49, 0x44, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, + 0x52, 0x0b, 0x61, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, 0x49, 0x44, 0x22, 0x53, 0x0a, + 0x0b, 0x44, 0x61, 0x74, 0x61, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, 0x18, 0x0a, 0x07, + 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x07, 0x64, + 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, + 0x65, 0x72, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, + 0x65, 0x72, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, + 0x69, 0x64, 0x22, 0x2c, 0x0a, 0x0c, 0x44, 0x61, 0x74, 0x61, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, + 0x73, 0x65, 0x12, 0x1c, 0x0a, 0x09, 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x49, 0x44, 0x18, + 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x09, 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x49, 0x44, + 0x22, 0x2b, 0x0a, 0x0d, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, + 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x18, 0x01, 0x20, + 0x01, 0x28, 0x09, 0x52, 0x08, 0x63, 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x22, 0x24, 0x0a, + 0x0e, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, + 0x12, 0x0a, 0x04, 0x66, 0x69, 0x6c, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x66, + 0x69, 0x6c, 0x65, 0x22, 0x14, 0x0a, 0x12, 0x41, 0x74, 0x74, 0x65, 0x73, 0x74, 0x61, 0x74, 0x69, + 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x22, 0x29, 0x0a, 0x13, 0x41, 0x74, 0x74, + 0x65, 0x73, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, + 0x12, 0x12, 0x0a, 0x04, 0x66, 0x69, 0x6c, 0x65, 0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, + 0x66, 0x69, 0x6c, 0x65, 0x22, 0xff, 0x01, 0x0a, 0x0e, 0x43, 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, + 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x71, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x01, 0x20, + 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, + 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x20, 0x0a, 0x0b, 0x64, + 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x03, 0x20, 0x01, 0x28, 0x09, + 0x52, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x2d, 0x0a, + 0x08, 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x73, 0x18, 0x04, 0x20, 0x03, 0x28, 0x0b, 0x32, + 0x11, 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x52, + 0x65, 0x71, 0x52, 0x08, 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x73, 0x12, 0x33, 0x0a, 0x0a, + 0x61, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, 0x73, 0x18, 0x05, 0x20, 0x03, 0x28, 0x0b, + 0x32, 0x13, 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x41, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, + 0x68, 0x6d, 0x52, 0x65, 0x71, 0x52, 0x0a, 0x61, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, + 0x73, 0x12, 0x29, 0x0a, 0x10, 0x72, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x5f, 0x63, 0x6f, 0x6e, 0x73, + 0x75, 0x6d, 0x65, 0x72, 0x73, 0x18, 0x06, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0f, 0x72, 0x65, 0x73, + 0x75, 0x6c, 0x74, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x73, 0x12, 0x18, 0x0a, 0x07, + 0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x18, 0x07, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x74, + 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x22, 0x38, 0x0a, 0x0a, 0x44, 0x61, 0x74, 0x61, 0x73, 0x65, + 0x74, 0x52, 0x65, 0x71, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, 0x65, 0x72, + 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, 0x65, 0x72, + 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, + 0x22, 0x3a, 0x0a, 0x0c, 0x41, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, 0x52, 0x65, 0x71, + 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, 0x65, 0x72, 0x18, 0x01, 0x20, 0x01, + 0x28, 0x09, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, 0x65, 0x72, 0x12, 0x0e, 0x0a, 0x02, + 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x32, 0xa5, 0x02, 0x0a, + 0x0c, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x2e, 0x0a, + 0x03, 0x52, 0x75, 0x6e, 0x12, 0x11, 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x52, 0x75, 0x6e, + 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x12, 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, + 0x52, 0x75, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x31, 0x0a, + 0x04, 0x41, 0x6c, 0x67, 0x6f, 0x12, 0x12, 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x41, 0x6c, + 0x67, 0x6f, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x13, 0x2e, 0x61, 0x67, 0x65, 0x6e, + 0x74, 0x2e, 0x41, 0x6c, 0x67, 0x6f, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, + 0x12, 0x31, 0x0a, 0x04, 0x44, 0x61, 0x74, 0x61, 0x12, 0x12, 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, + 0x2e, 0x44, 0x61, 0x74, 0x61, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x13, 0x2e, 0x61, + 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x44, 0x61, 0x74, 0x61, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, + 0x65, 0x22, 0x00, 0x12, 0x37, 0x0a, 0x06, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x12, 0x14, 0x2e, + 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x52, 0x65, 0x71, 0x75, + 0x65, 0x73, 0x74, 0x1a, 0x15, 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x52, 0x65, 0x73, 0x75, + 0x6c, 0x74, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x12, 0x46, 0x0a, 0x0b, + 0x41, 0x74, 0x74, 0x65, 0x73, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x19, 0x2e, 0x61, 0x67, + 0x65, 0x6e, 0x74, 0x2e, 0x41, 0x74, 0x74, 0x65, 0x73, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, + 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x1a, 0x2e, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x2e, 0x41, + 0x74, 0x74, 0x65, 0x73, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, + 0x73, 0x65, 0x22, 0x00, 0x42, 0x09, 0x5a, 0x07, 0x2e, 0x2f, 0x61, 0x67, 0x65, 0x6e, 0x74, 0x62, + 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, } var ( @@ -587,7 +817,7 @@ func file_agent_agent_proto_rawDescGZIP() []byte { return file_agent_agent_proto_rawDescData } -var file_agent_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 10) +var file_agent_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 13) var file_agent_agent_proto_goTypes = []interface{}{ (*RunRequest)(nil), // 0: agent.RunRequest (*RunResponse)(nil), // 1: agent.RunResponse @@ -599,23 +829,29 @@ var file_agent_agent_proto_goTypes = []interface{}{ (*ResultResponse)(nil), // 7: agent.ResultResponse (*AttestationRequest)(nil), // 8: agent.AttestationRequest (*AttestationResponse)(nil), // 9: agent.AttestationResponse + (*ComputationReq)(nil), // 10: agent.ComputationReq + (*DatasetReq)(nil), // 11: agent.DatasetReq + (*AlgorithmReq)(nil), // 12: agent.AlgorithmReq } var file_agent_agent_proto_depIdxs = []int32{ - 0, // 0: agent.AgentService.Run:input_type -> agent.RunRequest - 2, // 1: agent.AgentService.Algo:input_type -> agent.AlgoRequest - 4, // 2: agent.AgentService.Data:input_type -> agent.DataRequest - 6, // 3: agent.AgentService.Result:input_type -> agent.ResultRequest - 8, // 4: agent.AgentService.Attestation:input_type -> agent.AttestationRequest - 1, // 5: agent.AgentService.Run:output_type -> agent.RunResponse - 3, // 6: agent.AgentService.Algo:output_type -> agent.AlgoResponse - 5, // 7: agent.AgentService.Data:output_type -> agent.DataResponse - 7, // 8: agent.AgentService.Result:output_type -> agent.ResultResponse - 9, // 9: agent.AgentService.Attestation:output_type -> agent.AttestationResponse - 5, // [5:10] is the sub-list for method output_type - 0, // [0:5] is the sub-list for method input_type - 0, // [0:0] is the sub-list for extension type_name - 0, // [0:0] is the sub-list for extension extendee - 0, // [0:0] is the sub-list for field type_name + 10, // 0: agent.RunRequest.computation:type_name -> agent.ComputationReq + 11, // 1: agent.ComputationReq.datasets:type_name -> agent.DatasetReq + 12, // 2: agent.ComputationReq.algorithms:type_name -> agent.AlgorithmReq + 0, // 3: agent.AgentService.Run:input_type -> agent.RunRequest + 2, // 4: agent.AgentService.Algo:input_type -> agent.AlgoRequest + 4, // 5: agent.AgentService.Data:input_type -> agent.DataRequest + 6, // 6: agent.AgentService.Result:input_type -> agent.ResultRequest + 8, // 7: agent.AgentService.Attestation:input_type -> agent.AttestationRequest + 1, // 8: agent.AgentService.Run:output_type -> agent.RunResponse + 3, // 9: agent.AgentService.Algo:output_type -> agent.AlgoResponse + 5, // 10: agent.AgentService.Data:output_type -> agent.DataResponse + 7, // 11: agent.AgentService.Result:output_type -> agent.ResultResponse + 9, // 12: agent.AgentService.Attestation:output_type -> agent.AttestationResponse + 8, // [8:13] is the sub-list for method output_type + 3, // [3:8] is the sub-list for method input_type + 3, // [3:3] is the sub-list for extension type_name + 3, // [3:3] is the sub-list for extension extendee + 0, // [0:3] is the sub-list for field type_name } func init() { file_agent_agent_proto_init() } @@ -744,6 +980,42 @@ func file_agent_agent_proto_init() { return nil } } + file_agent_agent_proto_msgTypes[10].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*ComputationReq); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_agent_agent_proto_msgTypes[11].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*DatasetReq); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_agent_agent_proto_msgTypes[12].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*AlgorithmReq); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } } type x struct{} out := protoimpl.TypeBuilder{ @@ -751,7 +1023,7 @@ func file_agent_agent_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_agent_agent_proto_rawDesc, NumEnums: 0, - NumMessages: 10, + NumMessages: 13, NumExtensions: 0, NumServices: 1, }, diff --git a/agent/agent.proto b/agent/agent.proto index c8b5ff99..c6e175d7 100644 --- a/agent/agent.proto +++ b/agent/agent.proto @@ -15,7 +15,7 @@ service AgentService { rpc Attestation(AttestationRequest) returns (AttestationResponse) {} } -message RunRequest { bytes computation = 1; } +message RunRequest { ComputationReq computation = 1; } message RunResponse { string Computation = 1; } @@ -44,3 +44,23 @@ message ResultResponse { bytes file = 1; } message AttestationRequest { } message AttestationResponse { bytes file = 1; } + +message ComputationReq { + string id = 1; + string name = 2; + string description = 3; + repeated DatasetReq datasets = 4; + repeated AlgorithmReq algorithms = 5; + repeated string result_consumers = 6; + string timeout = 7; +} + +message DatasetReq { + string provider = 1; + string id = 2; +} + +message AlgorithmReq { + string provider = 1; + string id = 2; +} diff --git a/agent/api/grpc/endpoint.go b/agent/api/grpc/endpoint.go index 16bc1e23..d1f361a5 100644 --- a/agent/api/grpc/endpoint.go +++ b/agent/api/grpc/endpoint.go @@ -4,7 +4,7 @@ package grpc import ( "context" - "encoding/json" + "time" "github.com/go-kit/kit/endpoint" "github.com/ultravioletrs/cocos/agent" @@ -18,12 +18,26 @@ func runEndpoint(svc agent.Service) endpoint.Endpoint { return runRes{}, err } - var computation agent.Computation - err := json.Unmarshal(req.Computation, &computation) - if err != nil { - return nil, err + computation := agent.Computation{ + ID: req.Computation.Id, + Name: req.Computation.Name, + Description: req.Computation.Description, + ResultConsumers: req.Computation.ResultConsumers, } + for _, algo := range req.Computation.Algorithms { + computation.Algorithms = append(computation.Algorithms, agent.Algorithm{ID: algo.Id, Provider: algo.Provider}) + } + for _, data := range req.Computation.Datasets { + computation.Datasets = append(computation.Datasets, agent.Dataset{ID: data.Id, Provider: data.Provider}) + } + + timeout, err := time.ParseDuration(req.Computation.Timeout) + if err != nil { + return runRes{}, err + } + computation.Timeout.Duration = timeout + computationStr, err := svc.Run(ctx, computation) if err != nil { return runRes{}, err diff --git a/agent/api/grpc/requests.go b/agent/api/grpc/requests.go index 3a38e5e2..ecdb37f2 100644 --- a/agent/api/grpc/requests.go +++ b/agent/api/grpc/requests.go @@ -4,16 +4,15 @@ package grpc import ( "errors" + + "github.com/ultravioletrs/cocos/agent" ) type runReq struct { - Computation []byte `protobuf:"bytes,1,opt,name=algorithm,proto3" json:"algorithm,omitempty"` + Computation *agent.ComputationReq } -func (req runReq) validate() error { - if len(req.Computation) == 0 { - return errors.New("algorithm binary is required") - } +func (req *runReq) validate() error { return nil } diff --git a/agent/computations.go b/agent/computations.go index 71fc40cc..ebc6c159 100644 --- a/agent/computations.go +++ b/agent/computations.go @@ -19,15 +19,9 @@ type Computation struct { ID string `json:"id,omitempty"` Name string `json:"name,omitempty"` Description string `json:"description,omitempty"` - Status string `json:"status,omitempty"` - Owner string `json:"owner,omitempty"` - StartTime time.Time `json:"start_time,omitempty"` - EndTime time.Time `json:"end_time,omitempty"` Datasets Datasets `json:"datasets,omitempty"` Algorithms Algorithms `json:"algorithms,omitempty"` ResultConsumers []string `json:"result_consumers,omitempty"` - Ttl int32 `json:"ttl,omitempty"` - Metadata Metadata `json:"metadata,omitempty"` Timeout Duration `json:"timeout,omitempty"` } @@ -76,8 +70,6 @@ func (d *Duration) UnmarshalJSON(b []byte) error { } } -type Metadata map[string]interface{} - type Dataset struct { Dataset []byte `json:"-"` Provider string `json:"provider,omitempty"` diff --git a/agent/tracing/tracing.go b/agent/tracing/tracing.go index c4be50fe..1643ff73 100644 --- a/agent/tracing/tracing.go +++ b/agent/tracing/tracing.go @@ -27,9 +27,6 @@ func (tm *tracingMiddleware) Run(ctx context.Context, cmp agent.Computation) (st attribute.String("id", cmp.ID), attribute.String("name", cmp.Name), attribute.String("description", cmp.Description), - attribute.String("status", cmp.Status), - attribute.String("start_time", cmp.StartTime.String()), - attribute.String("end_time", cmp.EndTime.String()), attribute.StringSlice("result_consumers", cmp.ResultConsumers), attribute.Stringer("datasets", &cmp.Datasets), attribute.Stringer("algorithms", &cmp.Algorithms), diff --git a/api/manager.yml b/api/manager.yml index 8babdb48..496dd9ed 100644 --- a/api/manager.yml +++ b/api/manager.yml @@ -68,56 +68,6 @@ components: type: string description: Service build time. example: 1970-01-01_00:00:00 - ComputationReq: - type: object - properties: - computation: - type: object - properties: - id: - type: string - name: - type: string - description: - type: string - status: - type: string - owner: - type: string - start_time: - type: string - end_time: - type: string - datasets: - type: array - items: - type: object - properties: - provider: - type: string - id: - type: string - algorithms: - type: array - items: - type: object - properties: - provider: - type: string - id: - type: string - results_consumers: - type: array - items: - type: string - ttl: - type: integer - metadata: - type: object - example: { "domain": "example.com" } - timeout: - type: string - example: "2s" requestBodies: Run: required: true @@ -128,10 +78,39 @@ components: properties: computation: description: byte array of computation request as defined in ComputationReq - example: [123, 34, 105, 100, 34, 58, 34, 48, 48, 97, 99, 102, 101, 57, 100, 45, 53, 101, 49, 98, 45, 52, 97, 101, 99, 45, 56, 50, 99, 48, 45, 56, 48, 97, 98, 56, 101, 100, 53, 101, 99, 48, 98, 34, 44, 34, 110, 97, 109, 101, 34, 58, 34, 77, 97, 99, 104, 105, 110, 101, 32, 68, 105, 97, 103, 110, 111, 115, 116, 105, 99, 115, 32, 65, 110, 97, 108, 121, 115, 105, 115, 34, 44, 34, 100, 101, 115, 99, 114, 105, 112, 116, 105, 111, 110, 34, 58, 34, 80, 101, 114, 102, 111, 114, 109, 105, 110, 103, 32, 100, 105, 97, 103, 110, 111, 115, 116, 105, 99, 115, 32, 97, 110, 97, 108, 121, 115, 105, 115, 32, 111, 110, 32, 109, 97, 99, 104, 105, 110, 101, 32, 100, 97, 116, 97, 34, 44, 34, 115, 116, 97, 116, 117, 115, 34, 58, 34, 101, 120, 101, 99, 117, 116, 97, 98, 108, 101, 34, 44, 34, 111, 119, 110, 101, 114, 34, 58, 34, 77, 97, 99, 104, 105, 110, 101, 32, 73, 110, 100, 117, 115, 116, 114, 105, 101, 115, 32, 73, 110, 99, 46, 34, 44, 34, 115, 116, 97, 114, 116, 95, 116, 105, 109, 101, 34, 58, 34, 50, 48, 50, 51, 45, 48, 56, 45, 50, 49, 84, 49, 50, 58, 48, 50, 58, 51, 49, 46, 48, 48, 55, 53, 48, 53, 90, 34, 44, 34, 101, 110, 100, 95, 116, 105, 109, 101, 34, 58, 34, 48, 48, 48, 49, 45, 48, 49, 45, 48, 49, 84, 48, 48, 58, 48, 48, 58, 48, 48, 90, 34, 44, 34, 100, 97, 116, 97, 115, 101, 116, 115, 34, 58, 91, 34, 83, 101, 110, 115, 111, 114, 32, 68, 97, 116, 97, 32, 76, 111, 103, 115, 34, 44, 34, 77, 97, 99, 104, 105, 110, 101, 32, 72, 101, 97, 108, 116, 104, 32, 82, 101, 99, 111, 114, 100, 115, 34, 44, 34, 77, 97, 105, 110, 116, 101, 110, 97, 110, 99, 101, 32, 82, 101, 112, 111, 114, 116, 115, 34, 93, 44, 34, 97, 108, 103, 111, 114, 105, 116, 104, 109, 115, 34, 58, 91, 34, 83, 117, 112, 112, 111, 114, 116, 32, 86, 101, 99, 116, 111, 114, 32, 77, 97, 99, 104, 105, 110, 101, 115, 34, 44, 34, 75, 45, 78, 101, 97, 114, 101, 115, 116, 32, 78, 101, 105, 103, 104, 98, 111, 114, 115, 34, 44, 34, 72, 105, 101, 114, 97, 114, 99, 104, 105, 99, 97, 108, 32, 67, 108, 117, 115, 116, 101, 114, 105, 110, 103, 34, 93, 44, 34, 100, 97, 116, 97, 115, 101, 116, 95, 112, 114, 111, 118, 105, 100, 101, 114, 115, 34, 58, 91, 34, 83, 101, 110, 115, 111, 114, 84, 101, 99, 104, 32, 83, 111, 108, 117, 116, 105, 111, 110, 115, 34, 44, 34, 77, 97, 99, 104, 105, 110, 101, 114, 121, 32, 68, 97, 116, 97, 32, 83, 121, 115, 116, 101, 109, 115, 34, 93, 44, 34, 97, 108, 103, 111, 114, 105, 116, 104, 109, 95, 112, 114, 111, 118, 105, 100, 101, 114, 115, 34, 58, 91, 34, 65, 108, 103, 111, 65, 73, 32, 82, 101, 115, 101, 97, 114, 99, 104, 32, 76, 97, 98, 115, 34, 44, 34, 84, 101, 99, 104, 66, 111, 116, 115, 32, 73, 110, 110, 111, 118, 97, 116, 105, 111, 110, 115, 34, 93, 44, 34, 114, 101, 115, 117, 108, 116, 95, 99, 111, 110, 115, 117, 109, 101, 114, 115, 34, 58, 91, 34, 77, 97, 99, 104, 105, 110, 101, 32, 77, 97, 105, 110, 116, 101, 110, 97, 110, 99, 101, 32, 68, 101, 112, 97, 114, 116, 109, 101, 110, 116, 34, 44, 34, 80, 114, 101, 100, 105, 99, 116, 105, 118, 101, 32, 65, 110, 97, 108, 121, 116, 105, 99, 115, 32, 84, 101, 97, 109, 34, 44, 34, 73, 110, 100, 117, 115, 116, 114, 105, 97, 108, 32, 65, 117, 116, 111, 109, 97, 116, 105, 111, 110, 32, 68, 105, 118, 105, 115, 105, 111, 110, 34, 93, 44, 34, 116, 116, 108, 34, 58, 52, 56, 44, 34, 109, 101, 116, 97, 100, 97, 116, 97, 34, 58, 123, 34, 97, 110, 97, 108, 121, 115, 105, 115, 95, 112, 117, 114, 112, 111, 115, 101, 34, 58, 34, 79, 112, 116, 105, 109, 105, 122, 101, 32, 109, 97, 99, 104, 105, 110, 101, 32, 112, 101, 114, 102, 111, 114, 109, 97, 110, 99, 101, 32, 97, 110, 100, 32, 112, 114, 101, 118, 101, 110, 116, 32, 100, 111, 119, 110, 116, 105, 109, 101, 34, 44, 34, 100, 97, 116, 97, 95, 102, 114, 101, 113, 117, 101, 110, 99, 121, 34, 58, 34, 72, 111, 117, 114, 108, 121, 34, 44, 34, 105, 110, 100, 117, 115, 116, 114, 121, 34, 58, 34, 77, 97, 110, 117, 102, 97, 99, 116, 117, 114, 105, 110, 103, 34, 44, 34, 109, 97, 99, 104, 105, 110, 101, 95, 116, 121, 112, 101, 34, 58, 34, 65, 117, 116, 111, 109, 97, 116, 101, 100, 32, 65, 115, 115, 101, 109, 98, 108, 121, 32, 76, 105, 110, 101, 34, 125, 125] - type: array - items: - type: integer + type: object + properties: + id: + type: string + name: + type: string + description: + type: string + datasets: + type: array + items: + type: object + properties: + provider: + type: string + id: + type: string + algorithms: + type: array + items: + type: object + properties: + provider: + type: string + id: + type: string + results_consumers: + type: array + items: + type: string + timeout: + type: string + example: "2s" ca_certs: description: agent grpc ca_certs type: string diff --git a/manager/api/grpc/requests.go b/manager/api/grpc/requests.go index b8cf760f..06a45f4e 100644 --- a/manager/api/grpc/requests.go +++ b/manager/api/grpc/requests.go @@ -9,15 +9,12 @@ import ( ) type runReq struct { - Computation []byte `json:"computation,omitempty"` - ClientTLS bool `json:"client_tls,omitempty"` - CACerts string `json:"ca_certs,omitempty"` - Timeout time.Duration `json:"timeout,omitempty"` + Computation *manager.Computation `json:"computation"` + ClientTLS bool `json:"client_tls,omitempty"` + CACerts string `json:"ca_certs,omitempty"` + Timeout time.Duration `json:"timeout,omitempty"` } func (req runReq) validate() error { - if len(req.Computation) == 0 { - return manager.ErrMalformedEntity - } return nil } diff --git a/manager/api/http/endpoint.go b/manager/api/http/endpoint.go index 8ea03086..921d3d75 100644 --- a/manager/api/http/endpoint.go +++ b/manager/api/http/endpoint.go @@ -27,8 +27,23 @@ func runEndpoint(svc manager.Service) endpoint.Endpoint { if agentConf.Timeout == 0 { agentConf.Timeout = 60 * time.Second } + + computation := manager.Computation{ + Id: req.Computation.ID, + Name: req.Computation.Name, + Description: req.Computation.Description, + ResultConsumers: req.Computation.ResultConsumers, + Timeout: req.Computation.Timeout.String(), + } + for _, algo := range req.Computation.Algorithms { + computation.Algorithms = append(computation.Algorithms, &manager.Algorithm{Id: algo.ID, Provider: algo.Provider}) + } + for _, data := range req.Computation.Datasets { + computation.Datasets = append(computation.Datasets, &manager.Dataset{Id: data.ID, Provider: data.Provider}) + } + // Call the Run method on the service - runID, err := svc.Run(ctx, req.Computation, agentConf) + runID, err := svc.Run(ctx, &computation, agentConf) if err != nil { return nil, err } diff --git a/manager/api/http/requests.go b/manager/api/http/requests.go index 33daecba..062f69a6 100644 --- a/manager/api/http/requests.go +++ b/manager/api/http/requests.go @@ -4,7 +4,6 @@ package http import ( "github.com/ultravioletrs/cocos/agent" - "github.com/ultravioletrs/cocos/manager" ) var _ apiReq = (*runReq)(nil) @@ -14,15 +13,12 @@ type apiReq interface { } type runReq struct { - Computation []byte `json:"computation,omitempty"` - ClientTLS bool `json:"client_tls,omitempty"` - CACerts string `json:"ca_certs,omitempty"` - Timeout agent.Duration `json:"timeout,omitempty"` + Computation agent.Computation `json:"computation"` + ClientTLS bool `json:"client_tls,omitempty"` + CACerts string `json:"ca_certs,omitempty"` + Timeout agent.Duration `json:"timeout,omitempty"` } -func (req runReq) validate() error { - if len(req.Computation) == 0 { - return manager.ErrMalformedEntity - } +func (req *runReq) validate() error { return nil } diff --git a/manager/api/logging.go b/manager/api/logging.go index ac9b21dc..eb21698c 100644 --- a/manager/api/logging.go +++ b/manager/api/logging.go @@ -28,7 +28,7 @@ func LoggingMiddleware(svc manager.Service, logger mglog.Logger) manager.Service return &loggingMiddleware{logger, svc} } -func (lm *loggingMiddleware) Run(ctx context.Context, computation []byte, agentConfig grpc.Config) (id string, err error) { +func (lm *loggingMiddleware) Run(ctx context.Context, computation *manager.Computation, agentConfig grpc.Config) (id string, err error) { defer func(begin time.Time) { message := fmt.Sprintf("Method Run for computation took %s to complete", time.Since(begin)) if err != nil { diff --git a/manager/api/metrics.go b/manager/api/metrics.go index add63205..896cc0a5 100644 --- a/manager/api/metrics.go +++ b/manager/api/metrics.go @@ -33,7 +33,7 @@ func MetricsMiddleware(svc manager.Service, counter metrics.Counter, latency met } } -func (ms *metricsMiddleware) Run(ctx context.Context, computation []byte, agentConfig grpc.Config) (string, error) { +func (ms *metricsMiddleware) Run(ctx context.Context, computation *manager.Computation, agentConfig grpc.Config) (string, error) { defer func(begin time.Time) { ms.counter.With("method", "Run").Add(1) ms.latency.With("method", "Run").Observe(time.Since(begin).Seconds()) diff --git a/manager/manager.pb.go b/manager/manager.pb.go index 881d8503..072bfd9e 100644 --- a/manager/manager.pb.go +++ b/manager/manager.pb.go @@ -28,10 +28,10 @@ type RunRequest struct { sizeCache protoimpl.SizeCache unknownFields protoimpl.UnknownFields - Computation []byte `protobuf:"bytes,1,opt,name=computation,proto3" json:"computation,omitempty"` - CaCerts string `protobuf:"bytes,2,opt,name=ca_certs,json=caCerts,proto3" json:"ca_certs,omitempty"` - ClientTls bool `protobuf:"varint,3,opt,name=client_tls,json=clientTls,proto3" json:"client_tls,omitempty"` - Timeout string `protobuf:"bytes,4,opt,name=timeout,proto3" json:"timeout,omitempty"` + Computation *Computation `protobuf:"bytes,1,opt,name=computation,proto3" json:"computation,omitempty"` + CaCerts string `protobuf:"bytes,2,opt,name=ca_certs,json=caCerts,proto3" json:"ca_certs,omitempty"` + ClientTls bool `protobuf:"varint,3,opt,name=client_tls,json=clientTls,proto3" json:"client_tls,omitempty"` + Timeout string `protobuf:"bytes,4,opt,name=timeout,proto3" json:"timeout,omitempty"` } func (x *RunRequest) Reset() { @@ -66,7 +66,7 @@ func (*RunRequest) Descriptor() ([]byte, []int) { return file_manager_manager_proto_rawDescGZIP(), []int{0} } -func (x *RunRequest) GetComputation() []byte { +func (x *RunRequest) GetComputation() *Computation { if x != nil { return x.Computation } @@ -94,6 +94,211 @@ func (x *RunRequest) GetTimeout() string { return "" } +type Computation struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Id string `protobuf:"bytes,1,opt,name=id,proto3" json:"id,omitempty"` + Name string `protobuf:"bytes,2,opt,name=name,proto3" json:"name,omitempty"` + Description string `protobuf:"bytes,3,opt,name=description,proto3" json:"description,omitempty"` + Datasets []*Dataset `protobuf:"bytes,4,rep,name=datasets,proto3" json:"datasets,omitempty"` + Algorithms []*Algorithm `protobuf:"bytes,5,rep,name=algorithms,proto3" json:"algorithms,omitempty"` + ResultConsumers []string `protobuf:"bytes,6,rep,name=result_consumers,json=resultConsumers,proto3" json:"result_consumers,omitempty"` + Timeout string `protobuf:"bytes,7,opt,name=timeout,proto3" json:"timeout,omitempty"` +} + +func (x *Computation) Reset() { + *x = Computation{} + if protoimpl.UnsafeEnabled { + mi := &file_manager_manager_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *Computation) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Computation) ProtoMessage() {} + +func (x *Computation) ProtoReflect() protoreflect.Message { + mi := &file_manager_manager_proto_msgTypes[1] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Computation.ProtoReflect.Descriptor instead. +func (*Computation) Descriptor() ([]byte, []int) { + return file_manager_manager_proto_rawDescGZIP(), []int{1} +} + +func (x *Computation) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +func (x *Computation) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +func (x *Computation) GetDescription() string { + if x != nil { + return x.Description + } + return "" +} + +func (x *Computation) GetDatasets() []*Dataset { + if x != nil { + return x.Datasets + } + return nil +} + +func (x *Computation) GetAlgorithms() []*Algorithm { + if x != nil { + return x.Algorithms + } + return nil +} + +func (x *Computation) GetResultConsumers() []string { + if x != nil { + return x.ResultConsumers + } + return nil +} + +func (x *Computation) GetTimeout() string { + if x != nil { + return x.Timeout + } + return "" +} + +type Dataset struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Provider string `protobuf:"bytes,1,opt,name=provider,proto3" json:"provider,omitempty"` + Id string `protobuf:"bytes,2,opt,name=id,proto3" json:"id,omitempty"` +} + +func (x *Dataset) Reset() { + *x = Dataset{} + if protoimpl.UnsafeEnabled { + mi := &file_manager_manager_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *Dataset) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Dataset) ProtoMessage() {} + +func (x *Dataset) ProtoReflect() protoreflect.Message { + mi := &file_manager_manager_proto_msgTypes[2] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Dataset.ProtoReflect.Descriptor instead. +func (*Dataset) Descriptor() ([]byte, []int) { + return file_manager_manager_proto_rawDescGZIP(), []int{2} +} + +func (x *Dataset) GetProvider() string { + if x != nil { + return x.Provider + } + return "" +} + +func (x *Dataset) GetId() string { + if x != nil { + return x.Id + } + return "" +} + +type Algorithm struct { + state protoimpl.MessageState + sizeCache protoimpl.SizeCache + unknownFields protoimpl.UnknownFields + + Provider string `protobuf:"bytes,1,opt,name=provider,proto3" json:"provider,omitempty"` + Id string `protobuf:"bytes,2,opt,name=id,proto3" json:"id,omitempty"` +} + +func (x *Algorithm) Reset() { + *x = Algorithm{} + if protoimpl.UnsafeEnabled { + mi := &file_manager_manager_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) + } +} + +func (x *Algorithm) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Algorithm) ProtoMessage() {} + +func (x *Algorithm) ProtoReflect() protoreflect.Message { + mi := &file_manager_manager_proto_msgTypes[3] + if protoimpl.UnsafeEnabled && x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Algorithm.ProtoReflect.Descriptor instead. +func (*Algorithm) Descriptor() ([]byte, []int) { + return file_manager_manager_proto_rawDescGZIP(), []int{3} +} + +func (x *Algorithm) GetProvider() string { + if x != nil { + return x.Provider + } + return "" +} + +func (x *Algorithm) GetId() string { + if x != nil { + return x.Id + } + return "" +} + type RunResponse struct { state protoimpl.MessageState sizeCache protoimpl.SizeCache @@ -105,7 +310,7 @@ type RunResponse struct { func (x *RunResponse) Reset() { *x = RunResponse{} if protoimpl.UnsafeEnabled { - mi := &file_manager_manager_proto_msgTypes[1] + mi := &file_manager_manager_proto_msgTypes[4] ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ms.StoreMessageInfo(mi) } @@ -118,7 +323,7 @@ func (x *RunResponse) String() string { func (*RunResponse) ProtoMessage() {} func (x *RunResponse) ProtoReflect() protoreflect.Message { - mi := &file_manager_manager_proto_msgTypes[1] + mi := &file_manager_manager_proto_msgTypes[4] if protoimpl.UnsafeEnabled && x != nil { ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) if ms.LoadMessageInfo() == nil { @@ -131,7 +336,7 @@ func (x *RunResponse) ProtoReflect() protoreflect.Message { // Deprecated: Use RunResponse.ProtoReflect.Descriptor instead. func (*RunResponse) Descriptor() ([]byte, []int) { - return file_manager_manager_proto_rawDescGZIP(), []int{1} + return file_manager_manager_proto_rawDescGZIP(), []int{4} } func (x *RunResponse) GetID() string { @@ -146,22 +351,47 @@ var File_manager_manager_proto protoreflect.FileDescriptor var file_manager_manager_proto_rawDesc = []byte{ 0x0a, 0x15, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2f, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x07, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, - 0x22, 0x82, 0x01, 0x0a, 0x0a, 0x52, 0x75, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, - 0x20, 0x0a, 0x0b, 0x63, 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, - 0x20, 0x01, 0x28, 0x0c, 0x52, 0x0b, 0x63, 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, - 0x6e, 0x12, 0x19, 0x0a, 0x08, 0x63, 0x61, 0x5f, 0x63, 0x65, 0x72, 0x74, 0x73, 0x18, 0x02, 0x20, - 0x01, 0x28, 0x09, 0x52, 0x07, 0x63, 0x61, 0x43, 0x65, 0x72, 0x74, 0x73, 0x12, 0x1d, 0x0a, 0x0a, - 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x74, 0x6c, 0x73, 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, - 0x52, 0x09, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x54, 0x6c, 0x73, 0x12, 0x18, 0x0a, 0x07, 0x74, - 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x18, 0x04, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x74, 0x69, - 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x22, 0x1d, 0x0a, 0x0b, 0x52, 0x75, 0x6e, 0x52, 0x65, 0x73, 0x70, - 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x0e, 0x0a, 0x02, 0x49, 0x44, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, - 0x52, 0x02, 0x49, 0x44, 0x32, 0x44, 0x0a, 0x0e, 0x4d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x53, - 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x32, 0x0a, 0x03, 0x52, 0x75, 0x6e, 0x12, 0x13, 0x2e, - 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x52, 0x75, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, - 0x73, 0x74, 0x1a, 0x14, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x52, 0x75, 0x6e, - 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x0b, 0x5a, 0x09, 0x2e, 0x2f, - 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x33, + 0x22, 0x98, 0x01, 0x0a, 0x0a, 0x52, 0x75, 0x6e, 0x52, 0x65, 0x71, 0x75, 0x65, 0x73, 0x74, 0x12, + 0x36, 0x0a, 0x0b, 0x63, 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x01, + 0x20, 0x01, 0x28, 0x0b, 0x32, 0x14, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x43, + 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x52, 0x0b, 0x63, 0x6f, 0x6d, 0x70, + 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x19, 0x0a, 0x08, 0x63, 0x61, 0x5f, 0x63, 0x65, + 0x72, 0x74, 0x73, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x07, 0x63, 0x61, 0x43, 0x65, 0x72, + 0x74, 0x73, 0x12, 0x1d, 0x0a, 0x0a, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x5f, 0x74, 0x6c, 0x73, + 0x18, 0x03, 0x20, 0x01, 0x28, 0x08, 0x52, 0x09, 0x63, 0x6c, 0x69, 0x65, 0x6e, 0x74, 0x54, 0x6c, + 0x73, 0x12, 0x18, 0x0a, 0x07, 0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x18, 0x04, 0x20, 0x01, + 0x28, 0x09, 0x52, 0x07, 0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x22, 0xfa, 0x01, 0x0a, 0x0b, + 0x43, 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x12, 0x0e, 0x0a, 0x02, 0x69, + 0x64, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x12, 0x12, 0x0a, 0x04, 0x6e, + 0x61, 0x6d, 0x65, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, + 0x20, 0x0a, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, 0x6e, 0x18, 0x03, + 0x20, 0x01, 0x28, 0x09, 0x52, 0x0b, 0x64, 0x65, 0x73, 0x63, 0x72, 0x69, 0x70, 0x74, 0x69, 0x6f, + 0x6e, 0x12, 0x2c, 0x0a, 0x08, 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x73, 0x18, 0x04, 0x20, + 0x03, 0x28, 0x0b, 0x32, 0x10, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x44, 0x61, + 0x74, 0x61, 0x73, 0x65, 0x74, 0x52, 0x08, 0x64, 0x61, 0x74, 0x61, 0x73, 0x65, 0x74, 0x73, 0x12, + 0x32, 0x0a, 0x0a, 0x61, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, 0x73, 0x18, 0x05, 0x20, + 0x03, 0x28, 0x0b, 0x32, 0x12, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x41, 0x6c, + 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, 0x52, 0x0a, 0x61, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, + 0x68, 0x6d, 0x73, 0x12, 0x29, 0x0a, 0x10, 0x72, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x5f, 0x63, 0x6f, + 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x73, 0x18, 0x06, 0x20, 0x03, 0x28, 0x09, 0x52, 0x0f, 0x72, + 0x65, 0x73, 0x75, 0x6c, 0x74, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x73, 0x12, 0x18, + 0x0a, 0x07, 0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x18, 0x07, 0x20, 0x01, 0x28, 0x09, 0x52, + 0x07, 0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x22, 0x35, 0x0a, 0x07, 0x44, 0x61, 0x74, 0x61, + 0x73, 0x65, 0x74, 0x12, 0x1a, 0x0a, 0x08, 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, 0x65, 0x72, 0x18, + 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, 0x65, 0x72, 0x12, + 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x02, 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x22, + 0x37, 0x0a, 0x09, 0x41, 0x6c, 0x67, 0x6f, 0x72, 0x69, 0x74, 0x68, 0x6d, 0x12, 0x1a, 0x0a, 0x08, + 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, 0x65, 0x72, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x08, + 0x70, 0x72, 0x6f, 0x76, 0x69, 0x64, 0x65, 0x72, 0x12, 0x0e, 0x0a, 0x02, 0x69, 0x64, 0x18, 0x02, + 0x20, 0x01, 0x28, 0x09, 0x52, 0x02, 0x69, 0x64, 0x22, 0x1d, 0x0a, 0x0b, 0x52, 0x75, 0x6e, 0x52, + 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x12, 0x0e, 0x0a, 0x02, 0x49, 0x44, 0x18, 0x01, 0x20, + 0x01, 0x28, 0x09, 0x52, 0x02, 0x49, 0x44, 0x32, 0x44, 0x0a, 0x0e, 0x4d, 0x61, 0x6e, 0x61, 0x67, + 0x65, 0x72, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 0x12, 0x32, 0x0a, 0x03, 0x52, 0x75, 0x6e, + 0x12, 0x13, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, 0x52, 0x75, 0x6e, 0x52, 0x65, + 0x71, 0x75, 0x65, 0x73, 0x74, 0x1a, 0x14, 0x2e, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x2e, + 0x52, 0x75, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 0x22, 0x00, 0x42, 0x0b, 0x5a, + 0x09, 0x2e, 0x2f, 0x6d, 0x61, 0x6e, 0x61, 0x67, 0x65, 0x72, 0x62, 0x06, 0x70, 0x72, 0x6f, 0x74, + 0x6f, 0x33, } var ( @@ -176,19 +406,25 @@ func file_manager_manager_proto_rawDescGZIP() []byte { return file_manager_manager_proto_rawDescData } -var file_manager_manager_proto_msgTypes = make([]protoimpl.MessageInfo, 2) +var file_manager_manager_proto_msgTypes = make([]protoimpl.MessageInfo, 5) var file_manager_manager_proto_goTypes = []interface{}{ (*RunRequest)(nil), // 0: manager.RunRequest - (*RunResponse)(nil), // 1: manager.RunResponse + (*Computation)(nil), // 1: manager.Computation + (*Dataset)(nil), // 2: manager.Dataset + (*Algorithm)(nil), // 3: manager.Algorithm + (*RunResponse)(nil), // 4: manager.RunResponse } var file_manager_manager_proto_depIdxs = []int32{ - 0, // 0: manager.ManagerService.Run:input_type -> manager.RunRequest - 1, // 1: manager.ManagerService.Run:output_type -> manager.RunResponse - 1, // [1:2] is the sub-list for method output_type - 0, // [0:1] is the sub-list for method input_type - 0, // [0:0] is the sub-list for extension type_name - 0, // [0:0] is the sub-list for extension extendee - 0, // [0:0] is the sub-list for field type_name + 1, // 0: manager.RunRequest.computation:type_name -> manager.Computation + 2, // 1: manager.Computation.datasets:type_name -> manager.Dataset + 3, // 2: manager.Computation.algorithms:type_name -> manager.Algorithm + 0, // 3: manager.ManagerService.Run:input_type -> manager.RunRequest + 4, // 4: manager.ManagerService.Run:output_type -> manager.RunResponse + 4, // [4:5] is the sub-list for method output_type + 3, // [3:4] is the sub-list for method input_type + 3, // [3:3] is the sub-list for extension type_name + 3, // [3:3] is the sub-list for extension extendee + 0, // [0:3] is the sub-list for field type_name } func init() { file_manager_manager_proto_init() } @@ -210,6 +446,42 @@ func file_manager_manager_proto_init() { } } file_manager_manager_proto_msgTypes[1].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*Computation); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_manager_manager_proto_msgTypes[2].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*Dataset); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_manager_manager_proto_msgTypes[3].Exporter = func(v interface{}, i int) interface{} { + switch v := v.(*Algorithm); i { + case 0: + return &v.state + case 1: + return &v.sizeCache + case 2: + return &v.unknownFields + default: + return nil + } + } + file_manager_manager_proto_msgTypes[4].Exporter = func(v interface{}, i int) interface{} { switch v := v.(*RunResponse); i { case 0: return &v.state @@ -228,7 +500,7 @@ func file_manager_manager_proto_init() { GoPackagePath: reflect.TypeOf(x{}).PkgPath(), RawDescriptor: file_manager_manager_proto_rawDesc, NumEnums: 0, - NumMessages: 2, + NumMessages: 5, NumExtensions: 0, NumServices: 1, }, diff --git a/manager/manager.proto b/manager/manager.proto index 084aaa2f..8b95ced7 100644 --- a/manager/manager.proto +++ b/manager/manager.proto @@ -12,10 +12,30 @@ service ManagerService { } message RunRequest { - bytes computation = 1; + Computation computation = 1; string ca_certs = 2; bool client_tls = 3; string timeout = 4; } +message Computation { + string id = 1; + string name = 2; + string description = 3; + repeated Dataset datasets = 4; + repeated Algorithm algorithms = 5; + repeated string result_consumers = 6; + string timeout = 7; +} + +message Dataset { + string provider = 1; + string id = 2; +} + +message Algorithm { + string provider = 1; + string id = 2; +} + message RunResponse { string ID = 1; } diff --git a/manager/service.go b/manager/service.go index deae6abf..c6e71171 100644 --- a/manager/service.go +++ b/manager/service.go @@ -30,7 +30,7 @@ var ( // Service specifies an API that must be fulfilled by the domain service // implementation, and all of its decorators (e.g. logging & metrics). type Service interface { - Run(ctx context.Context, computation []byte, agentConfig grpc.Config) (string, error) + Run(ctx context.Context, computation *Computation, agentConfig grpc.Config) (string, error) } type managerService struct { @@ -46,7 +46,7 @@ func New(qemuCfg qemu.Config) Service { } } -func (ms *managerService) Run(ctx context.Context, computation []byte, agentConfig grpc.Config) (string, error) { +func (ms *managerService) Run(ctx context.Context, computation *Computation, agentConfig grpc.Config) (string, error) { _, err := qemu.CreateVM(ctx, ms.qemuCfg) if err != nil { return "", err @@ -58,6 +58,21 @@ func (ms *managerService) Run(ctx context.Context, computation []byte, agentConf ms.qemuCfg.NetDevConfig.HostFwd3++ }() + runReq := &agent.ComputationReq{ + Id: computation.Id, + Name: computation.Name, + Description: computation.Description, + ResultConsumers: computation.ResultConsumers, + Timeout: computation.Timeout, + } + + for _, algo := range computation.Algorithms { + runReq.Algorithms = append(runReq.Algorithms, &agent.AlgorithmReq{Id: algo.Id, Provider: algo.Provider}) + } + for _, data := range computation.Datasets { + runReq.Datasets = append(runReq.Datasets, &agent.DatasetReq{Id: data.Id, Provider: data.Provider}) + } + var res *agent.RunResponse agentConfig.URL = fmt.Sprintf("localhost:%d", ms.qemuCfg.HostFwd3) @@ -68,7 +83,9 @@ func (ms *managerService) Run(ctx context.Context, computation []byte, agentConf return err } defer agentGRPCClient.Close() - res, err = agentClient.Run(ctx, &agent.RunRequest{Computation: computation}) + res, err = agentClient.Run(ctx, &agent.RunRequest{ + Computation: runReq, + }) return err }, backoff.NewExponentialBackOff()) diff --git a/manager/tracing/tracing.go b/manager/tracing/tracing.go index 767cad2c..b27cea5a 100644 --- a/manager/tracing/tracing.go +++ b/manager/tracing/tracing.go @@ -22,7 +22,7 @@ func New(svc manager.Service, tracer trace.Tracer) manager.Service { return &tracingMiddleware{tracer, svc} } -func (tm *tracingMiddleware) Run(ctx context.Context, computation []byte, agentConfig grpc.Config) (string, error) { +func (tm *tracingMiddleware) Run(ctx context.Context, computation *manager.Computation, agentConfig grpc.Config) (string, error) { ctx, span := tm.tracer.Start(ctx, "run") defer span.End() diff --git a/pkg/sdk/agent.go b/pkg/sdk/agent.go index 7cc4fe66..ca504d19 100644 --- a/pkg/sdk/agent.go +++ b/pkg/sdk/agent.go @@ -4,7 +4,6 @@ package sdk import ( "context" - "encoding/json" mglog "github.com/absmach/magistrala/logger" "github.com/ultravioletrs/cocos/agent" @@ -25,14 +24,25 @@ func NewAgentSDK(log mglog.Logger, agentClient agent.AgentServiceClient) *agentS } func (sdk *agentSDK) Run(ctx context.Context, computation agent.Computation) (string, error) { - computationBytes, err := json.Marshal(computation) - if err != nil { - sdk.logger.Error("Failed to marshal computation") - return "", err + var datasets []*agent.DatasetReq + for _, data := range computation.Datasets { + datasets = append(datasets, &agent.DatasetReq{Id: data.ID, Provider: data.Provider}) + } + var algos []*agent.AlgorithmReq + for _, algo := range computation.Algorithms { + algos = append(algos, &agent.AlgorithmReq{Id: algo.ID, Provider: algo.Provider}) } request := &agent.RunRequest{ - Computation: computationBytes, + Computation: &agent.ComputationReq{ + Id: computation.ID, + Name: computation.Name, + Description: computation.Description, + Datasets: datasets, + Algorithms: algos, + ResultConsumers: computation.ResultConsumers, + Timeout: computation.Timeout.String(), + }, } response, err := sdk.client.Run(ctx, request) if err != nil {