mirror of
https://github.com/ultravioletrs/cocos.git
synced 2026-08-07 07:14:50 +00:00
COCOS-24 - Add events service via http (#27)
* add status endpoint Signed-off-by: sammy <sammyoina@gmail.com> * feat: Update code generation tools to latest versions Update the code generation tools, including protoc-gen-go and protoc-gen-go-grpc, to their latest versions (v1.31.0 and v1.3.0, respectively). This ensures compatibility with the latest features and improvements. The updated tools also require gRPC-Go v1.32.0 or later. The new versions bring important updates and bug fixes, enhancing the performance and stability of the generated code. By staying up-to-date with the latest tooling, we can take advantage of the latest functionality and ensure a smooth development experience. No code changes are included in this commit. These updates only impact the code generation process. Signed-off-by: sammy <sammyoina@gmail.com> * Update Go version to 1.21.x Signed-off-by: sammy <sammyoina@gmail.com> * Refactor agent and manager services to publish event notifications The refactoring includes changes to the agent and manager services to incorporate event notifications. By publishing events, the services can inform subscribers about the current state of the computation or any updates. Specifically, the `agentService` now includes a `cmpHash` field to store the SHA-256 hash of the computation, which is subsequently used when publishing events. The `agentService` and `managerService` now use the `publisher` interface to publish events to the topic "manager". Notably, the removed `pubsub.go` file is no longer necessary. This commit improves the service architecture by allowing subscribers to receive relevant updates and monitor the progress of computations. It enhances the overall system by providing more transparency and enabling better coordination between the agent and manager services. Signed-off-by: sammy <sammyoina@gmail.com> * Improve generated Go file comparison in checkproto workflow Refactor the file comparison logic in the checkproto workflow to use the `-p` flag instead of `-s` for improved accuracy. This change ensures that the generated Go files are thoroughly compared with the original ones, detecting any discrepancies and preventing out-of-sync files from passing the validation. By using the `-p` flag, we now check both the contents and the metadata of the files, providing more robust synchronization checks. This update enhances the reliability of the checkproto workflow and helps maintain consistency between the proto files and their corresponding generated Go files. Signed-off-by: sammy <sammyoina@gmail.com> * Update file comparison command to detect differences line by line The code change updates the file comparison command used in the CI workflow to detect differences line by line instead of only reporting the first difference encountered. This change improves the accuracy of detecting inconsistencies between the original protobuf files and the generated Go files. Previously, only the first difference was reported, leading to potential missed issues. By comparing the files line by line, we can now detect and report all differences accurately. This change enhances the reliability of our CI pipeline and ensures that the generated Go files stay in sync with the protobuf files. Signed-off-by: sammy <sammyoina@gmail.com> * add event exporting to external server Signed-off-by: sammy <sammyoina@gmail.com> * feat: Add support for notification server URL The commit adds a new environment variable, `COCOS_NOTIFICATION_SERVER_URL`, which allows specifying the server to receive notification events from the agent. This addition provides flexibility to configure the notification server URL based on the deployment environment. This change enables seamless integration with different notification server instances and enhances the extensibility of the system. It resolves the need to modify the code directly when changing the server URL. Signed-off-by: sammy <sammyoina@gmail.com> * Refactor gRPC client and server, remove unused handlers The commit refactors the gRPC client and server code by removing the unused `nopDecoder` and `status` handlers from the client and server, respectively. This cleanup reduces code clutter and improves maintenance. No significant consequences are expected. Signed-off-by: sammy <sammyoina@gmail.com> * Ensure generated Go files stay in sync with proto files during the CI workflow Fixes an issue in the CI workflow where proto files and their corresponding generated Go files were not being properly compared for synchronization. Previously, the `cmp -l` command was used, which only printed differing byte positions, leading to false negatives. This has been corrected by using `cmp -s` instead, which outputs nothing if the files are identical. This change ensures that any differences between the proto files and their generated Go files will be detected, helping to maintain consistency and accuracy in the codebase. Signed-off-by: sammy <sammyoina@gmail.com> * Enhance notification payload and endpoint Extended the notification system to include 'status' and 'details' in the payload, improving traceability and debugging. Adapted the serialized JSON structure for clarity and added an 'originator' field to track the source service. Transitioned to a generalized event endpoint, facilitating a more streamlined event handling process. Refactors POST request to a more appropriate endpoint and updates the notification service interface to reflect new payload requirements. Signed-off-by: SammyOina <sammyoina@gmail.com> * Refactor event notification logic Removed the legacy notifications package and consolidated event notification functionality using the new internal events service. Modified agent, manager, and main application code to use this service for consistent event reporting and error handling workflows across services. This change simplifies event management, improves error visibility, and allows for more maintainable code by centralizing event-related logic. The substitution of verbose state-specific publishEvent calls with generic status reporting aligns with the new service's capabilities. Signed-off-by: SammyOina <sammyoina@gmail.com> * Expand agent protobuf message types and improve error handling The protobuf definition for agent messages has been updated to include an additional message type, facilitating future data structure expansions. Additionally, error handling for event sending in the main agent execution has been enhanced to log errors when sending 'init' events fail, ensuring issues are properly tracked. The unused `notificationTopic` constant in the manager service has been removed for cleaner code maintenance. Signed-off-by: SammyOina <sammyoina@gmail.com> * Optimize JSON parsing and fix header omission Removed unnecessary unquoting of a JSON string before unmarshaling, streamlining the computation value extraction process. Also corrected a missing Content-Type header in the event sending function, ensuring proper handling of JSON requests by recipients. These changes improve performance and communication reliability. Signed-off-by: SammyOina <sammyoina@gmail.com> * align vars Signed-off-by: SammyOina <sammyoina@gmail.com> * Remove computation request timeout feature The timeout feature for computation requests has been removed to simplify the computation execution flow. This involved changes across multiple files, including protobuf definitions, HTTP endpoint handling, and the internal computation logic. We eliminated the timeout field, associated logic, and error handling to ensure the system no longer supports timeouts for computations, mitigating any unintended timeout impacts on long-running processes. Signed-off-by: SammyOina <sammyoina@gmail.com> --------- Signed-off-by: sammy <sammyoina@gmail.com> Signed-off-by: SammyOina <sammyoina@gmail.com>
This commit is contained in:
Sammy Kerata Oina
committed by
GitHub
GitHub
parent
b55b6c9734
commit
c29ef354fa
@@ -29,7 +29,7 @@ jobs:
|
||||
- name: Install Go
|
||||
uses: actions/setup-go@v3
|
||||
with:
|
||||
go-version: 1.20.x
|
||||
go-version: 1.21.x
|
||||
|
||||
- name: Set up protoc
|
||||
run: |
|
||||
|
||||
+12
-11
@@ -6,17 +6,18 @@ Agent service provides a barebones HTTP and gRPC API and Service interface imple
|
||||
|
||||
The service is configured using the environment variables from the following table. Note that any unset variables will be replaced with their default values.
|
||||
|
||||
| Variable | Description | Default |
|
||||
| ------------------------ | ------------------------------------------------------ | ------------------------------ |
|
||||
| AGENT_LOG_LEVEL | Log level for agent service (debug, info, warn, error) | info |
|
||||
| AGENT_HTTP_HOST | Agent service HTTP host | "" |
|
||||
| AGENT_HTTP_PORT | Agent service HTTP port | 9031 |
|
||||
| AGENT_HTTP_SERVER_CERT | Path to HTTP server certificate in pem format | "" |
|
||||
| AGENT_HTTP_SERVER_KEY | Path to HTTP server key in pem format | "" |
|
||||
| AGENT_GRPC_HOST | Agent service gRPC host | "" |
|
||||
| AGENT_GRPC_PORT | Agent service gRPC port | 7002 |
|
||||
| AGENT_GRPC_SERVER_CERT | Path to gRPC server certificate in pem format | "" |
|
||||
| AGENT_GRPC_SERVER_KEY | Path to gRPC server key in pem format | "" |
|
||||
| Variable | Description | Default |
|
||||
| ----------------------------- | ------------------------------------------------------ | ------------------------------ |
|
||||
| AGENT_LOG_LEVEL | Log level for agent service (debug, info, warn, error) | info |
|
||||
| AGENT_HTTP_HOST | Agent service HTTP host | "" |
|
||||
| AGENT_HTTP_PORT | Agent service HTTP port | 9031 |
|
||||
| AGENT_HTTP_SERVER_CERT | Path to HTTP server certificate in pem format | "" |
|
||||
| AGENT_HTTP_SERVER_KEY | Path to HTTP server key in pem format | "" |
|
||||
| AGENT_GRPC_HOST | Agent service gRPC host | "" |
|
||||
| AGENT_GRPC_PORT | Agent service gRPC port | 7002 |
|
||||
| AGENT_GRPC_SERVER_CERT | Path to gRPC server certificate in pem format | "" |
|
||||
| AGENT_GRPC_SERVER_KEY | Path to gRPC server key in pem format | "" |
|
||||
| COCOS_NOTIFICATION_SERVER_URL | Server to receive notification events from agent. | http:/localhost:9000 |
|
||||
|
||||
|
||||
## Deployment
|
||||
|
||||
+26
-35
@@ -433,7 +433,6 @@ type ComputationReq struct {
|
||||
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() {
|
||||
@@ -510,13 +509,6 @@ func (x *ComputationReq) GetResultConsumers() []string {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *ComputationReq) GetTimeout() string {
|
||||
if x != nil {
|
||||
return x.Timeout
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
type DatasetReq struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
@@ -657,7 +649,7 @@ var file_agent_agent_proto_rawDesc = []byte{
|
||||
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,
|
||||
0x18, 0x01, 0x20, 0x01, 0x28, 0x0c, 0x52, 0x04, 0x66, 0x69, 0x6c, 0x65, 0x22, 0xe5, 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,
|
||||
@@ -672,32 +664,31 @@ var file_agent_agent_proto_rawDesc = []byte{
|
||||
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, 0xf5, 0x01, 0x0a, 0x0c, 0x41, 0x67, 0x65, 0x6e, 0x74, 0x53, 0x65,
|
||||
0x72, 0x76, 0x69, 0x63, 0x65, 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,
|
||||
0x6d, 0x65, 0x72, 0x73, 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, 0xf5, 0x01, 0x0a, 0x0c, 0x41,
|
||||
0x67, 0x65, 0x6e, 0x74, 0x53, 0x65, 0x72, 0x76, 0x69, 0x63, 0x65, 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 (
|
||||
|
||||
@@ -47,7 +47,6 @@ message ComputationReq {
|
||||
repeated DatasetReq datasets = 4;
|
||||
repeated AlgorithmReq algorithms = 5;
|
||||
repeated string result_consumers = 6;
|
||||
string timeout = 7;
|
||||
}
|
||||
|
||||
message DatasetReq {
|
||||
|
||||
@@ -4,10 +4,8 @@ package agent
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"reflect"
|
||||
"time"
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -22,7 +20,6 @@ type Computation struct {
|
||||
Datasets Datasets `json:"datasets,omitempty"`
|
||||
Algorithms Algorithms `json:"algorithms,omitempty"`
|
||||
ResultConsumers []string `json:"result_consumers,omitempty"`
|
||||
Timeout Duration `json:"timeout,omitempty"`
|
||||
}
|
||||
|
||||
func (d *Datasets) String() string {
|
||||
@@ -41,35 +38,6 @@ func (a *Algorithms) String() string {
|
||||
return string(dat)
|
||||
}
|
||||
|
||||
type Duration struct {
|
||||
time.Duration
|
||||
}
|
||||
|
||||
func (d Duration) MarshalJSON() ([]byte, error) {
|
||||
return json.Marshal(d.String())
|
||||
}
|
||||
|
||||
func (d *Duration) UnmarshalJSON(b []byte) error {
|
||||
var v interface{}
|
||||
if err := json.Unmarshal(b, &v); err != nil {
|
||||
return err
|
||||
}
|
||||
switch value := v.(type) {
|
||||
case float64:
|
||||
d.Duration = time.Duration(value)
|
||||
return nil
|
||||
case string:
|
||||
var err error
|
||||
d.Duration, err = time.ParseDuration(value)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
default:
|
||||
return errors.New("invalid duration")
|
||||
}
|
||||
}
|
||||
|
||||
type Dataset struct {
|
||||
Dataset []byte `json:"-"`
|
||||
Provider string `json:"provider,omitempty"`
|
||||
|
||||
+21
-21
@@ -12,9 +12,9 @@ import (
|
||||
"fmt"
|
||||
"os/exec"
|
||||
"slices"
|
||||
"time"
|
||||
|
||||
mglog "github.com/absmach/magistrala/logger"
|
||||
"github.com/ultravioletrs/cocos/internal/events"
|
||||
"github.com/ultravioletrs/cocos/pkg/socket"
|
||||
)
|
||||
|
||||
@@ -61,6 +61,7 @@ type agentService struct {
|
||||
attestation []byte
|
||||
sm *StateMachine
|
||||
runError error
|
||||
eventSvc events.Service
|
||||
}
|
||||
|
||||
const (
|
||||
@@ -72,18 +73,19 @@ const (
|
||||
var _ Service = (*agentService)(nil)
|
||||
|
||||
// New instantiates the agent service implementation.
|
||||
func New(ctx context.Context, logger mglog.Logger) Service {
|
||||
func New(ctx context.Context, logger mglog.Logger, eventSvc events.Service) Service {
|
||||
svc := &agentService{
|
||||
sm: NewStateMachine(logger),
|
||||
sm: NewStateMachine(logger),
|
||||
eventSvc: eventSvc,
|
||||
}
|
||||
go svc.sm.Start(ctx)
|
||||
svc.sm.SendEvent(start)
|
||||
svc.sm.StateFunctions[idle] = svc.publishEvent(ctx, "idle", "agent has started")
|
||||
svc.sm.StateFunctions[receivingManifests] = svc.publishEvent(ctx, "run", "agent ready to receive manifests")
|
||||
svc.sm.StateFunctions[receivingAlgorithms] = svc.publishEvent(ctx, "algorithms", "agent is ready to receiving algorithms")
|
||||
svc.sm.StateFunctions[receivingData] = svc.publishEvent(ctx, "datasets", "agent is ready to receiving datasets")
|
||||
svc.sm.StateFunctions[resultsReady] = svc.publishEvent(ctx, "results", "agent computation results are ready")
|
||||
svc.sm.StateFunctions[complete] = svc.publishEvent(ctx, "complete", "agent results have been consumed")
|
||||
svc.sm.StateFunctions[idle] = svc.publishEvent("in-progress", json.RawMessage{})
|
||||
svc.sm.StateFunctions[receivingManifests] = svc.publishEvent("in-progress", json.RawMessage{})
|
||||
svc.sm.StateFunctions[receivingAlgorithms] = svc.publishEvent("in-progress", json.RawMessage{})
|
||||
svc.sm.StateFunctions[receivingData] = svc.publishEvent("in-progress", json.RawMessage{})
|
||||
svc.sm.StateFunctions[resultsReady] = svc.publishEvent("in-progress", json.RawMessage{})
|
||||
svc.sm.StateFunctions[complete] = svc.publishEvent("in-progress", json.RawMessage{})
|
||||
svc.sm.StateFunctions[running] = svc.runComputation
|
||||
return svc
|
||||
}
|
||||
@@ -204,29 +206,29 @@ func (as *agentService) Attestation(ctx context.Context) ([]byte, error) {
|
||||
}
|
||||
|
||||
func (as *agentService) runComputation() {
|
||||
ctx := context.Background()
|
||||
as.publishEvent(ctx, "running", "computation run has started")()
|
||||
as.publishEvent("starting", json.RawMessage{})()
|
||||
as.sm.logger.Debug("computation run started")
|
||||
defer as.sm.SendEvent(runComplete)
|
||||
var cancel context.CancelFunc
|
||||
if as.computation.Timeout.Duration != 0 {
|
||||
ctx, cancel = context.WithDeadline(ctx, time.Now().Add(as.computation.Timeout.Duration))
|
||||
defer cancel()
|
||||
}
|
||||
result, err := run(ctx, as.algorithms[0], as.datasets[0])
|
||||
as.publishEvent("in-progress", json.RawMessage{})()
|
||||
result, err := run(as.algorithms[0], as.datasets[0])
|
||||
if err != nil {
|
||||
as.runError = err
|
||||
as.publishEvent("failed", json.RawMessage{})()
|
||||
return
|
||||
}
|
||||
as.publishEvent("complete", json.RawMessage{})()
|
||||
as.result = result
|
||||
}
|
||||
|
||||
func (as *agentService) publishEvent(ctx context.Context, subtopic, body string) func() {
|
||||
func (as *agentService) publishEvent(status string, details json.RawMessage) func() {
|
||||
return func() {
|
||||
if err := as.eventSvc.SendEvent(as.sm.State.String(), as.computation.ID, status, details); err != nil {
|
||||
as.sm.logger.Warn(err.Error())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func run(ctx context.Context, algoContent, dataContent []byte) ([]byte, error) {
|
||||
func run(algoContent, dataContent []byte) ([]byte, error) {
|
||||
listener, err := socket.StartUnixSocketServer(socketPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("error creating stdout pipe: %v", err)
|
||||
@@ -255,8 +257,6 @@ func run(ctx context.Context, algoContent, dataContent []byte) ([]byte, error) {
|
||||
}
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, errors.New("computation timed out")
|
||||
case result = <-dataChannel:
|
||||
return result, nil
|
||||
case err = <-errorChannel:
|
||||
|
||||
@@ -10,6 +10,7 @@ import (
|
||||
mglog "github.com/absmach/magistrala/logger"
|
||||
)
|
||||
|
||||
//go:generate stringer -type=state
|
||||
type state int
|
||||
|
||||
const (
|
||||
|
||||
@@ -0,0 +1,29 @@
|
||||
// Code generated by "stringer -type=state"; DO NOT EDIT.
|
||||
|
||||
package agent
|
||||
|
||||
import "strconv"
|
||||
|
||||
func _() {
|
||||
// An "invalid array index" compiler error signifies that the constant values have changed.
|
||||
// Re-run the stringer command to generate them again.
|
||||
var x [1]struct{}
|
||||
_ = x[idle-0]
|
||||
_ = x[receivingManifests-1]
|
||||
_ = x[receivingAlgorithms-2]
|
||||
_ = x[receivingData-3]
|
||||
_ = x[running-4]
|
||||
_ = x[resultsReady-5]
|
||||
_ = x[complete-6]
|
||||
}
|
||||
|
||||
const _state_name = "idlereceivingManifestsreceivingAlgorithmsreceivingDatarunningresultsReadycomplete"
|
||||
|
||||
var _state_index = [...]uint8{0, 4, 22, 41, 54, 61, 73, 81}
|
||||
|
||||
func (i state) String() string {
|
||||
if i < 0 || i >= state(len(_state_index)-1) {
|
||||
return "state(" + strconv.FormatInt(int64(i), 10) + ")"
|
||||
}
|
||||
return _state_name[_state_index[i]:_state_index[i+1]]
|
||||
}
|
||||
+12
-12
@@ -9,7 +9,6 @@ import (
|
||||
"fmt"
|
||||
"log"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
mglog "github.com/absmach/magistrala/logger"
|
||||
@@ -19,6 +18,7 @@ import (
|
||||
agentgrpc "github.com/ultravioletrs/cocos/agent/api/grpc"
|
||||
"github.com/ultravioletrs/cocos/internal"
|
||||
"github.com/ultravioletrs/cocos/internal/env"
|
||||
"github.com/ultravioletrs/cocos/internal/events"
|
||||
"github.com/ultravioletrs/cocos/internal/server"
|
||||
grpcserver "github.com/ultravioletrs/cocos/internal/server/grpc"
|
||||
"golang.org/x/sync/errgroup"
|
||||
@@ -37,8 +37,9 @@ const (
|
||||
var errComputationNotFound = errors.New("computation not found in command line")
|
||||
|
||||
type config struct {
|
||||
LogLevel string `env:"AGENT_LOG_LEVEL" envDefault:"info"`
|
||||
InstanceID string `env:"AGENT_INSTANCE_ID" envDefault:""`
|
||||
LogLevel string `env:"AGENT_LOG_LEVEL" envDefault:"info"`
|
||||
InstanceID string `env:"AGENT_INSTANCE_ID" envDefault:""`
|
||||
NotificationServerURL string `env:"COCOS_NOTIFICATION_SERVER_URL" envDefault:"http://localhost:9000"`
|
||||
}
|
||||
|
||||
func main() {
|
||||
@@ -62,13 +63,17 @@ func main() {
|
||||
}
|
||||
}
|
||||
|
||||
svc := newService(ctx, logger)
|
||||
eventSvc := events.New(svcName, cfg.NotificationServerURL)
|
||||
svc := newService(ctx, logger, eventSvc)
|
||||
|
||||
ac, err := extractComputationValue()
|
||||
if err != nil {
|
||||
logger.Fatal(fmt.Sprintf("computation not loaded from cmdline : %s", err.Error()))
|
||||
}
|
||||
if _, err := svc.Run(ctx, ac); err != nil {
|
||||
if err := eventSvc.SendEvent("init", ac.ID, "failed", json.RawMessage{}); err != nil {
|
||||
logger.Warn(err.Error())
|
||||
}
|
||||
logger.Fatal(fmt.Sprintf("failed to run computation with err: %s", err))
|
||||
}
|
||||
|
||||
@@ -96,8 +101,8 @@ func main() {
|
||||
}
|
||||
}
|
||||
|
||||
func newService(ctx context.Context, logger mglog.Logger) agent.Service {
|
||||
svc := agent.New(ctx, logger)
|
||||
func newService(ctx context.Context, logger mglog.Logger, eventSvc events.Service) agent.Service {
|
||||
svc := agent.New(ctx, logger, eventSvc)
|
||||
|
||||
svc = api.LoggingMiddleware(svc, logger)
|
||||
counter, latency := internal.MakeMetrics(svcName, "api")
|
||||
@@ -129,12 +134,7 @@ func extractComputationValue() (agent.Computation, error) {
|
||||
cmpUnescaped = cmdLine[start:]
|
||||
}
|
||||
|
||||
escapedCmpStr, err := strconv.Unquote(cmpUnescaped)
|
||||
if err != nil {
|
||||
return agent.Computation{}, err
|
||||
}
|
||||
|
||||
if err := json.Unmarshal([]byte(escapedCmpStr), &ac); err != nil {
|
||||
if err := json.Unmarshal([]byte(cmpUnescaped), &ac); err != nil {
|
||||
return agent.Computation{}, err
|
||||
}
|
||||
return ac, nil
|
||||
|
||||
+8
-6
@@ -15,6 +15,7 @@ import (
|
||||
"github.com/ultravioletrs/cocos/agent"
|
||||
"github.com/ultravioletrs/cocos/internal"
|
||||
"github.com/ultravioletrs/cocos/internal/env"
|
||||
"github.com/ultravioletrs/cocos/internal/events"
|
||||
jaegerclient "github.com/ultravioletrs/cocos/internal/jaeger"
|
||||
"github.com/ultravioletrs/cocos/internal/server"
|
||||
grpcserver "github.com/ultravioletrs/cocos/internal/server/grpc"
|
||||
@@ -42,9 +43,10 @@ const (
|
||||
)
|
||||
|
||||
type config struct {
|
||||
LogLevel string `env:"MANAGER_LOG_LEVEL" envDefault:"info"`
|
||||
JaegerURL string `env:"COCOS_JAEGER_URL" envDefault:"http://localhost:14268/api/traces"`
|
||||
InstanceID string `env:"MANAGER_INSTANCE_ID" envDefault:""`
|
||||
LogLevel string `env:"MANAGER_LOG_LEVEL" envDefault:"info"`
|
||||
JaegerURL string `env:"COCOS_JAEGER_URL" envDefault:"http://localhost:14268/api/traces"`
|
||||
InstanceID string `env:"MANAGER_INSTANCE_ID" envDefault:""`
|
||||
NotificationServerURL string `env:"COCOS_NOTIFICATION_SERVER_URL" envDefault:"http://localhost:9000"`
|
||||
}
|
||||
|
||||
func main() {
|
||||
@@ -89,7 +91,7 @@ func main() {
|
||||
}
|
||||
logger.Info(fmt.Sprintf("%s %s", exe, strings.Join(args, " ")))
|
||||
|
||||
svc := newService(logger, tracer, qemuCfg)
|
||||
svc := newService(logger, tracer, qemuCfg, events.New(svcName, cfg.NotificationServerURL))
|
||||
|
||||
httpServerConfig := server.Config{Port: defSvcHTTPPort}
|
||||
if err := env.Parse(&httpServerConfig, env.Options{Prefix: envPrefixHTTP}); err != nil {
|
||||
@@ -130,8 +132,8 @@ func main() {
|
||||
}
|
||||
}
|
||||
|
||||
func newService(logger mglog.Logger, tracer trace.Tracer, qemuCfg qemu.Config) manager.Service {
|
||||
svc := manager.New(qemuCfg)
|
||||
func newService(logger mglog.Logger, tracer trace.Tracer, qemuCfg qemu.Config, eventSvc events.Service) manager.Service {
|
||||
svc := manager.New(qemuCfg, logger, eventSvc)
|
||||
|
||||
svc = api.LoggingMiddleware(svc, logger)
|
||||
counter, latency := internal.MakeMetrics(svcName, "api")
|
||||
|
||||
@@ -8,6 +8,9 @@ COCOS_JAEGER_CONFIGS=5778
|
||||
COCOS_JAEGER_URL=http://jaeger:14268/api/traces
|
||||
COCOS_JAEGER_TRACE_RATIO=1.0
|
||||
|
||||
## Notification Server
|
||||
COCOS_NOTIFICATION_SERVER_URL="http://localhost:9000"
|
||||
|
||||
## Core Services
|
||||
|
||||
### Manager
|
||||
|
||||
@@ -0,0 +1,11 @@
|
||||
# Events package
|
||||
|
||||
## Overview
|
||||
The events package provides a simple client for sending events to a server. It is designed to be used in conjunction with a server that can handle and process events.
|
||||
|
||||
`service`: A string representing the name of the service. Agent or manager.
|
||||
`serverUrl`: A string representing the URL of the events server.
|
||||
`event`: A string representing the event type of the events.
|
||||
`computationId`: A string representing the computation ID associated with the events. This can be null for idle events.
|
||||
|
||||
The notifications are sent to endpoint `<server_url>/notifications/events` using post method.
|
||||
@@ -0,0 +1,65 @@
|
||||
// Copyright (c) Ultraviolet
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
package events
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"time"
|
||||
)
|
||||
|
||||
var errFailedToCreateNotification = errors.New("failed to create notification on server")
|
||||
|
||||
type service struct {
|
||||
service string
|
||||
serverUrl string
|
||||
}
|
||||
|
||||
type Service interface {
|
||||
SendEvent(event, computationId, status string, details json.RawMessage) error
|
||||
}
|
||||
|
||||
func New(svc, serverUrl string) Service {
|
||||
return &service{
|
||||
service: svc,
|
||||
serverUrl: serverUrl,
|
||||
}
|
||||
}
|
||||
|
||||
func (s *service) SendEvent(event, computationId, status string, details json.RawMessage) error {
|
||||
body := struct {
|
||||
EventType string `json:"event_type"`
|
||||
Timestamp time.Time `json:"timestamp"`
|
||||
ComputationID string `json:"computation_id,omitempty"`
|
||||
Details json.RawMessage `json:"details,omitempty"`
|
||||
Originator string `json:"originator"`
|
||||
Status string `json:"status,omitempty"`
|
||||
}{
|
||||
EventType: event,
|
||||
Timestamp: time.Now(),
|
||||
ComputationID: computationId,
|
||||
Originator: s.service,
|
||||
Status: status,
|
||||
Details: details,
|
||||
}
|
||||
jsonBody, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req, err := http.NewRequest(http.MethodPost, fmt.Sprintf("%s/computations/events", s.serverUrl), bytes.NewReader(jsonBody))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
res, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if res.StatusCode != http.StatusCreated {
|
||||
return errFailedToCreateNotification
|
||||
}
|
||||
return nil
|
||||
}
|
||||
+14
-13
@@ -6,19 +6,20 @@ Manager service provides a barebones HTTP and gRPC API and Service interface imp
|
||||
|
||||
The service is configured using the environment variables from the following table. Note that any unset variables will be replaced with their default values.
|
||||
|
||||
| Variable | Description | Default |
|
||||
| ------------------------ | -------------------------------------------------------- | --------------------------------- |
|
||||
| MANAGER_LOG_LEVEL | Log level for manager service (debug, info, warn, error) | info |
|
||||
| MANAGER_HTTP_HOST | Manager service HTTP host | |
|
||||
| MANAGER_HTTP_PORT | Manager service HTTP port | 9021 |
|
||||
| MANAGER_HTTP_SERVER_CERT | Path to server certificate in pem format | |
|
||||
| MANAGER_HTTP_SERVER_KEY | Path to server key in pem format | |
|
||||
| MANAGER_GRPC_HOST | Manager service gRPC host | |
|
||||
| MANAGER_GRPC_PORT | Manager service gRPC port | 7001 |
|
||||
| MANAGER_GRPC_SERVER_CERT | Path to server certificate in pem format | |
|
||||
| MANAGER_GRPC_SERVER_KEY | Path to server key in pem format | |
|
||||
| COCOS_JAEGER_URL | Jaeger server URL | http://localhost:14268/api/traces |
|
||||
| MANAGER_INSTANCE_ID | Manager service instance ID | |
|
||||
| Variable | Description | Default |
|
||||
| ----------------------------- | -------------------------------------------------------- | --------------------------------- |
|
||||
| MANAGER_LOG_LEVEL | Log level for manager service (debug, info, warn, error) | info |
|
||||
| MANAGER_HTTP_HOST | Manager service HTTP host | |
|
||||
| MANAGER_HTTP_PORT | Manager service HTTP port | 9021 |
|
||||
| MANAGER_HTTP_SERVER_CERT | Path to server certificate in pem format | |
|
||||
| MANAGER_HTTP_SERVER_KEY | Path to server key in pem format | |
|
||||
| MANAGER_GRPC_HOST | Manager service gRPC host | |
|
||||
| MANAGER_GRPC_PORT | Manager service gRPC port | 7001 |
|
||||
| MANAGER_GRPC_SERVER_CERT | Path to server certificate in pem format | |
|
||||
| MANAGER_GRPC_SERVER_KEY | Path to server key in pem format | |
|
||||
| COCOS_JAEGER_URL | Jaeger server URL | http://localhost:14268/api/traces |
|
||||
| MANAGER_INSTANCE_ID | Manager service instance ID | |
|
||||
| COCOS_NOTIFICATION_SERVER_URL | Server to receive notification events from agent. | http:/localhost:9000 |
|
||||
|
||||
## Deployment
|
||||
|
||||
|
||||
@@ -22,7 +22,6 @@ func runEndpoint(svc manager.Service) endpoint.Endpoint {
|
||||
agentConf := grpc.Config{
|
||||
ClientTLS: req.ClientTLS,
|
||||
CACerts: req.CACerts,
|
||||
Timeout: req.Timeout.Duration,
|
||||
}
|
||||
if agentConf.Timeout == 0 {
|
||||
agentConf.Timeout = 60 * time.Second
|
||||
@@ -33,7 +32,6 @@ func runEndpoint(svc manager.Service) endpoint.Endpoint {
|
||||
Name: req.Computation.Name,
|
||||
Description: req.Computation.Description,
|
||||
ResultConsumers: req.Computation.ResultConsumers,
|
||||
Timeout: req.Computation.Timeout.String(),
|
||||
}
|
||||
for _, algo := range req.Computation.Algorithms {
|
||||
mc.Algorithms = append(mc.Algorithms, &manager.Algorithm{Id: algo.ID, Provider: algo.Provider})
|
||||
|
||||
@@ -16,7 +16,6 @@ type runReq struct {
|
||||
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 {
|
||||
|
||||
+16
-26
@@ -105,7 +105,6 @@ type Computation struct {
|
||||
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() {
|
||||
@@ -182,13 +181,6 @@ func (x *Computation) GetResultConsumers() []string {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (x *Computation) GetTimeout() string {
|
||||
if x != nil {
|
||||
return x.Timeout
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
type Dataset struct {
|
||||
state protoimpl.MessageState
|
||||
sizeCache protoimpl.SizeCache
|
||||
@@ -351,7 +343,7 @@ var file_manager_manager_proto_rawDesc = []byte{
|
||||
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,
|
||||
0x28, 0x09, 0x52, 0x07, 0x74, 0x69, 0x6d, 0x65, 0x6f, 0x75, 0x74, 0x22, 0xe0, 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,
|
||||
@@ -365,23 +357,21 @@ var file_manager_manager_proto_rawDesc = []byte{
|
||||
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, 0x0d, 0x0a, 0x0b, 0x52, 0x75, 0x6e, 0x52,
|
||||
0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 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,
|
||||
0x65, 0x73, 0x75, 0x6c, 0x74, 0x43, 0x6f, 0x6e, 0x73, 0x75, 0x6d, 0x65, 0x72, 0x73, 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, 0x0d,
|
||||
0x0a, 0x0b, 0x52, 0x75, 0x6e, 0x52, 0x65, 0x73, 0x70, 0x6f, 0x6e, 0x73, 0x65, 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 (
|
||||
|
||||
@@ -25,7 +25,6 @@ message Computation {
|
||||
repeated Dataset datasets = 4;
|
||||
repeated Algorithm algorithms = 5;
|
||||
repeated string result_consumers = 6;
|
||||
string timeout = 7;
|
||||
}
|
||||
|
||||
message Dataset {
|
||||
|
||||
+21
-11
@@ -4,10 +4,12 @@ package manager
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
mglog "github.com/absmach/magistrala/logger"
|
||||
"github.com/ultravioletrs/cocos/agent"
|
||||
"github.com/ultravioletrs/cocos/internal/events"
|
||||
"github.com/ultravioletrs/cocos/manager/qemu"
|
||||
)
|
||||
|
||||
@@ -31,30 +33,29 @@ type Service interface {
|
||||
}
|
||||
|
||||
type managerService struct {
|
||||
qemuCfg qemu.Config
|
||||
qemuCfg qemu.Config
|
||||
logger mglog.Logger
|
||||
eventSvc events.Service
|
||||
}
|
||||
|
||||
var _ Service = (*managerService)(nil)
|
||||
|
||||
// New instantiates the manager service implementation.
|
||||
func New(qemuCfg qemu.Config) Service {
|
||||
func New(qemuCfg qemu.Config, logger mglog.Logger, eventSvc events.Service) Service {
|
||||
return &managerService{
|
||||
qemuCfg: qemuCfg,
|
||||
qemuCfg: qemuCfg,
|
||||
eventSvc: eventSvc,
|
||||
}
|
||||
}
|
||||
|
||||
func (ms *managerService) Run(ctx context.Context, c *Computation) error {
|
||||
ms.publishEvent("vm-provision", c.Id, "starting", json.RawMessage{})
|
||||
ac := agent.Computation{
|
||||
ID: c.Id,
|
||||
Name: c.Name,
|
||||
Description: c.Description,
|
||||
ResultConsumers: c.ResultConsumers,
|
||||
}
|
||||
dur, err := time.ParseDuration(c.Timeout)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ac.Timeout.Duration = dur
|
||||
for _, algo := range c.Algorithms {
|
||||
ac.Algorithms = append(ac.Algorithms, agent.Algorithm{ID: algo.Id, Provider: algo.Provider})
|
||||
}
|
||||
@@ -62,15 +63,24 @@ func (ms *managerService) Run(ctx context.Context, c *Computation) error {
|
||||
ac.Datasets = append(ac.Datasets, agent.Dataset{ID: data.Id, Provider: data.Provider})
|
||||
}
|
||||
|
||||
if _, err = qemu.CreateVM(ctx, ms.qemuCfg, ac); err != nil {
|
||||
ms.publishEvent("vm-provision", c.Id, "in-progress", json.RawMessage{})
|
||||
if _, err := qemu.CreateVM(ctx, ms.qemuCfg, ac); err != nil {
|
||||
ms.publishEvent("vm-provision", c.Id, "failed", json.RawMessage{})
|
||||
return err
|
||||
}
|
||||
// different VM guests can't forward ports to the same ports on the same host
|
||||
// Different VM guests can't forward ports to the same ports on the same host.
|
||||
defer func() {
|
||||
ms.qemuCfg.HostFwd1++
|
||||
ms.qemuCfg.NetDevConfig.HostFwd2++
|
||||
ms.qemuCfg.NetDevConfig.HostFwd3++
|
||||
}()
|
||||
|
||||
ms.publishEvent("vm-provision", c.Id, "complete", json.RawMessage{})
|
||||
return nil
|
||||
}
|
||||
|
||||
func (ms *managerService) publishEvent(event, cmpID, status string, details json.RawMessage) {
|
||||
if err := ms.eventSvc.SendEvent(event, cmpID, status, details); err != nil {
|
||||
ms.logger.Warn(err.Error())
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user