COCOS-344 - New agent structure (#350)

* new agent structure Signed-off-by: Sammy Oina <sammyoina@gmail.com> * minor fixes and testing Signed-off-by: Sammy Oina <sammyoina@gmail.com> * fix lint Signed-off-by: Sammy Oina <sammyoina@gmail.com> * fix tests Signed-off-by: Sammy Oina <sammyoina@gmail.com> * cvm tests fix Signed-off-by: Sammy Oina <sammyoina@gmail.com> * fix test Signed-off-by: Sammy Oina <sammyoina@gmail.com> * fix cli test Signed-off-by: Sammy Oina <sammyoina@gmail.com> * rename Signed-off-by: Sammy Oina <sammyoina@gmail.com> * rename cvm to cvms plural Signed-off-by: Sammy Oina <sammyoina@gmail.com> * rename service Signed-off-by: Sammy Oina <sammyoina@gmail.com> * fix tests Signed-off-by: Sammy Oina <sammyoina@gmail.com> * remove context Signed-off-by: Sammy Oina <sammyoina@gmail.com> * refactor: reorder parameters in NewAlgorithm functions and update CVMClient to CVMSClient Signed-off-by: Sammy Oina <sammyoina@gmail.com> * fix(tests): update SendEvent mock to include an additional parameter Signed-off-by: Sammy Oina <sammyoina@gmail.com> * move expectations Signed-off-by: Sammy Oina <sammyoina@gmail.com> * fix(tests): move event initialization to the correct scope in service tests Signed-off-by: Sammy Oina <sammyoina@gmail.com> * fix(tests): update SendEvent mock to use EXPECT instead of On in service tests Signed-off-by: Sammy Oina <sammyoina@gmail.com> --------- Signed-off-by: Sammy Oina <sammyoina@gmail.com>
2026-06-23 04:10:25 +00:00 · 2025-01-17 14:50:53 +03:00
parent 59b8057e5c
commit ecad6514f3
53 changed files with 3300 additions and 1340 deletions
@@ -7,8 +7,9 @@ import (
 	"io"
 	"log/slog"

+	"github.com/ultravioletrs/cocos/agent/cvms"
 	"github.com/ultravioletrs/cocos/agent/events"
-	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/encoding/protojson"
 	"google.golang.org/protobuf/types/known/timestamppb"
 )

@@ -18,16 +19,17 @@ type handler struct {
 	opts  slog.HandlerOptions
 	w     io.Writer
 	cmpID string
+	queue chan *cvms.ClientStreamMessage
 }

-func NewProtoHandler(conn io.Writer, opts *slog.HandlerOptions, cmpID string) slog.Handler {
+func NewProtoHandler(conn io.Writer, opts *slog.HandlerOptions, queue chan *cvms.ClientStreamMessage) slog.Handler {
 	if opts == nil {
 		opts = &slog.HandlerOptions{}
 	}
 	h := &handler{
 		opts:  *opts,
 		w:     conn,
-		cmpID: cmpID,
+		queue: queue,
 	}

 	return h
@@ -69,7 +71,18 @@ func (h *handler) Handle(_ context.Context, r slog.Record) error {
 			},
 		}

-		b, err := proto.Marshal(&agentLog)
+		h.queue <- &cvms.ClientStreamMessage{
+			Message: &cvms.ClientStreamMessage_AgentLog{
+				AgentLog: &cvms.AgentLog{
+					Timestamp:     timestamp,
+					Message:       chunk,
+					Level:         level,
+					ComputationId: h.cmpID,
+				},
+			},
+		}
+
+		b, err := protojson.Marshal(&agentLog)
 		if err != nil {
 			return err
 		}
@@ -78,6 +91,11 @@ func (h *handler) Handle(_ context.Context, r slog.Record) error {
 		if err != nil {
 			return err
 		}
+
+		_, err = h.w.Write([]byte("\n"))
+		if err != nil {
+			return err
+		}
 	}

 	return nil
@@ -88,7 +106,8 @@ func (h *handler) WithAttrs(attrs []slog.Attr) slog.Handler {
 }

 func (h *handler) WithGroup(name string) slog.Handler {
-	panic("unimplemented")
+	h.cmpID = name
+	return h
 }

 func (h *handler) Close() error {
@@ -11,6 +11,7 @@ import (

 	"github.com/absmach/magistrala/pkg/errors"
 	"github.com/stretchr/testify/assert"
+	"github.com/ultravioletrs/cocos/agent/cvms"
 )

 type failedWriter struct{}
@@ -21,14 +22,14 @@ func (f *failedWriter) Write(p []byte) (n int, err error) {

 // TestNewProtoHandler tests the initialization of the ProtoHandler.
 func TestNewProtoHandler(t *testing.T) {
-	handler := NewProtoHandler(io.Discard, nil, "testCmpID")
+	handler := NewProtoHandler(io.Discard, nil, make(chan *cvms.ClientStreamMessage))

 	assert.NotNil(t, handler, "Handler should not be nil")
 }

 // TestHandleMessageSuccess tests the handling of a message when the write succeeds.
 func TestHandleMessageSuccess(t *testing.T) {
-	handler := NewProtoHandler(io.Discard, nil, "testCmpID")
+	handler := NewProtoHandler(io.Discard, nil, make(chan *cvms.ClientStreamMessage, 1))
 	record := slog.Record{
 		Time:    time.Now(),
 		Message: "Test message",
@@ -42,7 +43,7 @@ func TestHandleMessageSuccess(t *testing.T) {

 // TestHandleMessageFailure tests the caching mechanism when the write fails.
 func TestHandleMessageFailure(t *testing.T) {
-	protohandler := NewProtoHandler(&failedWriter{}, nil, "testCmpID")
+	protohandler := NewProtoHandler(&failedWriter{}, nil, make(chan *cvms.ClientStreamMessage, 1))
 	record := slog.Record{
 		Time:    time.Now(),
 		Message: "Test message",
@@ -56,7 +57,7 @@ func TestHandleMessageFailure(t *testing.T) {

 // TestEnabled tests that the handler enables logging based on level.
 func TestEnabled(t *testing.T) {
-	handler := NewProtoHandler(io.Discard, nil, "testCmpID")
+	handler := NewProtoHandler(io.Discard, nil, make(chan *cvms.ClientStreamMessage, 1))

 	assert.True(t, handler.Enabled(context.Background(), slog.LevelInfo), "Logging should be enabled for LevelInfo")
 	assert.False(t, handler.Enabled(context.Background(), slog.LevelDebug), "Logging should be disabled for LevelDebug by default")
@@ -66,7 +67,7 @@ func TestEnabled(t *testing.T) {
 func TestCloseStopsRetry(t *testing.T) {
 	mockWriter := io.Discard

-	handler := NewProtoHandler(mockWriter, nil, "testCmpID").(*handler)
+	handler := NewProtoHandler(mockWriter, nil, make(chan *cvms.ClientStreamMessage, 1)).(*handler)

 	time.Sleep(2 * time.Second)
 	err := handler.Close()