123 lines
3.4 KiB
Go
123 lines
3.4 KiB
Go
package pipeline_api
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"net/http"
|
|
"strings"
|
|
"sync"
|
|
|
|
apiv2 "git.ohea.xyz/cursorius/pipeline-api/go/api/v2"
|
|
"git.ohea.xyz/cursorius/pipeline-api/go/api/v2/apiv2connect"
|
|
"git.ohea.xyz/cursorius/server/runnermanager"
|
|
"github.com/bufbuild/connect-go"
|
|
"github.com/google/uuid"
|
|
"github.com/op/go-logging"
|
|
)
|
|
|
|
var log = logging.MustGetLogger("cursorius-server")
|
|
|
|
type ApiServer struct {
|
|
getRunnerCh chan runnermanager.GetRunnerRequest
|
|
allocatedRunners map[uuid.UUID]*RunnerWrapper
|
|
allocatedRunnersMutex sync.RWMutex
|
|
}
|
|
|
|
type RunnerWrapper struct {
|
|
runner runnermanager.Runner
|
|
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(u uuid.UUID) *RunnerWrapper {
|
|
s.allocatedRunnersMutex.RLock()
|
|
defer s.allocatedRunnersMutex.RUnlock()
|
|
return s.allocatedRunners[u]
|
|
}
|
|
|
|
func (s *ApiServer) GetRunner(
|
|
ctx context.Context,
|
|
req *connect.Request[apiv2.GetRunnerRequest],
|
|
) (*connect.Response[apiv2.GetRunnerResponse], error) {
|
|
|
|
respChan := make(chan runnermanager.GetRunnerResponse)
|
|
s.getRunnerCh <- runnermanager.GetRunnerRequest{
|
|
Tags: req.Msg.Tags,
|
|
RespChan: respChan,
|
|
}
|
|
|
|
var runnerTagsStr strings.Builder
|
|
fmt.Fprintf(&runnerTagsStr, "%v", req.Msg.Tags[0])
|
|
for _, tag := range req.Msg.Tags[1:] {
|
|
fmt.Fprintf(&runnerTagsStr, " %v", tag)
|
|
}
|
|
|
|
response := <-respChan
|
|
if response.Err != nil {
|
|
log.Errorf("Could not get runner with tags \"%v\": %v", runnerTagsStr, response.Err)
|
|
return nil, connect.NewError(connect.CodeNotFound, fmt.Errorf("Could not get runner"))
|
|
}
|
|
|
|
log.Info("Got runner with tags: %v", runnerTagsStr)
|
|
|
|
runnerUuid := uuid.New()
|
|
|
|
s.allocatedRunnersMutex.Lock()
|
|
s.allocatedRunners[runnerUuid] = &RunnerWrapper{runner: response.Runner}
|
|
s.allocatedRunnersMutex.Unlock()
|
|
|
|
res := connect.NewResponse(&apiv2.GetRunnerResponse{
|
|
Id: runnerUuid.String(),
|
|
})
|
|
res.Header().Set("GetRunner-Version", "v2")
|
|
return res, nil
|
|
}
|
|
|
|
func (s *ApiServer) RunCommand(
|
|
ctx context.Context,
|
|
req *connect.Request[apiv2.RunCommandRequest],
|
|
) (*connect.Response[apiv2.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(&apiv2.RunCommandResponse{
|
|
ReturnCode: return_code,
|
|
Stdout: stdout,
|
|
Stderr: stderr,
|
|
})
|
|
res.Header().Set("RunCommand-Version", "v2")
|
|
return res, nil
|
|
}
|
|
|
|
func CreateHandler(mux *http.ServeMux, getRunnerCh chan runnermanager.GetRunnerRequest) {
|
|
api_server := &ApiServer{
|
|
getRunnerCh: getRunnerCh,
|
|
allocatedRunners: make(map[uuid.UUID]*RunnerWrapper),
|
|
}
|
|
path, handler := apiv2connect.NewGetRunnerServiceHandler(api_server)
|
|
mux.Handle(path, handler)
|
|
path, handler = apiv2connect.NewRunCommandServiceHandler(api_server)
|
|
mux.Handle(path, handler)
|
|
}
|