Implement skeleton of RunCommand

pipeline_api/pipeline_api.go

@@ -18,8 +18,9 @@ import (
 var log = logging.MustGetLogger("cursorius-server")
 
 type ApiServer struct {
-	getRunnerCh      chan runnermanager.GetRunnerRequest
-	allocatedRunners map[uuid.UUID]RunnerWrapper
+	getRunnerCh           chan runnermanager.GetRunnerRequest
+	allocatedRunners      map[uuid.UUID]*RunnerWrapper
+	allocatedRunnersMutex sync.RWMutex
 }
 
 type RunnerWrapper struct {
@@ -27,6 +28,23 @@ type RunnerWrapper struct {
 	mutex  sync.Mutex
 }
 
+func (r *RunnerWrapper) RunCommand(cmd string) (int64, string, string, error) {
+	r.mutex.Unlock()
+	defer r.mutex.Lock()
+
+	return_code, stdout, stderr, err := r.runner.RunCommand(cmd)
+
+	// TODO: run command by sending websocket packet
+	// TODO: get stdout and stderr response
+	return return_code, stdout, stderr, err
+}
+
+func (s *ApiServer) GetRunnerFromMap(uuid uuid) *RunnerWrapper {
+	s.allocatedRunnersMutex.RLock()
+	defer s.allocatedRunnersMutex.RUnlock()
+	return s.allocatedRunners[uuid]
+}
+
 func (s *ApiServer) GetRunner(
 	ctx context.Context,
 	req *connect.Request[apiv1.GetRunnerRequest],
@@ -54,7 +72,9 @@ func (s *ApiServer) GetRunner(
 
 	runnerUuid := uuid.New()
 
-	s.allocatedRunners[runnerUuid] = RunnerWrapper{runner: response.Runner}
+	s.allocatedRunnersMutex.Lock()
+	s.allocatedRunners[runnerUuid] = &RunnerWrapper{runner: response.Runner}
+	s.allocatedRunnersMutex.Unlock()
 
 	res := connect.NewResponse(&apiv1.GetRunnerResponse{
 		Id: runnerUuid.String(),
@@ -68,10 +88,23 @@ func (s *ApiServer) RunCommand(
 	req *connect.Request[apiv1.RunCommandRequest],
 ) (*connect.Response[apiv1.RunCommandResponse], error) {
 
+	uuid, err := uuid.Parse(req.Msg.Id)
+	if err != nil {
+		return nil, connect.NewError(connect.CodeInvalidArgument, fmt.Errorf("Invalid runner id"))
+	}
+
+	runner := s.GetRunnerFromMap(uuid)
+
+	return_code, stdout, stderr, err := runner.RunCommand(req.Msg.Command)
+	if err != nil {
+		log.Errorf("Could not run command on runner \"%v\", %v", runner.runner.Id(), err)
+		return nil, connect.NewError(connect.CodeInternal, fmt.Errorf("Could not run command"))
+	}
+
 	res := connect.NewResponse(&apiv1.RunCommandResponse{
-		ReturnCode: 0,
-		Stdout:     "",
-		Stderr:     "",
+		ReturnCode: return_code,
+		Stdout:     stdout,
+		Stderr:     stderr,
 	})
 	res.Header().Set("RunCommand-Version", "v1")
 	return res, nil
@@ -80,7 +113,7 @@ func (s *ApiServer) RunCommand(
 func CreateHandler(mux *http.ServeMux, getRunnerCh chan runnermanager.GetRunnerRequest) {
 	api_server := &ApiServer{
 		getRunnerCh:      getRunnerCh,
-		allocatedRunners: make(map[uuid.UUID]RunnerWrapper),
+		allocatedRunners: make(map[uuid.UUID]*RunnerWrapper),
 	}
 	path, handler := apiv1connect.NewGetRunnerServiceHandler(api_server)
 	mux.Handle(path, handler)