opensource/cocos
1
0
Fork 0
mirror of https://github.com/ultravioletrs/cocos.git synced 2026-08-07 07:14:50 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

NOISSUE - Remove HTTP server from agent and update agent service (#15)

Browse Source 
* Update agent.pb.go to include a new field "Provider" in the AlgoRequest struct.

Signed-off-by: SammyOina <sammyoina@gmail.com>

* fix build

Signed-off-by: SammyOina <sammyoina@gmail.com>

* Fix undeclared consumer error in agent service

The commit fixes an error in the agent service where an undeclared consumer was requesting results in the computation manifest. The error is resolved by adding an error message for the undeclared consumer case.

Signed-off-by: SammyOina <sammyoina@gmail.com>

---------

Signed-off-by: SammyOina <sammyoina@gmail.com>
This commit is contained in:
Sammy Kerata Oina
2023-11-10 15:52:25 +03:00
committed by GitHub
parent da590bfb63
commit 8c7d4a76ec
17 changed files with 218 additions and 302 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/agent.pb.go
+70 -40
View File
@@ -4,7 +4,7 @@
// Code generated by protoc-gen-go. DO NOT EDIT.
// versions:
// protoc-gen-go v1.31.0
// protoc v4.23.3
// protoc v4.24.3
// source: agent/agent.proto
package agent
@@ -123,6 +123,7 @@ type AlgoRequest struct {
unknownFields protoimpl.UnknownFields
Algorithm []byte `protobuf:"bytes,1,opt,name=algorithm,proto3" json:"algorithm,omitempty"`
Provider string `protobuf:"bytes,2,opt,name=provider,proto3" json:"provider,omitempty"`
}
func (x *AlgoRequest) Reset() {
@@ -164,6 +165,13 @@ func (x *AlgoRequest) GetAlgorithm() []byte {
return nil
}
func (x *AlgoRequest) GetProvider() string {
if x != nil {
return x.Provider
}
return ""
}
type AlgoResponse struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
@@ -216,7 +224,8 @@ type DataRequest struct {
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Dataset []byte `protobuf:"bytes,1,opt,name=dataset,proto3" json:"dataset,omitempty"`
Dataset []byte `protobuf:"bytes,1,opt,name=dataset,proto3" json:"dataset,omitempty"`
Provider string `protobuf:"bytes,2,opt,name=provider,proto3" json:"provider,omitempty"`
}
func (x *DataRequest) Reset() {
@@ -258,6 +267,13 @@ func (x *DataRequest) GetDataset() []byte {
return nil
}
func (x *DataRequest) GetProvider() string {
if x != nil {
return x.Provider
}
return ""
}
type DataResponse struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
@@ -309,6 +325,8 @@ type ResultRequest struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
unknownFields protoimpl.UnknownFields
Consumer string `protobuf:"bytes,1,opt,name=consumer,proto3" json:"consumer,omitempty"`
}
func (x *ResultRequest) Reset() {
@@ -343,6 +361,13 @@ func (*ResultRequest) Descriptor() ([]byte, []int) {
return file_agent_agent_proto_rawDescGZIP(), []int{6}
}
func (x *ResultRequest) GetConsumer() string {
if x != nil {
return x.Consumer
}
return ""
}
type ResultResponse struct {
state protoimpl.MessageState
sizeCache protoimpl.SizeCache
@@ -485,46 +510,51 @@ var file_agent_agent_proto_rawDesc = []byte{
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, 0x2b, 0x0a, 0x0b, 0x41,
0x43, 0x6f, 0x6d, 0x70, 0x75, 0x74, 0x61, 0x74, 0x69, 0x6f, 0x6e, 0x22, 0x47, 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, 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, 0x27, 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, 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, 0x0f, 0x0a, 0x0d, 0x52, 0x65, 0x73, 0x75, 0x6c, 0x74, 0x52, 0x65, 0x71, 0x75, 0x65,
0x73, 0x74, 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,
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, 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, 0x43, 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, 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,
}
var (
agent/agent.proto
+11 -3
View File
@@ -19,15 +19,23 @@ message RunRequest { bytes computation = 1; }
message RunResponse { string Computation = 1; }
message AlgoRequest { bytes algorithm = 1; }
message AlgoRequest {
bytes algorithm = 1;
string provider = 2;
}
message AlgoResponse { string algorithmID = 1; }
message DataRequest { bytes dataset = 1; }
message DataRequest {
bytes dataset = 1;
string provider = 2;
}
message DataResponse { string datasetID = 1; }
message ResultRequest {}
message ResultRequest {
string consumer = 1;
}
message ResultResponse { bytes file = 1; }
agent/agent_grpc.pb.go
+1 -1
View File
@@ -4,7 +4,7 @@
// Code generated by protoc-gen-go-grpc. DO NOT EDIT.
// versions:
// - protoc-gen-go-grpc v1.3.0
// - protoc v4.23.3
// - protoc v4.24.3
// source: agent/agent.proto
package agent
agent/api/grpc/endpoint.go
+7 -3
View File
@@ -41,7 +41,9 @@ func algoEndpoint(svc agent.Service) endpoint.Endpoint {
return algoRes{}, err
}
algorithmID, err := svc.Algo(ctx, req.Algorithm)
algo := agent.Algorithm{Algorithm: req.Algorithm, Provider: req.Provider}
algorithmID, err := svc.Algo(ctx, algo)
if err != nil {
return algoRes{}, err
}
@@ -58,7 +60,9 @@ func dataEndpoint(svc agent.Service) endpoint.Endpoint {
return dataRes{}, err
}
datasetID, err := svc.Data(ctx, req.Dataset)
dataset := agent.Dataset{Dataset: req.Dataset, Provider: req.Provider}
datasetID, err := svc.Data(ctx, dataset)
if err != nil {
return dataRes{}, err
}
@@ -74,7 +78,7 @@ func resultEndpoint(svc agent.Service) endpoint.Endpoint {
if err := req.validate(); err != nil {
return resultRes{}, err
}
file, err := svc.Result(ctx)
file, err := svc.Result(ctx, req.Consumer)
if err != nil {
return resultRes{}, err
}
agent/api/grpc/requests.go
+4 -2
View File
@@ -17,6 +17,7 @@ func (req runReq) validate() error {
type algoReq struct {
Algorithm []byte `protobuf:"bytes,1,opt,name=algorithm,proto3" json:"algorithm,omitempty"`
Provider string `protobuf:"bytes,2,opt,name=provider,proto3" json:"provider,omitempty"`
}
func (req algoReq) validate() error {
@@ -27,7 +28,8 @@ func (req algoReq) validate() error {
}
type dataReq struct {
Dataset []byte `protobuf:"bytes,1,opt,name=dataset,proto3" json:"dataset,omitempty"`
Dataset []byte `protobuf:"bytes,1,opt,name=dataset,proto3" json:"dataset,omitempty"`
Provider string `protobuf:"bytes,2,opt,name=provider,proto3" json:"provider,omitempty"`
}
func (req dataReq) validate() error {
@@ -38,7 +40,7 @@ func (req dataReq) validate() error {
}
type resultReq struct {
// No request parameters needed for retrieving computation result file
Consumer string `protobuf:"bytes,1,opt,name=consumer,proto3" json:"consumer,omitempty"`
}
func (req resultReq) validate() error {
agent/api/http/doc.go
-5
View File
@@ -1,5 +0,0 @@
// Copyright (c) Ultraviolet
// SPDX-License-Identifier: Apache-2.0
// Package http contains implementation of kit service HTTP API.
package http
agent/api/http/endpoint.go
-44
View File
@@ -1,44 +0,0 @@
// Copyright (c) Ultraviolet
// SPDX-License-Identifier: Apache-2.0
package http
import (
"context"
"github.com/go-kit/kit/endpoint"
"github.com/ultravioletrs/cocos-ai/agent"
)
func runEndpoint(svc agent.Service) endpoint.Endpoint {
return func(ctx context.Context, request interface{}) (interface{}, error) {
req := request.(runReq)
if err := req.validate(); err != nil {
return runRes{}, err
}
cmp := agent.Computation{
ID: req.computation.ID,
Name: req.computation.Name,
Description: req.computation.Description,
Status: req.computation.Status,
Owner: req.computation.Owner,
StartTime: req.computation.StartTime,
EndTime: req.computation.EndTime,
Datasets: req.computation.Datasets,
Algorithms: req.computation.Algorithms,
DatasetProviders: req.computation.DatasetProviders,
AlgorithmProviders: req.computation.AlgorithmProviders,
ResultConsumers: req.computation.ResultConsumers,
Ttl: req.computation.Ttl,
Metadata: req.computation.Metadata,
}
cmpStr, err := svc.Run(ctx, cmp)
if err != nil {
return runRes{}, err
}
return runRes{Computation: cmpStr}, nil
}
}
agent/api/http/requests.go
-14
View File
@@ -1,14 +0,0 @@
// Copyright (c) Ultraviolet
// SPDX-License-Identifier: Apache-2.0
package http
import "github.com/ultravioletrs/cocos-ai/agent"
type runReq struct {
computation agent.Computation
}
func (req runReq) validate() error {
return nil
}
agent/api/http/responses.go
-28
View File
@@ -1,28 +0,0 @@
// Copyright (c) Ultraviolet
// SPDX-License-Identifier: Apache-2.0
package http
import (
"net/http"
"github.com/mainflux/mainflux"
)
var _ mainflux.Response = (*runRes)(nil)
type runRes struct {
Computation string `json:"computation"`
}
func (res runRes) Code() int {
return http.StatusOK
}
func (res runRes) Headers() map[string]string {
return map[string]string{}
}
func (res runRes) Empty() bool {
return false
}
agent/api/http/transport.go
-107
View File
@@ -1,107 +0,0 @@
// Copyright (c) Ultraviolet
// SPDX-License-Identifier: Apache-2.0
package http
import (
"context"
"encoding/json"
"errors"
"io"
"net/http"
"strings"
kithttp "github.com/go-kit/kit/transport/http"
"github.com/go-zoo/bone"
"github.com/mainflux/mainflux"
"github.com/prometheus/client_golang/prometheus/promhttp"
"github.com/ultravioletrs/cocos-ai/agent"
"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
)
const contentType = "application/json"
var (
errUnsupportedContentType = errors.New("unsupported content type")
errInvalidQueryParams = errors.New("invalid query params")
)
// MakeHandler returns a HTTP handler for API endpoints.
func MakeHandler(svc agent.Service, instanceID string) http.Handler {
opts := []kithttp.ServerOption{
kithttp.ServerErrorEncoder(encodeError),
}
r := bone.New()
r.Post("/run", otelhttp.NewHandler(kithttp.NewServer(
runEndpoint(svc),
decodeRun,
encodeResponse,
opts...,
), "run"))
r.GetFunc("/health", mainflux.Health("agent", instanceID))
r.Handle("/metrics", promhttp.Handler())
return r
}
func decodeRun(_ context.Context, r *http.Request) (interface{}, error) {
if !strings.Contains(r.Header.Get("Content-Type"), contentType) {
return nil, errUnsupportedContentType
}
var req runReq
if err := json.NewDecoder(r.Body).Decode(&req.computation); err != nil {
return nil, err
}
return req, nil
}
func encodeResponse(_ context.Context, w http.ResponseWriter, response interface{}) error {
w.Header().Set("Content-Type", contentType)
if ar, ok := response.(mainflux.Response); ok {
for k, v := range ar.Headers() {
w.Header().Set(k, v)
}
w.WriteHeader(ar.Code())
if ar.Empty() {
return nil
}
}
return json.NewEncoder(w).Encode(response)
}
func encodeError(_ context.Context, err error, w http.ResponseWriter) {
w.Header().Set("Content-Type", contentType)
switch err {
case agent.ErrMalformedEntity:
w.WriteHeader(http.StatusBadRequest)
case agent.ErrUnauthorizedAccess:
w.WriteHeader(http.StatusForbidden)
case errUnsupportedContentType:
w.WriteHeader(http.StatusUnsupportedMediaType)
case errInvalidQueryParams:
w.WriteHeader(http.StatusBadRequest)
case io.ErrUnexpectedEOF:
w.WriteHeader(http.StatusBadRequest)
case io.EOF:
w.WriteHeader(http.StatusBadRequest)
default:
switch err.(type) {
case *json.SyntaxError:
w.WriteHeader(http.StatusBadRequest)
case *json.UnmarshalTypeError:
w.WriteHeader(http.StatusBadRequest)
default:
w.WriteHeader(http.StatusInternalServerError)
}
}
}
agent/api/logging.go
+4 -4
View File
@@ -40,7 +40,7 @@ func (lm *loggingMiddleware) Run(ctx context.Context, cmp agent.Computation) (re
return lm.svc.Run(ctx, cmp)
}
func (lm *loggingMiddleware) Algo(ctx context.Context, algorithm []byte) (response string, err error) {
func (lm *loggingMiddleware) Algo(ctx context.Context, algorithm agent.Algorithm) (response string, err error) {
defer func(begin time.Time) {
message := fmt.Sprintf("Method Algo took %s to complete", time.Since(begin))
if err != nil {
@@ -53,7 +53,7 @@ func (lm *loggingMiddleware) Algo(ctx context.Context, algorithm []byte) (respon
return lm.svc.Algo(ctx, algorithm)
}
func (lm *loggingMiddleware) Data(ctx context.Context, dataset []byte) (response string, err error) {
func (lm *loggingMiddleware) Data(ctx context.Context, dataset agent.Dataset) (response string, err error) {
defer func(begin time.Time) {
message := fmt.Sprintf("Method Data took %s to complete", time.Since(begin))
if err != nil {
@@ -66,7 +66,7 @@ func (lm *loggingMiddleware) Data(ctx context.Context, dataset []byte) (response
return lm.svc.Data(ctx, dataset)
}
func (lm *loggingMiddleware) Result(ctx context.Context) (response []byte, err error) {
func (lm *loggingMiddleware) Result(ctx context.Context, consumer string) (response []byte, err error) {
defer func(begin time.Time) {
message := fmt.Sprintf("Method Result took %s to complete", time.Since(begin))
if err != nil {
@@ -76,7 +76,7 @@ func (lm *loggingMiddleware) Result(ctx context.Context) (response []byte, err e
lm.logger.Info(fmt.Sprintf("%s without errors", message))
}(time.Now())
return lm.svc.Result(ctx)
return lm.svc.Result(ctx, consumer)
}
func (lm *loggingMiddleware) Attestation(ctx context.Context) (response []byte, err error) {
agent/api/metrics.go
+4 -4
View File
@@ -41,7 +41,7 @@ func (ms *metricsMiddleware) Run(ctx context.Context, cmp agent.Computation) (st
return ms.svc.Run(ctx, cmp)
}
func (ms *metricsMiddleware) Algo(ctx context.Context, algorithm []byte) (string, error) {
func (ms *metricsMiddleware) Algo(ctx context.Context, algorithm agent.Algorithm) (string, error) {
defer func(begin time.Time) {
ms.counter.With("method", "algo").Add(1)
ms.latency.With("method", "algo").Observe(time.Since(begin).Seconds())
@@ -50,7 +50,7 @@ func (ms *metricsMiddleware) Algo(ctx context.Context, algorithm []byte) (string
return ms.svc.Algo(ctx, algorithm)
}
func (ms *metricsMiddleware) Data(ctx context.Context, dataset []byte) (string, error) {
func (ms *metricsMiddleware) Data(ctx context.Context, dataset agent.Dataset) (string, error) {
defer func(begin time.Time) {
ms.counter.With("method", "data").Add(1)
ms.latency.With("method", "data").Observe(time.Since(begin).Seconds())
@@ -59,13 +59,13 @@ func (ms *metricsMiddleware) Data(ctx context.Context, dataset []byte) (string,
return ms.svc.Data(ctx, dataset)
}
func (ms *metricsMiddleware) Result(ctx context.Context) ([]byte, error) {
func (ms *metricsMiddleware) Result(ctx context.Context, consumer string) ([]byte, error) {
defer func(begin time.Time) {
ms.counter.With("method", "result").Add(1)
ms.latency.With("method", "result").Observe(time.Since(begin).Seconds())
}(time.Now())
return ms.svc.Result(ctx)
return ms.svc.Result(ctx, consumer)
}
func (ms *metricsMiddleware) Attestation(ctx context.Context) ([]byte, error) {
agent/computations.go
+42 -15
View File
@@ -2,21 +2,48 @@
// SPDX-License-Identifier: Apache-2.0
package agent
import "time"
import (
reflect "reflect"
"time"
)
type Computation struct {
ID string `json:"id,omitempty" db:"id"`
Name string `json:"name,omitempty" db:"name"`
Description string `json:"description,omitempty" db:"description"`
Status string `json:"status,omitempty" db:"status"`
Owner string `json:"owner,omitempty" db:"owner"`
StartTime time.Time `json:"start_time,omitempty" db:"start_time"`
EndTime time.Time `json:"end_time,omitempty" db:"end_time"`
Datasets []string `json:"datasets,omitempty" db:"datasets"`
Algorithms []string `json:"algorithms,omitempty" db:"algorithms"`
DatasetProviders []string `json:"dataset_providers,omitempty" db:"dataset_providers"`
AlgorithmProviders []string `json:"algorithm_providers,omitempty" db:"algorithm_providers"`
ResultConsumers []string `json:"result_consumers,omitempty" db:"result_consumers"`
Ttl int32 `json:"ttl,omitempty" db:"ttl"`
Metadata Metadata `json:"metadata,omitempty" db:"metadata"`
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 []Dataset `json:"datasets,omitempty"`
Algorithms []Algorithm `json:"algorithms,omitempty"`
ResultConsumers []string `json:"result_consumers,omitempty"`
Ttl int32 `json:"ttl,omitempty"`
Metadata Metadata `json:"metadata,omitempty"`
}
type Dataset struct {
Dataset []byte `json:"-"`
Provider string `json:"provider,omitempty"`
ID string `json:"id,omitempty"`
}
type Algorithm struct {
Algorithm []byte `json:"-"`
Provider string `json:"provider,omitempty"`
ID string `json:"id,omitempty"`
}
func containsID(slice interface{}, id string) int {
rangeOnMe := reflect.ValueOf(slice)
for i := 0; i < rangeOnMe.Len(); i++ {
s := rangeOnMe.Index(i)
f := s.FieldByName("ID")
if f.IsValid() {
if f.Interface() == id {
return i
}
}
}
return -1
}
agent/service.go
+59 -17
View File
@@ -11,18 +11,30 @@ import (
"errors"
"fmt"
"os/exec"
"slices"
"github.com/ultravioletrs/cocos-ai/pkg/socket"
)
var _ Service = (*agentService)(nil)
var (
// ErrMalformedEntity indicates malformed entity specification (e.g.
// invalid username or password).
ErrMalformedEntity = errors.New("malformed entity specification")
// ErrUnauthorizedAccess indicates missing or invalid credentials provided
// when accessing a protected resource.
ErrUnauthorizedAccess = errors.New("missing or invalid credentials provided")
// errUndeclaredAlgorithm indicates algorithm was not declared in computation manifest.
errUndeclaredAlgorithm = errors.New("algorithm not declared in computation manifest")
// errUndeclaredAlgorithm indicates algorithm was not declared in computation manifest.
errUndeclaredDataset = errors.New("dataset not declared in computation manifest")
// errProviderMissmatch algorithm/dataset provider does not match computation manifest.
errProviderMissmatch = errors.New("provider does not match declaration on manifest")
// errAllManifestItemsReceived indicates no new computation manifest items expected.
errAllManifestItemsReceived = errors.New("all expected manifest Items have been received")
// errUndeclaredConsumer indicates the consumer requesting results in not declared in computation manifest.
errUndeclaredConsumer = errors.New("result consumer is undeclared in computation manifest")
)
type Metadata map[string]interface{}
@@ -31,9 +43,9 @@ type Metadata map[string]interface{}
// implementation, and all of its decorators (e.g. logging & metrics).
type Service interface {
Run(ctx context.Context, cmp Computation) (string, error)
Algo(ctx context.Context, algorithm []byte) (string, error)
Data(ctx context.Context, dataset []byte) (string, error)
Result(ctx context.Context) ([]byte, error)
Algo(ctx context.Context, algorithm Algorithm) (string, error)
Data(ctx context.Context, dataset Dataset) (string, error)
Result(ctx context.Context, consumer string) ([]byte, error)
Attestation(ctx context.Context) ([]byte, error)
}
@@ -72,37 +84,67 @@ func (as *agentService) Run(ctx context.Context, cmp Computation) (string, error
return cmpHash, nil // return computation hash.
}
func (as *agentService) Algo(ctx context.Context, algorithm []byte) (string, error) {
// Implement the logic for the Algo method based on your requirements.
// Use the provided ctx and algorithm parameters as needed.
func (as *agentService) Algo(ctx context.Context, algorithm Algorithm) (string, error) {
if len(as.computation.Algorithms) == 0 {
return "", errAllManifestItemsReceived
}
index := containsID(as.computation.Algorithms, algorithm.ID)
switch index {
case -1:
return "", errUndeclaredAlgorithm
default:
if as.computation.Algorithms[index].Provider != algorithm.Provider {
return "", errProviderMissmatch
}
as.computation.Algorithms = slices.Delete(as.computation.Algorithms, index, index+1)
}
as.algorithms = append(as.algorithms, algorithm)
as.algorithms = append(as.algorithms, algorithm.Algorithm)
// Calculate the SHA-256 hash of the algorithm.
hash := sha256.Sum256(algorithm)
hash := sha256.Sum256(algorithm.Algorithm)
algorithmHash := hex.EncodeToString(hash[:])
// Return the algorithm hash or an error.
return algorithmHash, nil
}
func (as *agentService) Data(ctx context.Context, dataset []byte) (string, error) {
// Implement the logic for the Data method based on your requirements.
// Use the provided ctx and dataset parameters as needed.
func (as *agentService) Data(ctx context.Context, dataset Dataset) (string, error) {
if len(as.computation.Datasets) == 0 {
return "", errAllManifestItemsReceived
}
index := containsID(as.computation.Datasets, dataset.ID)
switch index {
case -1:
return "", errUndeclaredDataset
default:
if as.computation.Datasets[index].Provider != dataset.Provider {
return "", errProviderMissmatch
}
as.computation.Datasets = slices.Delete(as.computation.Datasets, index, index+1)
}
as.datasets = append(as.datasets, dataset)
as.datasets = append(as.datasets, dataset.Dataset)
// Calculate the SHA-256 hash of the dataset.
hash := sha256.Sum256(dataset)
hash := sha256.Sum256(dataset.Dataset)
datasetHash := hex.EncodeToString(hash[:])
// Return the dataset hash or an error.
return datasetHash, nil
}
func (as *agentService) Result(ctx context.Context) ([]byte, error) {
// Implement the logic for the Result method based on your requirements
// Use the provided ctx parameter as needed
func (as *agentService) Result(ctx context.Context, consumer string) ([]byte, error) {
if len(as.computation.ResultConsumers) == 0 {
return []byte{}, errAllManifestItemsReceived
}
index := slices.Index(as.computation.ResultConsumers, consumer)
switch index {
case -1:
return []byte{}, errUndeclaredConsumer
default:
as.computation.ResultConsumers = slices.Delete(as.computation.ResultConsumers, index, index+1)
}
result, err := run(as.algorithms[0], as.datasets[0])
if err != nil {
agent/tracing/tracing.go
+14 -6
View File
@@ -23,6 +23,14 @@ func New(svc agent.Service, tracer trace.Tracer) agent.Service {
}
func (tm *tracingMiddleware) Run(ctx context.Context, cmp agent.Computation) (string, error) {
var datasetProviders, algorithmProviders []string
for _, dataset := range cmp.Datasets {
datasetProviders = append(datasetProviders, dataset.Provider)
}
for _, algos := range cmp.Algorithms {
algorithmProviders = append(algorithmProviders, algos.Provider)
}
ctx, span := tm.tracer.Start(ctx, "run", trace.WithAttributes(
attribute.String("id", cmp.ID),
attribute.String("name", cmp.Name),
@@ -30,8 +38,8 @@ func (tm *tracingMiddleware) Run(ctx context.Context, cmp agent.Computation) (st
attribute.String("status", cmp.Status),
attribute.String("start_time", cmp.StartTime.String()),
attribute.String("end_time", cmp.EndTime.String()),
attribute.StringSlice("dataset_providers", cmp.DatasetProviders),
attribute.StringSlice("algorithm_providers", cmp.AlgorithmProviders),
attribute.StringSlice("dataset_providers", datasetProviders),
attribute.StringSlice("algorithm_providers", algorithmProviders),
attribute.StringSlice("result_consumers", cmp.ResultConsumers),
))
defer span.End()
@@ -39,25 +47,25 @@ func (tm *tracingMiddleware) Run(ctx context.Context, cmp agent.Computation) (st
return tm.svc.Run(ctx, cmp)
}
func (tm *tracingMiddleware) Algo(ctx context.Context, algorithm []byte) (string, error) {
func (tm *tracingMiddleware) Algo(ctx context.Context, algorithm agent.Algorithm) (string, error) {
ctx, span := tm.tracer.Start(ctx, "algo")
defer span.End()
return tm.svc.Algo(ctx, algorithm)
}
func (tm *tracingMiddleware) Data(ctx context.Context, dataset []byte) (string, error) {
func (tm *tracingMiddleware) Data(ctx context.Context, dataset agent.Dataset) (string, error) {
ctx, span := tm.tracer.Start(ctx, "data")
defer span.End()
return tm.svc.Data(ctx, dataset)
}
func (tm *tracingMiddleware) Result(ctx context.Context) ([]byte, error) {
func (tm *tracingMiddleware) Result(ctx context.Context, consumer string) ([]byte, error) {
ctx, span := tm.tracer.Start(ctx, "result")
defer span.End()
return tm.svc.Result(ctx)
return tm.svc.Result(ctx, consumer)
}
func (tm *tracingMiddleware) Attestation(ctx context.Context) ([]byte, error) {
cmd/agent/main.go
+1 -8
View File
@@ -13,14 +13,12 @@ import (
"github.com/ultravioletrs/cocos-ai/agent"
"github.com/ultravioletrs/cocos-ai/agent/api"
agentgrpc "github.com/ultravioletrs/cocos-ai/agent/api/grpc"
httpapi "github.com/ultravioletrs/cocos-ai/agent/api/http"
"github.com/ultravioletrs/cocos-ai/agent/tracing"
"github.com/ultravioletrs/cocos-ai/internal"
"github.com/ultravioletrs/cocos-ai/internal/env"
jaegerclient "github.com/ultravioletrs/cocos-ai/internal/jaeger"
"github.com/ultravioletrs/cocos-ai/internal/server"
grpcserver "github.com/ultravioletrs/cocos-ai/internal/server/grpc"
httpserver "github.com/ultravioletrs/cocos-ai/internal/server/http"
"go.opentelemetry.io/otel/trace"
"golang.org/x/sync/errgroup"
"google.golang.org/grpc"
@@ -79,7 +77,6 @@ func main() {
if err := env.Parse(&httpServerConfig, env.Options{Prefix: envPrefixHTTP}); err != nil {
logger.Fatal(fmt.Sprintf("failed to load %s gRPC server configuration : %s", svcName, err))
}
hs := httpserver.New(ctx, cancel, svcName, httpServerConfig, httpapi.MakeHandler(svc, cfg.InstanceID), logger)
grpcServerConfig := server.Config{Port: defSvcGRPCPort}
if err := env.Parse(&grpcServerConfig, env.Options{Prefix: envPrefixGRPC}); err != nil {
@@ -92,16 +89,12 @@ func main() {
}
gs := grpcserver.New(ctx, cancel, svcName, grpcServerConfig, registerAgentServiceServer, logger)
g.Go(func() error {
return hs.Start()
})
g.Go(func() error {
return gs.Start()
})
g.Go(func() error {
return server.StopHandler(ctx, cancel, logger, svcName, hs, gs)
return server.StopHandler(ctx, cancel, logger, svcName, gs)
})
if err := g.Wait(); err != nil {
go.mod
+1 -1
View File
@@ -17,6 +17,7 @@ require (
go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.45.0
go.opentelemetry.io/contrib/propagators/jaeger v1.20.0
go.opentelemetry.io/otel v1.19.0
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.19.0
go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.19.0
go.opentelemetry.io/otel/sdk v1.19.0
go.opentelemetry.io/otel/trace v1.19.0
@@ -40,7 +41,6 @@ require (
github.com/prometheus/client_model v0.5.0 // indirect
github.com/prometheus/common v0.44.0 // indirect
github.com/prometheus/procfs v0.12.0 // indirect
go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.19.0 // indirect
go.opentelemetry.io/otel/metric v1.19.0 // indirect
go.opentelemetry.io/proto/otlp v1.0.0 // indirect
golang.org/x/net v0.17.0 // indirect