ultravioletrs · dborovcanin · Sep 11, 2024 · Sep 5, 2024 · Sep 5, 2024 · Sep 5, 2024
diff --git a/.github/workflows/checkproto.yaml b/.github/workflows/checkproto.yaml
@@ -29,7 +29,7 @@ jobs:
       - name: Install Go
         uses: actions/setup-go@v5
         with:
-          go-version: 1.22.x
+          go-version: 1.23.x
 
       - name: Set up protoc
         run: |

diff --git a/.github/workflows/hal.yml b/.github/workflows/hal.yml
@@ -18,7 +18,7 @@ jobs:
     - name: Install Go
       uses: actions/setup-go@v5
       with:
-        go-version: 1.22.x
+        go-version: 1.23.x
         cache-dependency-path: "go.sum"
 
     - name: Checkout cocos

diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml
@@ -19,12 +19,12 @@ jobs:
       - name: Install Go
         uses: actions/setup-go@v5
         with:
-          go-version: 1.22.x
+          go-version: 1.23.x
 
       - name: golangci-lint
         uses: golangci/golangci-lint-action@v6
         with:
-          version: v1.59.1
+          version: v1.60
 
       - name: Build
         run: |

diff --git a/.golangci.yaml b/.golangci.yaml
@@ -68,7 +68,7 @@ linters:
     - errchkjson
     - errname
     - execinquery
-    - exportloopref
+    - copyloopvar
     - ginkgolinter
     - gocheckcompilerdirectives
     - gofumpt

diff --git a/agent/api/grpc/interceptors.go b/agent/api/grpc/interceptors.go
@@ -35,20 +35,20 @@ func (s *authInterceptor) AuthStreamInterceptor() grpc.StreamServerInterceptor {
 		switch info.FullMethod {
 		case agent.AgentService_Algo_FullMethodName:
 			if _, err := s.auth.AuthenticateUser(stream.Context(), auth.AlgorithmProviderRole); err != nil {
-				return status.Errorf(codes.Unauthenticated, err.Error())
+				return status.Errorf(codes.Unauthenticated, "%v", err.Error())
 			}
 			return handler(srv, stream)
 		case agent.AgentService_Data_FullMethodName:
 			ctx, err := s.auth.AuthenticateUser(stream.Context(), auth.DataProviderRole)
 			if err != nil {
-				return status.Errorf(codes.Unauthenticated, err.Error())
+				return status.Errorf(codes.Unauthenticated, "%s", err.Error())
 			}
 			wrapped := &wrappedServerStream{ServerStream: stream, ctx: ctx}
 			return handler(srv, wrapped)
 		case agent.AgentService_Result_FullMethodName:
 			ctx, err := s.auth.AuthenticateUser(stream.Context(), auth.ConsumerRole)
 			if err != nil {
-				return status.Errorf(codes.Unauthenticated, err.Error())
+				return status.Errorf(codes.Unauthenticated, "%v", err.Error())
 			}
 			wrapped := &wrappedServerStream{ServerStream: stream, ctx: ctx}
 			return handler(srv, wrapped)
@@ -64,7 +64,7 @@ func (s *authInterceptor) AuthUnaryInterceptor() grpc.UnaryServerInterceptor {
 		case agent.AgentService_Result_FullMethodName:
 			ctx, err := s.auth.AuthenticateUser(ctx, auth.ConsumerRole)
 			if err != nil {
-				return nil, status.Errorf(codes.Unauthenticated, err.Error())
+				return nil, status.Errorf(codes.Unauthenticated, "%v", err.Error())
 			}
 			return handler(ctx, req)
 		default:

diff --git a/cmd/manager/main.go b/cmd/manager/main.go
@@ -57,7 +57,7 @@ func main() {
 
 	logger, err := mglog.New(os.Stdout, cfg.LogLevel)
 	if err != nil {
-		log.Fatalf(err.Error())
+		log.Fatal(err.Error())
 	}
 
 	var exitCode int

diff --git a/go.mod b/go.mod
@@ -1,6 +1,6 @@
 module github.com/ultravioletrs/cocos
 
-go 1.22.4
+go 1.23.0
 
 require (
 	github.com/absmach/magistrala v0.14.1-0.20240709113739-04c359462746

diff --git a/manager/agentEventsLogs.go b/manager/agentEventsLogs.go
@@ -5,17 +5,25 @@ package manager
 import (
 	"fmt"
 	"net"
+	"regexp"
+	"strconv"
 
 	"github.com/mdlayher/vsock"
 	"github.com/ultravioletrs/cocos/pkg/manager"
 	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 const (
 	ManagerVsockPort     = 9997
 	messageSize      int = 1024
 )
 
+var (
+	errFailedToParseCID    = fmt.Errorf("failed to parse computation ID")
+	errComputationNotFound = fmt.Errorf("computation not found")
+)
+
 // RetrieveAgentEventsLogs Retrieve and forward agent logs and events via vsock.
 func (ms *managerService) RetrieveAgentEventsLogs() {
 	l, err := vsock.Listen(ManagerVsockPort, nil)
@@ -42,25 +50,61 @@ func (ms *managerService) handleConnections(conn net.Conn) {
 		n, err := conn.Read(b)
 		if err != nil {
 			ms.logger.Warn(err.Error())
+			cmpID, err := ms.computationIDFromAddress(conn.RemoteAddr().String())
+			if err != nil {
+				ms.logger.Warn(err.Error())
+				continue
+			}
+			go ms.reportBrokenConnection(cmpID)
 			return
 		}
 		var message manager.ClientStreamMessage
 		if err := proto.Unmarshal(b[:n], &message); err != nil {
 			ms.logger.Warn(err.Error())
 			continue
 		}
-		cmpID := ""
-		switch mes := message.Message.(type) {
-		case *manager.ClientStreamMessage_AgentEvent:
-			cmpID = mes.AgentEvent.ComputationId
-			ms.eventsChan <- &manager.ClientStreamMessage{Message: mes}
-		case *manager.ClientStreamMessage_AgentLog:
-			cmpID = mes.AgentLog.ComputationId
-			ms.eventsChan <- &manager.ClientStreamMessage{Message: mes}
-		default:
-			ms.logger.Warn("Unexpected agent log or event type")
+		ms.eventsChan <- &message
+
+		ms.logger.WithGroup("agent-events-logs").Info(message.String())
+	}
+}
+
+func (ms *managerService) computationIDFromAddress(address string) (string, error) {
+	re := regexp.MustCompile(`vm\((\d+)\)`)
+	matches := re.FindStringSubmatch(address)
+
+	if len(matches) > 1 {
+		cid, err := strconv.Atoi(matches[1])
+		if err != nil {
+			return "", err
 		}
+		return ms.findComputationID(cid)
+	}
+	return "", errFailedToParseCID
+}
+
+func (ms *managerService) findComputationID(cid int) (string, error) {
+	ms.mu.Lock()
+	defer ms.mu.Unlock()
+	for cmpID, vm := range ms.vms {
+		if vm.GetCID() == cid {
+			return cmpID, nil
+		}
+	}
+
+	return "", errComputationNotFound
+}
 
-		ms.logger.Info(fmt.Sprintf("Agent Log/Event, Computation ID: %s, Message: %s", cmpID, message.String()))
+func (ms *managerService) reportBrokenConnection(cmpID string) {
+	ms.eventsChan <- &manager.ClientStreamMessage{
+		Message: &manager.ClientStreamMessage_AgentEvent{
+			AgentEvent: &manager.AgentEvent{
+				EventType:     "vm running",
+				ComputationId: cmpID,
+				Status:        "disconnected",
+				Timestamp:     timestamppb.Now(),
+				Originator:    "manager",
+			},
+		},
 	}
 }
diff --git a/manager/qemu/vm.go b/manager/qemu/vm.go
@@ -6,18 +6,22 @@ import (
 	"fmt"
 	"os"
 	"os/exec"
+	"syscall"
+	"time"
 
 	"github.com/gofrs/uuid"
 	"github.com/ultravioletrs/cocos/internal"
 	"github.com/ultravioletrs/cocos/manager/vm"
 	"github.com/ultravioletrs/cocos/pkg/manager"
+	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
 const (
 	firmwareVars = "OVMF_VARS"
 	KernelFile   = "bzImage"
 	rootfsFile   = "rootfs.cpio"
 	tmpDir       = "/tmp"
+	interval     = 5 * time.Second
 )
 
 type qemuVM struct {
@@ -35,7 +39,12 @@ func NewVM(config interface{}, logsChan chan *manager.ClientStreamMessage, compu
 	}
 }
 
-func (v *qemuVM) Start() error {
+func (v *qemuVM) Start() (err error) {
+	defer func() {
+		if err == nil {
+			go v.checkVMProcessPeriodically()
+		}
+	}()
 	// Create unique qemu device identifiers
 	id, err := uuid.NewV4()
 	if err != nil {
@@ -107,3 +116,44 @@ func (v *qemuVM) executableAndArgs() (string, []string, error) {
 
 	return exe, args, nil
 }
+
+func (v *qemuVM) checkVMProcessPeriodically() {
+	for {
+		if !processExists(v.GetProcess()) {
+			v.logsChan <- &manager.ClientStreamMessage{
+				Message: &manager.ClientStreamMessage_AgentEvent{
+					AgentEvent: &manager.AgentEvent{
+						ComputationId: v.computationId,
+						EventType:     "vm-running",
+						Status:        "stopped",
+						Timestamp:     timestamppb.Now(),
+						Originator:    "manager",
+					},
+				},
+			}
+			break
+		}
+		time.Sleep(interval)
+	}
+}
+
+func processExists(pid int) bool {
+	process, err := os.FindProcess(pid)
+	if err != nil {
+		return false
+	}
+
+	// On Unix systems, FindProcess always succeeds and returns a Process for the given pid, regardless of whether the process exists.
+	// To test whether the process actually exists, see whether p.Signal(syscall.Signal(0)) reports an error.
+	if err = process.Signal(syscall.Signal(0)); err == nil {
+		return true
+	}
+	if err == syscall.ESRCH {
+		return false
+	}
+	return false
+}
+
+func (v *qemuVM) GetCID() int {
+	return v.config.GuestCID
+}
diff --git a/manager/vm/logging.go b/manager/vm/logging.go
@@ -120,7 +120,7 @@ func (s *Stderr) Write(p []byte) (n int, err error) {
 				EventType:     "vm-provision",
 				Timestamp:     timestamppb.Now(),
 				Originator:    "manager",
-				Status:        "failed",
+				Status:        "error",
 			},
 		},
 	}

diff --git a/manager/vm/logging_test.go b/manager/vm/logging_test.go
@@ -121,7 +121,7 @@ func TestStderrWrite(t *testing.T) {
 						assert.NotNil(t, agentEvent)
 						assert.Equal(t, "test-computation", agentEvent.ComputationId)
 						assert.Equal(t, "vm-provision", agentEvent.EventType)
-						assert.Equal(t, "failed", agentEvent.Status)
+						assert.Equal(t, "error", agentEvent.Status)
 						assert.NotNil(t, agentEvent.Timestamp)
 					}
 				case <-time.After(time.Second):

diff --git a/manager/vm/mocks/vm.go b/manager/vm/mocks/vm.go
diff --git a/manager/vm/vm.go b/manager/vm/vm.go
@@ -16,6 +16,7 @@ type VM interface {
 	SendAgentConfig(ac agent.Computation) error
 	SetProcess(pid int) error
 	GetProcess() int
+	GetCID() int
 }
 
 //go:generate mockery --name Provider --output=./mocks --filename provider.go --quiet --note "Copyright (c) Ultraviolet \n // SPDX-License-Identifier: Apache-2.0"

diff --git a/test/computations/main.go b/test/computations/main.go
@@ -117,7 +117,7 @@ func main() {
 
 	logger, err := mglog.New(os.Stdout, "debug")
 	if err != nil {
-		log.Fatalf(err.Error())
+		log.Fatal(err.Error())
 	}
 
 	go func() {

diff --git a/test/manual/agent-config/main.go b/test/manual/agent-config/main.go
@@ -38,16 +38,16 @@ func main() {
 
 	pubKey, err := os.ReadFile(pubKeyFile)
 	if err != nil {
-		log.Fatalf(fmt.Sprintf("failed to read public key file: %s", err))
+		log.Fatalf("failed to read public key file: %s", err)
 	}
 	pubPem, _ := pem.Decode(pubKey)
 	algoHash, err := internal.Checksum(algoPath)
 	if err != nil {
-		log.Fatalf(fmt.Sprintf("failed to calculate checksum: %s", err))
+		log.Fatalf("failed to calculate checksum: %s", err)
 	}
 	dataHash, err := internal.Checksum(dataPath)
 	if err != nil {
-		log.Fatalf(fmt.Sprintf("failed to calculate checksum: %s", err))
+		log.Fatalf("failed to calculate checksum: %s", err)
 	}
 
 	l, err := vsock.Listen(manager.ManagerVsockPort, nil)