feat: Use UUIDs to identify log streaming jobs (#2051)

* Add UUID for Log Streaming Job ID (#167)

* Update log handler to close buffered  channels when an operation is complete (#170)

* Add preliminary check before registering new receivers in the log handler (#173)

* Using projectOutputBuffers to check for jobID instead of receiverBuffers (#181)

* Refactor log handler  (#175)

* Reverting go.mod and go.sum

* Fix lint errors

* Fix linting

server/controllers/events/events_controller_e2e_test.go

@@ -34,7 +34,7 @@ import (
 	"github.com/runatlantis/atlantis/server/events/vcs"
 	vcsmocks "github.com/runatlantis/atlantis/server/events/vcs/mocks"
 	"github.com/runatlantis/atlantis/server/events/webhooks"
-	handlermocks "github.com/runatlantis/atlantis/server/handlers/mocks"
+	jobmocks "github.com/runatlantis/atlantis/server/jobs/mocks"
 	"github.com/runatlantis/atlantis/server/logging"
 	. "github.com/runatlantis/atlantis/testing"
 )
@@ -836,7 +836,7 @@ func setupE2E(t *testing.T, repoDir string) (events_controllers.VCSEventsControl
 	e2eStatusUpdater := &events.DefaultCommitStatusUpdater{Client: e2eVCSClient}
 	e2eGithubGetter := mocks.NewMockGithubPullGetter()
 	e2eGitlabGetter := mocks.NewMockGitlabMergeRequestGetter()
-	projectCmdOutputHandler := handlermocks.NewMockProjectCommandOutputHandler()
+	projectCmdOutputHandler := jobmocks.NewMockProjectCommandOutputHandler()
 
 	// Real dependencies.
 	logger := logging.NewNoopLogger(t)

server/controllers/jobs_controller.go

@@ -5,17 +5,24 @@ import (
 	"net/http"
 	"net/url"
 
-	"strconv"
-
 	"github.com/gorilla/mux"
-	"github.com/pkg/errors"
 	"github.com/runatlantis/atlantis/server/controllers/templates"
 	"github.com/runatlantis/atlantis/server/controllers/websocket"
 	"github.com/runatlantis/atlantis/server/core/db"
-	"github.com/runatlantis/atlantis/server/events/models"
 	"github.com/runatlantis/atlantis/server/logging"
 )
 
+type JobIDKeyGenerator struct{}
+
+func (g JobIDKeyGenerator) Generate(r *http.Request) (string, error) {
+	jobID, ok := mux.Vars(r)["job-id"]
+	if !ok {
+		return "", fmt.Errorf("internal error: no job-id in route")
+	}
+
+	return jobID, nil
+}
+
 type JobsController struct {
 	AtlantisVersion          string
 	AtlantisURL              *url.URL
@@ -24,105 +31,24 @@ type JobsController struct {
 	ProjectJobsErrorTemplate templates.TemplateWriter
 	Db                       *db.BoltDB
 	WsMux                    *websocket.Multiplexor
-}
-
-type ProjectInfoKeyGenerator struct{}
-
-func (g ProjectInfoKeyGenerator) Generate(r *http.Request) (string, error) {
-	projectInfo, err := newProjectInfo(r)
-
-	if err != nil {
-		return "", errors.Wrap(err, "creating project info")
-	}
-
-	return projectInfo.String(), nil
-}
-
-type pullInfo struct {
-	org  string
-	repo string
-	pull int
-}
-
-func (p *pullInfo) String() string {
-	return fmt.Sprintf("%s/%s/%d", p.org, p.repo, p.pull)
-}
-
-type projectInfo struct {
-	projectName string
-	workspace   string
-	pullInfo
-}
-
-func (p *projectInfo) String() string {
-	return fmt.Sprintf("%s/%s/%d/%s/%s", p.org, p.repo, p.pull, p.projectName, p.workspace)
-}
-
-func newPullInfo(r *http.Request) (*pullInfo, error) {
-	org, ok := mux.Vars(r)["org"]
-	if !ok {
-		return nil, fmt.Errorf("Internal error: no org in route")
-	}
-	repo, ok := mux.Vars(r)["repo"]
-	if !ok {
-		return nil, fmt.Errorf("Internal error: no repo in route")
-	}
-	pull, ok := mux.Vars(r)["pull"]
-	if !ok {
-		return nil, fmt.Errorf("Internal error: no pull in route")
-	}
-	pullNum, err := strconv.Atoi(pull)
-	if err != nil {
-		return nil, err
-	}
-
-	return &pullInfo{
-		org:  org,
-		repo: repo,
-		pull: pullNum,
-	}, nil
-}
-
-// Gets the PR information from the HTTP request params
-func newProjectInfo(r *http.Request) (*projectInfo, error) {
-	pullInfo, err := newPullInfo(r)
-	if err != nil {
-		return nil, err
-	}
-
-	project, ok := mux.Vars(r)["project"]
-	if !ok {
-		return nil, fmt.Errorf("Internal error: no project in route")
-	}
-
-	workspace, ok := mux.Vars(r)["workspace"]
-	if !ok {
-		return nil, fmt.Errorf("Internal error: no workspace in route")
-	}
-
-	return &projectInfo{
-		pullInfo:    *pullInfo,
-		projectName: project,
-		workspace:   workspace,
-	}, nil
+	KeyGenerator             JobIDKeyGenerator
 }
 
 func (j *JobsController) GetProjectJobs(w http.ResponseWriter, r *http.Request) {
-	projectInfo, err := newProjectInfo(r)
+	jobID, err := j.KeyGenerator.Generate(r)
+
 	if err != nil {
-		j.respond(w, logging.Error, http.StatusInternalServerError, err.Error())
+		j.respond(w, logging.Error, http.StatusBadRequest, err.Error())
 		return
 	}
 
 	viewData := templates.ProjectJobData{
 		AtlantisVersion: j.AtlantisVersion,
-		ProjectPath:     projectInfo.String(),
+		ProjectPath:     jobID,
 		CleanedBasePath: j.AtlantisURL.Path,
-		ClearMsg:        models.LogStreamingClearMsg,
 	}
 
-	err = j.ProjectJobsTemplate.Execute(w, viewData)
-	if err != nil {
+	if err = j.ProjectJobsTemplate.Execute(w, viewData); err != nil {
 		j.Logger.Err(err.Error())
 	}
 }
@@ -131,7 +57,7 @@ func (j *JobsController) GetProjectJobsWS(w http.ResponseWriter, r *http.Request
 	err := j.WsMux.Handle(w, r)
 
 	if err != nil {
-		j.respond(w, logging.Error, http.StatusInternalServerError, err.Error())
+		j.respond(w, logging.Error, http.StatusBadRequest, err.Error())
 		return
 	}
 }

server/controllers/templates/web_templates.go

@@ -357,7 +357,6 @@ type ProjectJobData struct {
 	AtlantisVersion string
 	ProjectPath     string
 	CleanedBasePath string
-	ClearMsg        string
 }
 
 var ProjectJobsTemplate = template.Must(template.New("blank.html.tmpl").Parse(`
@@ -417,13 +416,9 @@ var ProjectJobsTemplate = template.Must(template.New("blank.html.tmpl").Parse(`
         document.location.host +
         document.location.pathname +
         "/ws");
-      socket.onmessage = function(event) {
-        var msg = String.fromCharCode.apply(null, new Uint8Array(event.data))
-        if (msg.trim() === "-----Starting New Process-----") {
-          term.clear()
-          return 
-        }
-      }
+      window.addEventListener("unload", function(event) {
+        websocket.close();
+      })
       var attachAddon = new AttachAddon.AttachAddon(socket);
       var fitAddon = new FitAddon.FitAddon();
       term.loadAddon(attachAddon);

server/controllers/websocket/mux.go

@@ -1,6 +1,7 @@
 package websocket
 
 import (
+	"fmt"
 	"net/http"
 
 	"github.com/gorilla/websocket"
@@ -18,6 +19,7 @@ type PartitionKeyGenerator interface {
 type PartitionRegistry interface {
 	Register(key string, buffer chan string)
 	Deregister(key string, buffer chan string)
+	IsKeyExists(key string) bool
 }
 
 // Multiplexor is responsible for handling the data transfer between the storage layer
@@ -51,6 +53,11 @@ func (m *Multiplexor) Handle(w http.ResponseWriter, r *http.Request) error {
 		return errors.Wrapf(err, "generating partition key")
 	}
 
+	// check if the job ID exists before registering receiver
+	if !m.registry.IsKeyExists(key) {
+		return fmt.Errorf("invalid key: %s", key)
+	}
+
 	// Buffer size set to 1000 to ensure messages get queued.
 	// TODO: make buffer size configurable
 	buffer := make(chan string, 1000)

server/controllers/websocket/writer.go

@@ -29,18 +29,6 @@ func (w *Writer) Write(rw http.ResponseWriter, r *http.Request, input chan strin
 		return errors.Wrap(err, "upgrading websocket connection")
 	}
 
-	conn.SetCloseHandler(func(code int, text string) error {
-		// Close the channnel after websocket connection closed.
-		// Will gracefully exit the ProjectCommandOutputHandler.Register() call and cleanup.
-		// is it good practice to close at the receiver? Probably not, we should figure out a better
-		// way to handle this case
-		close(input)
-		return nil
-	})
-
-	// Add a reader goroutine to listen for socket.close() events.
-	go w.setReadHandler(conn)
-
 	// block on reading our input channel
 	for msg := range input {
 		if err := conn.WriteMessage(websocket.BinaryMessage, []byte("\r"+msg+"\n")); err != nil {
@@ -49,20 +37,9 @@ func (w *Writer) Write(rw http.ResponseWriter, r *http.Request, input chan strin
 		}
 	}
 
-	return nil
-}
-
-func (w *Writer) setReadHandler(c *websocket.Conn) {
-	for {
-		_, _, err := c.ReadMessage()
-		if err != nil {
-			// CloseGoingAway (1001) when a browser tab is closed.
-			// Expected behaviour since we have a CloseHandler(), log warning if not a CloseGoingAway
-			if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway) {
-				w.log.Warn("Failed to read WS message: %s", err)
-			}
-			return
-		}
+	// close ws conn after input channel is closed
+	if err = conn.Close(); err != nil {
+		w.log.Warn("Failed to close ws connection: %s", err)
 	}
-
+	return nil
 }

server/core/terraform/terraform_client.go

@@ -33,7 +33,7 @@ import (
 
 	"github.com/runatlantis/atlantis/server/events/models"
 	"github.com/runatlantis/atlantis/server/events/terraform/ansi"
-	"github.com/runatlantis/atlantis/server/handlers"
+	"github.com/runatlantis/atlantis/server/jobs"
 	"github.com/runatlantis/atlantis/server/logging"
 )
 
@@ -79,7 +79,7 @@ type DefaultClient struct {
 	// usePluginCache determines whether or not to set the TF_PLUGIN_CACHE_DIR env var
 	usePluginCache bool
 
-	projectCmdOutputHandler handlers.ProjectCommandOutputHandler
+	projectCmdOutputHandler jobs.ProjectCommandOutputHandler
 }
 
 //go:generate pegomock generate -m --use-experimental-model-gen --package mocks -o mocks/mock_downloader.go Downloader
@@ -111,7 +111,7 @@ func NewClientWithDefaultVersion(
 	tfDownloader Downloader,
 	usePluginCache bool,
 	fetchAsync bool,
-	projectCmdOutputHandler handlers.ProjectCommandOutputHandler,
+	projectCmdOutputHandler jobs.ProjectCommandOutputHandler,
 ) (*DefaultClient, error) {
 	var finalDefaultVersion *version.Version
 	var localVersion *version.Version
@@ -194,7 +194,7 @@ func NewTestClient(
 	tfDownloadURL string,
 	tfDownloader Downloader,
 	usePluginCache bool,
-	projectCmdOutputHandler handlers.ProjectCommandOutputHandler,
+	projectCmdOutputHandler jobs.ProjectCommandOutputHandler,
 ) (*DefaultClient, error) {
 	return NewClientWithDefaultVersion(
 		log,
@@ -231,7 +231,7 @@ func NewClient(
 	tfDownloadURL string,
 	tfDownloader Downloader,
 	usePluginCache bool,
-	projectCmdOutputHandler handlers.ProjectCommandOutputHandler,
+	projectCmdOutputHandler jobs.ProjectCommandOutputHandler,
 ) (*DefaultClient, error) {
 	return NewClientWithDefaultVersion(
 		log,
@@ -437,7 +437,7 @@ func (c *DefaultClient) RunCommandAsync(ctx models.ProjectCommandContext, path s
 			for s.Scan() {
 				message := s.Text()
 				outCh <- Line{Line: message}
-				c.projectCmdOutputHandler.Send(ctx, message)
+				c.projectCmdOutputHandler.Send(ctx, message, false)
 			}
 			wg.Done()
 		}()
@@ -446,7 +446,7 @@ func (c *DefaultClient) RunCommandAsync(ctx models.ProjectCommandContext, path s
 			for s.Scan() {
 				message := s.Text()
 				outCh <- Line{Line: message}
-				c.projectCmdOutputHandler.Send(ctx, message)
+				c.projectCmdOutputHandler.Send(ctx, message, false)
 			}
 			wg.Done()
 		}()

server/core/terraform/terraform_client_internal_test.go

@@ -9,7 +9,7 @@ import (
 
 	version "github.com/hashicorp/go-version"
 	"github.com/runatlantis/atlantis/server/events/models"
-	handlermocks "github.com/runatlantis/atlantis/server/handlers/mocks"
+	jobmocks "github.com/runatlantis/atlantis/server/jobs/mocks"
 	"github.com/runatlantis/atlantis/server/logging"
 	. "github.com/runatlantis/atlantis/testing"
 )
@@ -91,7 +91,7 @@ func TestDefaultClient_RunCommandWithVersion_EnvVars(t *testing.T) {
 	Ok(t, err)
 	tmp, cleanup := TempDir(t)
 	logger := logging.NewNoopLogger(t)
-	projectCmdOutputHandler := handlermocks.NewMockProjectCommandOutputHandler()
+	projectCmdOutputHandler := jobmocks.NewMockProjectCommandOutputHandler()
 
 	ctx := models.ProjectCommandContext{
 		Log:                logger,
@@ -138,7 +138,7 @@ func TestDefaultClient_RunCommandWithVersion_Error(t *testing.T) {
 	Ok(t, err)
 	tmp, cleanup := TempDir(t)
 	logger := logging.NewNoopLogger(t)
-	projectCmdOutputHandler := handlermocks.NewMockProjectCommandOutputHandler()
+	projectCmdOutputHandler := jobmocks.NewMockProjectCommandOutputHandler()
 
 	ctx := models.ProjectCommandContext{
 		Log:                logger,
@@ -181,7 +181,7 @@ func TestDefaultClient_RunCommandAsync_Success(t *testing.T) {
 	Ok(t, err)
 	tmp, cleanup := TempDir(t)
 	logger := logging.NewNoopLogger(t)
-	projectCmdOutputHandler := handlermocks.NewMockProjectCommandOutputHandler()
+	projectCmdOutputHandler := jobmocks.NewMockProjectCommandOutputHandler()
 
 	ctx := models.ProjectCommandContext{
 		Log:                logger,
@@ -228,7 +228,7 @@ func TestDefaultClient_RunCommandAsync_BigOutput(t *testing.T) {
 	Ok(t, err)
 	tmp, cleanup := TempDir(t)
 	logger := logging.NewNoopLogger(t)
-	projectCmdOutputHandler := handlermocks.NewMockProjectCommandOutputHandler()
+	projectCmdOutputHandler := jobmocks.NewMockProjectCommandOutputHandler()
 
 	ctx := models.ProjectCommandContext{
 		Log:                logger,
@@ -276,7 +276,7 @@ func TestDefaultClient_RunCommandAsync_StderrOutput(t *testing.T) {
 	Ok(t, err)
 	tmp, cleanup := TempDir(t)
 	logger := logging.NewNoopLogger(t)
-	projectCmdOutputHandler := handlermocks.NewMockProjectCommandOutputHandler()
+	projectCmdOutputHandler := jobmocks.NewMockProjectCommandOutputHandler()
 
 	ctx := models.ProjectCommandContext{
 		Log:                logger,
@@ -313,7 +313,7 @@ func TestDefaultClient_RunCommandAsync_ExitOne(t *testing.T) {
 	Ok(t, err)
 	tmp, cleanup := TempDir(t)
 	logger := logging.NewNoopLogger(t)
-	projectCmdOutputHandler := handlermocks.NewMockProjectCommandOutputHandler()
+	projectCmdOutputHandler := jobmocks.NewMockProjectCommandOutputHandler()
 
 	ctx := models.ProjectCommandContext{
 		Log:                logger,
@@ -351,7 +351,7 @@ func TestDefaultClient_RunCommandAsync_Input(t *testing.T) {
 	Ok(t, err)
 	tmp, cleanup := TempDir(t)
 	logger := logging.NewNoopLogger(t)
-	projectCmdOutputHandler := handlermocks.NewMockProjectCommandOutputHandler()
+	projectCmdOutputHandler := jobmocks.NewMockProjectCommandOutputHandler()
 
 	ctx := models.ProjectCommandContext{
 		Log:                logger,