live status update

core/job_linux.go

@@ -29,6 +29,7 @@ type Project struct {
 
 // result processing is local to individual job
 // result processing is only started after a job has been started
+// map key in resultSyncMap is projectID, the reason behind this is to have independent project build results
 
 type Result struct {
 	Error       error  `json:"error"`
@@ -41,13 +42,6 @@ type JobState struct {
 	LastBuildStart time.Time `json:"lastBuildStart"`
 	StepResults    []Result  `json:"stepResults"`
 	BuildStatus    string    `json:"buildStatus"`
-	Coverage       float64   `json:"coverage"`
-}
-
-type StatusResponse struct {
-	JobState
-	ProjectName    string `json:"projectName"`
-	DateTimeString string `json:"dateTimeString"`
 }
 
 type ResultSyncMap struct {
@@ -66,6 +60,7 @@ var Queues jobqueue.QueueMap
 var ServerConf Doc
 var ResultMap *ResultSyncMap
 var Ctx context.Context
+var LiveResultUpdates map[string]chan JobState
 
 // function that will be enqued by project specific queue
 // DONE: make this func fit into queue job function prototype
@@ -86,6 +81,11 @@ func Job(args ...any) error {
 	}
 	ResultMap.Mu.Unlock()
 
+	// draining the live result channel whenever a new build starts
+	for len(LiveResultUpdates[projectName]) > 0 {
+		<-LiveResultUpdates[projectName]
+	}
+
 	for _, step := range project.Steps {
 
 		if len(step) == 0 {
@@ -139,6 +139,14 @@ func Job(args ...any) error {
 				BuildStatus:    status,
 			}
 			ResultMap.Mu.Unlock()
+
+			// updating the live status
+			LiveResultUpdates[projectName] <- JobState{
+				LastBuildStart: buildTime,
+				StepResults:    steps,
+				BuildStatus:    status,
+			}
+
 		} else {
 			ResultMap.Mu.RUnlock()
 		}
@@ -152,10 +160,15 @@ func Job(args ...any) error {
 			ResultMap.Mu.Lock()
 			ResultMap.Map[projectName] = obj
 			ResultMap.Mu.Unlock()
+
+			// updating the live status
+			LiveResultUpdates[projectName] <- obj
 			break
-		} else {
-			fmt.Printf("step %v done\n", step)
 		}
+		// } else {
+		// 	// LOG: log here when some leveled logger is integrated
+		// 	// fmt.Printf("step %v done\n", step)
+		// }
 	}
 	ResultMap.Mu.RLock()
 	obj := ResultMap.Map[projectName]
@@ -165,6 +178,10 @@ func Job(args ...any) error {
 		ResultMap.Mu.Lock()
 		ResultMap.Map[projectName] = obj
 		ResultMap.Mu.Unlock()
+
+		// updating the live status
+		LiveResultUpdates[projectName] <- obj
+
 	}
 	return nil
 }
@@ -190,18 +207,21 @@ func ServerInit(configlocation string, l *log.Logger, wg *sync.WaitGroup) error
 	}
 	ServerConf = conf
 	Queues = make(jobqueue.QueueMap, 0)
-	for projectName := range ServerConf.Project {
+	LiveResultUpdates = make(map[string]chan JobState)
+	for projectName, project := range ServerConf.Project {
 		jg := jobqueue.NewJobQueue(projectName, make(chan jobqueue.Job, 25), 1)
 		err = Queues.Register(jg)
 		if err != nil {
 			return err
 		}
+		LiveResultUpdates[projectName] = make(chan JobState, len(project.Steps)+1)
 
 	}
 	Queues.StartAll(l, wg)
 	ResultMap = &ResultSyncMap{
 		Map: make(map[string]JobState),
 	}
+
 	Ctx = context.Background()
 	return nil
 }

core/job_windows.go

@@ -29,6 +29,7 @@ type Project struct {
 
 // result processing is local to individual job
 // result processing is only started after a job has been started
+// map key in resultSyncMap is projectID, the reason behind this is to have independent project build results
 
 type Result struct {
 	Error       error  `json:"error"`
@@ -41,14 +42,8 @@ type JobState struct {
 	LastBuildStart time.Time `json:"lastBuildStart"`
 	StepResults    []Result  `json:"stepResults"`
 	BuildStatus    string    `json:"buildStatus"`
-	Coverage       float64   `json:"coverage"`
 }
 
-type StatusResponse struct {
-	JobState
-	ProjectName    string `json:"projectName"`
-	DateTimeString string `json:"dateTimeString"`
-}
 
 type ResultSyncMap struct {
 	Mu  sync.RWMutex
@@ -66,6 +61,7 @@ var Queues jobqueue.QueueMap
 var ServerConf Doc
 var ResultMap *ResultSyncMap
 var Ctx context.Context
+var LiveResultUpdates map[string]chan JobState
 
 // function that will be enqued by project specific queue
 // DONE: make this func fit into queue job function prototype
@@ -86,6 +82,11 @@ func Job(args ...any) error {
 	}
 	ResultMap.Mu.Unlock()
 
+	// draining the live result channel whenever a new build starts
+	for len(LiveResultUpdates[projectName]) > 0 {
+		<-LiveResultUpdates[projectName]
+	}
+
 	for _, step := range project.Steps {
 
 		if len(step) == 0 {
@@ -139,6 +140,14 @@ func Job(args ...any) error {
 				BuildStatus:    status,
 			}
 			ResultMap.Mu.Unlock()
+
+			// updating the live status
+			LiveResultUpdates[projectName] <- JobState{
+				LastBuildStart: buildTime,
+				StepResults:    steps,
+				BuildStatus:    status,
+			}
+
 		} else {
 			ResultMap.Mu.RUnlock()
 		}
@@ -152,10 +161,15 @@ func Job(args ...any) error {
 			ResultMap.Mu.Lock()
 			ResultMap.Map[projectName] = obj
 			ResultMap.Mu.Unlock()
+
+			// updating the live status
+			LiveResultUpdates[projectName] <- obj
 			break
-		} else {
-			fmt.Printf("step %v done\n", step)
 		}
+		// } else {
+		// 	// LOG: log here when some leveled logger is integrated
+		// 	// fmt.Printf("step %v done\n", step)
+		// }
 	}
 	ResultMap.Mu.RLock()
 	obj := ResultMap.Map[projectName]
@@ -165,6 +179,10 @@ func Job(args ...any) error {
 		ResultMap.Mu.Lock()
 		ResultMap.Map[projectName] = obj
 		ResultMap.Mu.Unlock()
+
+		// updating the live status
+		LiveResultUpdates[projectName] <- obj
+
 	}
 	return nil
 }
@@ -190,18 +208,21 @@ func ServerInit(configlocation string, l *log.Logger, wg *sync.WaitGroup) error
 	}
 	ServerConf = conf
 	Queues = make(jobqueue.QueueMap, 0)
-	for projectName := range ServerConf.Project {
+	LiveResultUpdates = make(map[string]chan JobState)
+	for projectName, project := range ServerConf.Project {
 		jg := jobqueue.NewJobQueue(projectName, make(chan jobqueue.Job, 25), 1)
 		err = Queues.Register(jg)
 		if err != nil {
 			return err
 		}
+		LiveResultUpdates[projectName] = make(chan JobState, len(project.Steps)+1)
 
 	}
 	Queues.StartAll(l, wg)
 	ResultMap = &ResultSyncMap{
 		Map: make(map[string]JobState),
 	}
+
 	Ctx = context.Background()
 	return nil
 }

example.toml

@@ -2,12 +2,17 @@
 
 branch = "master"
 secret = "xxx"
-cwd = '/home/neelu/experiments'
+cwd = 'c:\games'
 steps = [
-    ["echo","start"],
-    ["sleep","2"],
-    ["echo","mid"],
-    ["sleep","10"],
-    ["echo","end"]
+    ["goprint","start"],
+    ["gosleep","2"],
+    ["goprint","mid"],
+    ["gosleep","5"],
+    ["goprint","end"],
+    ["gosleep","1"],
+    ["gosleep","1"],
+    ["gosleep","1"],
+    ["gosleep","1"],
+    ["gosleep","1"],
 ]
 stepTimeout = 600

go.mod

@@ -9,4 +9,7 @@ require (
 	github.com/gorilla/mux v1.8.0
 )
 
-require github.com/felixge/httpsnoop v1.0.1 // indirect
+require (
+	github.com/felixge/httpsnoop v1.0.1 // indirect
+	github.com/gorilla/websocket v1.5.0 // indirect
+)

go.sum

@@ -6,3 +6,5 @@ github.com/gorilla/handlers v1.5.1 h1:9lRY6j8DEeeBT10CvO9hGW0gmky0BprnvDI5vfhUHH
 github.com/gorilla/handlers v1.5.1/go.mod h1:t8XrUpc4KVXb7HGyJ4/cEnwQiaxrX/hz1Zv/4g96P1Q=
 github.com/gorilla/mux v1.8.0 h1:i40aqfkR1h2SlN9hojwV5ZA91wcXFOvkdNIeFDP5koI=
 github.com/gorilla/mux v1.8.0/go.mod h1:DVbg23sWSpFRCP0SfiEN6jmj59UnW/n46BH5rLB71So=
+github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
+github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=

httpinterface/api.go

@@ -10,6 +10,7 @@ import (
 	"ghhooks.com/hook/core"
 	"ghhooks.com/hook/jobqueue"
 	"github.com/gorilla/mux"
+	"github.com/gorilla/websocket"
 )
 
 // DONE: verify signature before accepting webhook
@@ -24,6 +25,26 @@ import (
 // TODO: blocking build run
 // DONE: html page for status
 // TODO: maybe put password on status page to prevent from builds being cancelled by just anyone
+// TODO: update progressbar using websockets,
+// TODO: individual step results on statuspage
+
+var upgrader = websocket.Upgrader{
+	ReadBufferSize:  1024,
+	WriteBufferSize: 1024,
+}
+
+type StatusResponse struct {
+	core.JobState
+	ProjectName    string  `json:"projectName"`
+	DateTimeString string  `json:"dateTimeString"`
+	Coverage       float64 `json:"coverage"`
+}
+
+type WebsocketResponse struct {
+	core.JobState
+	Coverage    float64 `json:"coverage"`
+	ProjectName string  `json:"projectName"`
+}
 
 func WebHookListener(w http.ResponseWriter, r *http.Request) {
 
@@ -157,7 +178,7 @@ func BuildStatus(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	result.Coverage = float64(successfullSteps * 100 / totalSteps)
+	coverage := float64(successfullSteps * 100 / totalSteps)
 
 	format := r.URL.Query().Get("format")
 
@@ -166,20 +187,77 @@ func BuildStatus(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	templateResponse := core.StatusResponse{
+	templateResponse := StatusResponse{
 		JobState:       result,
 		ProjectName:    projectID,
 		DateTimeString: result.LastBuildStart.Format(time.RFC3339),
+		Coverage:       coverage,
 	}
 
 	tmpl := template.Must(template.ParseFiles("statuspage.html"))
 	tmpl.Execute(w, templateResponse)
 
 }
 
+func LiveStatusUpdate(w http.ResponseWriter, r *http.Request) {
+
+	vars := mux.Vars(r)
+	projectID, ok := vars["project"]
+	if !ok {
+		Respond(w, 400, map[string]interface{}{
+			"error": "no vars found",
+		})
+		return
+	}
+
+	project, ok := core.ServerConf.Project[projectID]
+	if !ok {
+		Respond(w, 400, map[string]interface{}{
+			"error": "no project found with given project name",
+		})
+		return
+	}
+
+	totalSteps := len(project.Steps)
+
+	conn, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		Respond(w, 500, map[string]interface{}{
+			"error": err.Error(),
+		})
+		return
+	}
+	defer conn.Close()
+
+	for jobState := range core.LiveResultUpdates[projectID] {
+
+		var successfullSteps int
+		for _, v := range jobState.StepResults {
+			if v.Error == nil {
+				successfullSteps = successfullSteps + 1
+			}
+		}
+
+		coverage := float64(successfullSteps * 100 / totalSteps)
+
+		res := WebsocketResponse{
+			JobState:    jobState,
+			ProjectName: projectID,
+			Coverage:    coverage,
+		}
+		err := conn.WriteJSON(res)
+		if err != nil {
+			break
+		}
+	}
+
+}
+
 func RouterInit(r *mux.Router) {
 	r.HandleFunc("/{project}", WebHookListener).Methods("POST")
 	r.HandleFunc("/{project}/", WebHookListener).Methods("POST")
 	r.HandleFunc("/{project}/status", BuildStatus).Methods("GET")
 	r.HandleFunc("/{project}/status/", BuildStatus).Methods("GET")
+	r.HandleFunc("/{project}/livestatus", LiveStatusUpdate)
+	r.HandleFunc("/{project}/livestatus/", LiveStatusUpdate)
 }