wip: add analysis runner / agent, separate Server/Queue/Agent, use New* initializers

pkg/agent/agent.go

@@ -1,4 +1,86 @@
 package agent
 
+import (
+	"bufio"
+	"bytes"
+	"log/slog"
+
+	"mrvacommander/pkg/common"
+	"mrvacommander/pkg/queue"
+	"mrvacommander/pkg/storage"
+
+	"os"
+	"os/exec"
+	"path"
+	"strconv"
+	"strings"
+)
+
 type RunnerSingle struct {
+	queue queue.Queue
+}
+
+func NewRunnerSingle(numWorkers int, queue queue.Queue) *RunnerSingle {
+	r := RunnerSingle{queue: queue}
+
+	for id := 1; id <= numWorkers; id++ {
+		go r.worker(id)
+	}
+	return &r
+}
+
+func (r *RunnerSingle) worker(wid int) {
+	var job common.AnalyzeJob
+
+	for {
+		job = <-r.queue.Jobs()
+
+		slog.Debug("Picking up job", "job", job, "worker", wid)
+
+		cwd, err := os.Getwd()
+		if err != nil {
+			slog.Error("RunJob: cwd problem: ", "error", err)
+			continue
+		}
+
+		slog.Debug("Analysis: running", "job", job)
+		storage.SetStatus(job.QueryPackId, job.ORL, common.StatusQueued)
+		cmd := exec.Command(path.Join(cwd, "cmd", "run-analysis.sh"),
+			strconv.FormatInt(int64(job.QueryPackId), 10),
+			job.QueryLanguage, job.ORL.Owner, job.ORL.Repo)
+
+		out, err := cmd.CombinedOutput()
+		if err != nil {
+			slog.Error("Analysis command failed: exit code: ", "error", err, "job", job)
+			slog.Error("Analysis command failed: ", "job", job, "output", out)
+			storage.SetStatus(job.QueryPackId, job.ORL, common.StatusError)
+			continue
+		}
+		slog.Debug("Analysis run finished", "job", job)
+
+		// Get the SARIF ouput location
+		sr := bufio.NewScanner(bytes.NewReader(out))
+		sr.Split(bufio.ScanLines)
+		for {
+			more := sr.Scan()
+			if !more {
+				slog.Error("Analysis run failed to report result: ", "output", out)
+				break
+			}
+			fields := strings.Fields(sr.Text())
+			if len(fields) >= 3 {
+				if fields[0] == "run-analysis-output" {
+					slog.Debug("Analysis run successful: ", "job", job, "location", fields[2])
+					res := common.AnalyzeResult{
+						RunAnalysisSARIF: fields[2], // Abs. path from run-analysis.sh
+						RunAnalysisBQRS:  "",        // FIXME? see note in run-analysis.sh
+					}
+					r.queue.Results() <- res
+					storage.SetStatus(job.QueryPackId, job.ORL, common.StatusSuccess)
+					storage.SetResult(job.QueryPackId, job.ORL, res)
+					break
+				}
+			}
+		}
+	}
 }

pkg/logger/types.go

@@ -2,3 +2,8 @@ package logger
 
 type LoggerSingle struct {
 }
+
+func NewLoggerSingle() *LoggerSingle {
+	l := LoggerSingle{}
+	return &l
+}

pkg/queue/interfaces.go

@@ -1,4 +1,14 @@
 package queue
 
+import (
+	"mrvacommander/pkg/common"
+	"mrvacommander/pkg/storage"
+)
+
 type Queue interface {
+	Jobs() chan common.AnalyzeJob
+	Results() chan common.AnalyzeResult
+	StartAnalyses(analysis_repos *map[common.OwnerRepo]storage.DBLocation,
+		session_id int,
+		session_language string)
 }

pkg/queue/queue.go

@@ -6,13 +6,15 @@ import (
 	"mrvacommander/pkg/storage"
 )
 
-var (
-	NumWorkers int
-	Jobs       chan common.AnalyzeJob
-	Results    chan common.AnalyzeResult
-)
+func (q *QueueSingle) Jobs() chan common.AnalyzeJob {
+	return q.jobs
+}
 
-func StartAnalyses(analysis_repos *map[common.OwnerRepo]storage.DBLocation, session_id int,
+func (q *QueueSingle) Results() chan common.AnalyzeResult {
+	return q.results
+}
+
+func (q *QueueSingle) StartAnalyses(analysis_repos *map[common.OwnerRepo]storage.DBLocation, session_id int,
 	session_language string) {
 	slog.Debug("Queueing codeql database analyze jobs")
 
@@ -23,7 +25,7 @@ func StartAnalyses(analysis_repos *map[common.OwnerRepo]storage.DBLocation, sess
 
 			ORL: orl,
 		}
-		Jobs <- info
+		q.jobs <- info
 		storage.SetStatus(session_id, orl, common.StatusQueued)
 		storage.AddJob(session_id, info)
 	}

pkg/queue/types.go

@@ -1,4 +1,17 @@
 package queue
 
+import "mrvacommander/pkg/common"
+
 type QueueSingle struct {
+	NumWorkers int
+	jobs       chan common.AnalyzeJob
+	results    chan common.AnalyzeResult
+}
+
+func NewQueueSingle(numWorkers int) *QueueSingle {
+	q := QueueSingle{}
+	q.jobs = make(chan common.AnalyzeJob, 10)
+	q.results = make(chan common.AnalyzeResult, 10)
+	q.NumWorkers = numWorkers
+	return &q
 }

pkg/server/server.go

@@ -16,15 +16,11 @@ import (
 	"time"
 
 	"mrvacommander/pkg/common"
-	"mrvacommander/pkg/queue"
 	"mrvacommander/pkg/storage"
 
 	"github.com/gorilla/mux"
 )
 
-func (c *CommanderSingle) Run() {
-}
-
 func (c *CommanderSingle) Setup(st *State) {
 	r := mux.NewRouter()
 	c.st = st
@@ -289,7 +285,7 @@ func (c *CommanderSingle) MirvaRequest(w http.ResponseWriter, r *http.Request) {
 
 	not_found_repos, analysisRepos := c.st.Storage.FindAvailableDBs(session_repositories)
 
-	queue.StartAnalyses(analysisRepos, session_id, session_language)
+	c.queue.StartAnalyses(analysisRepos, session_id, session_language)
 
 	si := SessionInfo{
 		ID:             session_id,

pkg/server/types.go

@@ -26,7 +26,13 @@ type SessionInfo struct {
 }
 
 type CommanderSingle struct {
-	st *State
+	st    *State
+	queue queue.Queue
+}
+
+func NewCommanderSingle(s *State, q queue.Queue) *CommanderSingle {
+	c := CommanderSingle{s, q}
+	return &c
 }
 
 type State struct {

pkg/storage/storage.go

@@ -27,6 +27,11 @@ type StorageSingle struct {
 	CurrentID int
 }
 
+func NewStorageSingle(startingID int) *StorageSingle {
+	s := StorageSingle{CurrentID: startingID}
+	return &s
+}
+
 func (s *StorageSingle) NextID() int {
 	s.CurrentID += 1
 	return s.CurrentID
@@ -121,6 +126,12 @@ func GetResult(js common.JobSpec) common.AnalyzeResult {
 	return ar
 }
 
+func SetResult(sessionid int, orl common.OwnerRepo, ar common.AnalyzeResult) {
+	mutex.Lock()
+	defer mutex.Unlock()
+	result[common.JobSpec{ID: sessionid, OwnerRepo: orl}] = ar
+}
+
 func PackageResults(ar common.AnalyzeResult, owre common.OwnerRepo, vaid int) (zipPath string, e error) {
 	slog.Debug("Readying zip file with .sarif/.bqrs", "analyze-result", ar)