mirror of
https://github.com/ultravioletrs/cocos.git
synced 2026-08-07 07:14:50 +00:00
Sammy Kerata Oina
6d0100c096
* Refactor GRPC manager service and client The manager service and client have been restructured for stream communication, facilitating real-time agent events, logs, and run responses. The `Run` RPC is replaced by the `Process` stream RPC, enabling bidirectional streaming between clients and the manager service. This allows continuous interchange of different message types including `WhoAmIRequest`, `AgentLog`, `AgentEvent`, and `RunResponse`. Several message types have been adjusted and new fields introduced, like `AgentPort` in `RunResponse` and various agent-config attributes including CA files and instance IDs, to support TLS client authentication and distinguish between agent instances. We've also incorporated `google.protobuf.Timestamp` in `AgentEvent` for precise event logging. The client code reflects these modifications with updated method calls and stream handling logic for ongoing communication. Moreover, the updates necessitate corresponding changes throughout service, grpc, and sdk layers to interoperate with the new streaming approach. The transition to streaming paves the way for a more interactive, flexible communication system that can accommodate future expansion and real-time monitoring features. Signed-off-by: SammyOina <sammyoina@gmail.com> * add computation id to run response Signed-off-by: SammyOina <sammyoina@gmail.com> * rename request Signed-off-by: SammyOina <sammyoina@gmail.com> --------- Signed-off-by: SammyOina <sammyoina@gmail.com>
58 lines
1.3 KiB
Go
58 lines
1.3 KiB
Go
// Copyright (c) Ultraviolet
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
package grpc
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/ultravioletrs/cocos/manager"
|
|
"golang.org/x/sync/errgroup"
|
|
)
|
|
|
|
type ManagerClient struct {
|
|
stream manager.ManagerService_ProcessClient
|
|
svc manager.Service
|
|
responses chan *manager.ClientStreamMessage
|
|
}
|
|
|
|
// NewClient returns new gRPC client instance.
|
|
func NewClient(stream manager.ManagerService_ProcessClient, svc manager.Service, responses chan *manager.ClientStreamMessage) ManagerClient {
|
|
return ManagerClient{
|
|
stream: stream,
|
|
svc: svc,
|
|
responses: responses,
|
|
}
|
|
}
|
|
|
|
func (client ManagerClient) Process(ctx context.Context) error {
|
|
eg, ctx := errgroup.WithContext(ctx)
|
|
|
|
eg.Go(func() error {
|
|
for {
|
|
req, err := client.stream.Recv()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
port, err := client.svc.Run(ctx, req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
runRes := &manager.ClientStreamMessage_RunRes{RunRes: &manager.RunResponse{AgentPort: port, ComputationId: req.Id}}
|
|
if err := client.stream.Send(&manager.ClientStreamMessage{Message: runRes}); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
})
|
|
|
|
eg.Go(func() error {
|
|
for mes := range client.responses {
|
|
if err := client.stream.Send(mes); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
})
|
|
|
|
return eg.Wait()
|
|
}
|