coder · kylecarbs · Jan 29, 2022 · Jan 29, 2022 · Jan 29, 2022 · Jan 29, 2022
diff --git a/coderd/coderd.go b/coderd/coderd.go
@@ -16,18 +16,25 @@ import (
 type Options struct {
 	Logger   slog.Logger
 	Database database.Store
+	Pubsub   database.Pubsub
 }
 
 // New constructs the Coder API into an HTTP handler.
 func New(options *Options) http.Handler {
 	projects := &projects{
 		Database: options.Database,
+		Pubsub:   options.Pubsub,
+	}
+	provisioners := &provisioners{
+		Database: options.Database,
+		Pubsub:   options.Pubsub,
 	}
 	users := &users{
 		Database: options.Database,
 	}
 	workspaces := &workspaces{
 		Database: options.Database,
+		Pubsub:   options.Pubsub,
 	}
 
 	r := chi.NewRouter()
@@ -39,6 +46,8 @@ func New(options *Options) http.Handler {
 		})
 		r.Post("/login", users.loginWithPassword)
 		r.Post("/logout", users.logout)
+		r.Get("/provisionerd", provisioners.listen)
+
 		// Used for setup.
 		r.Post("/user", users.createInitialUser)
 		r.Route("/users", func(r chi.Router) {
@@ -67,6 +76,10 @@ func New(options *Options) http.Handler {
 					r.Route("/history", func(r chi.Router) {
 						r.Get("/", projects.allProjectHistory)
 						r.Post("/", projects.createProjectHistory)
+						r.Route("/{projecthistory}", func(r chi.Router) {
+							r.Use(httpmw.ExtractProjectHistoryParam(options.Database))
+							r.Get("/logs", projects.projectHistoryLogs)
+						})
 					})
 					r.Get("/workspaces", workspaces.allWorkspacesForProject)
 				})
@@ -89,10 +102,18 @@ func New(options *Options) http.Handler {
 						r.Post("/", workspaces.createWorkspaceHistory)
 						r.Get("/", workspaces.listAllWorkspaceHistory)
 						r.Get("/latest", workspaces.latestWorkspaceHistory)
+						r.Route("/{workspacehistory}", func(r chi.Router) {
+							r.Use(httpmw.ExtractWorkspaceHistoryParam(options.Database))
+							r.Get("/logs", workspaces.workspaceHistoryLogs)
+						})
 					})
 				})
 			})
 		})
+
+		r.Route("/provisioners", func(r chi.Router) {
+			r.Get("/daemons", provisioners.listDaemons)
+		})
 	})
 	r.NotFound(site.Handler().ServeHTTP)
 	return r

diff --git a/coderd/coderdtest/coderdtest.go b/coderd/coderdtest/coderdtest.go
@@ -3,20 +3,27 @@ package coderdtest
 import (
 	"context"
 	"database/sql"
+	"io"
 	"net/http/httptest"
 	"net/url"
 	"os"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
+	"cdr.dev/slog"
 	"cdr.dev/slog/sloggers/slogtest"
 	"github.com/coder/coder/coderd"
 	"github.com/coder/coder/codersdk"
 	"github.com/coder/coder/cryptorand"
 	"github.com/coder/coder/database"
 	"github.com/coder/coder/database/databasefake"
 	"github.com/coder/coder/database/postgres"
+	"github.com/coder/coder/provisioner/terraform"
+	"github.com/coder/coder/provisionerd"
+	"github.com/coder/coder/provisionersdk"
+	"github.com/coder/coder/provisionersdk/proto"
 )
 
 // Server represents a test instance of coderd.
@@ -57,11 +64,44 @@ func (s *Server) RandomInitialUser(t *testing.T) coderd.CreateInitialUserRequest
 	return req
 }
 
+// AddProvisionerd launches a new provisionerd instance!
+func (s *Server) AddProvisionerd(t *testing.T) io.Closer {
+	tfClient, tfServer := provisionersdk.TransportPipe()
+	ctx, cancelFunc := context.WithCancel(context.Background())
+	t.Cleanup(func() {
+		_ = tfClient.Close()
+		_ = tfServer.Close()
+		cancelFunc()
+	})
+	go func() {
+		err := terraform.Serve(ctx, &terraform.ServeOptions{
+			ServeOptions: &provisionersdk.ServeOptions{
+				Listener: tfServer,
+			},
+		})
+		require.NoError(t, err)
+	}()
+
+	closer := provisionerd.New(s.Client.ProvisionerDaemonClient, &provisionerd.Options{
+		Logger:       slogtest.Make(t, nil).Named("provisionerd").Leveled(slog.LevelInfo),
+		PollInterval: 50 * time.Millisecond,
+		Provisioners: provisionerd.Provisioners{
+			string(database.ProvisionerTypeTerraform): proto.NewDRPCProvisionerClient(provisionersdk.Conn(tfClient)),
+		},
+		WorkDirectory: t.TempDir(),
+	})
+	t.Cleanup(func() {
+		_ = closer.Close()
+	})
+	return closer
+}
+
 // New constructs a new coderd test instance. This returned Server
 // should contain no side-effects.
 func New(t *testing.T) Server {
 	// This can be hotswapped for a live database instance.
 	db := databasefake.New()
+	pubsub := database.NewPubsubInMemory()
 	if os.Getenv("DB") != "" {
 		connectionURL, close, err := postgres.Open()
 		require.NoError(t, err)
@@ -74,11 +114,15 @@ func New(t *testing.T) Server {
 		err = database.Migrate(sqlDB)
 		require.NoError(t, err)
 		db = database.New(sqlDB)
+
+		pubsub, err = database.NewPubsub(context.Background(), sqlDB, connectionURL)
+		require.NoError(t, err)
 	}
 
 	handler := coderd.New(&coderd.Options{
 		Logger:   slogtest.Make(t, nil),
 		Database: db,
+		Pubsub:   pubsub,
 	})
 	srv := httptest.NewServer(handler)
 	serverURL, err := url.Parse(srv.URL)

diff --git a/coderd/coderdtest/coderdtest_test.go b/coderd/coderdtest/coderdtest_test.go
@@ -16,4 +16,5 @@ func TestNew(t *testing.T) {
 	t.Parallel()
 	server := coderdtest.New(t)
 	_ = server.RandomInitialUser(t)
+	_ = server.AddProvisionerd(t)
 }
diff --git a/coderd/projects.go b/coderd/projects.go
@@ -3,7 +3,9 @@ package coderd
 import (
 	"archive/tar"
 	"bytes"
+	"context"
 	"database/sql"
+	"encoding/json"
 	"errors"
 	"fmt"
 	"net/http"
@@ -34,6 +36,14 @@ type ProjectHistory struct {
 	StorageMethod database.ProjectStorageMethod `json:"storage_method"`
 }
 
+type ProjectHistoryLog struct {
+	ID        uuid.UUID
+	CreatedAt time.Time          `json:"created_at"`
+	Source    database.LogSource `json:"log_source"`
+	Level     database.LogLevel  `json:"log_level"`
+	Output    string             `json:"output"`
+}
+
 // CreateProjectRequest enables callers to create a new Project.
 type CreateProjectRequest struct {
 	Name        string                   `json:"name" validate:"username,required"`
@@ -48,6 +58,7 @@ type CreateProjectVersionRequest struct {
 
 type projects struct {
 	Database database.Store
+	Pubsub   database.Pubsub
 }
 
 // Lists all projects the authenticated user has access to.
@@ -222,6 +233,115 @@ func (p *projects) createProjectHistory(rw http.ResponseWriter, r *http.Request)
 	render.JSON(rw, r, convertProjectHistory(history))
 }
 
+func (p *projects) projectHistoryLogs(rw http.ResponseWriter, r *http.Request) {
+	projectHistory := httpmw.ProjectHistoryParam(r)
+	follow := r.URL.Query().Has("follow")
+
+	if !follow {
+		// If we're not attempting to follow logs,
+		// we can exit immediately!
+		logs, err := p.Database.GetProjectHistoryLogsByIDBefore(r.Context(), database.GetProjectHistoryLogsByIDBeforeParams{
+			ProjectHistoryID: projectHistory.ID,
+			CreatedAt:        time.Now(),
+		})
+		if errors.Is(err, sql.ErrNoRows) {
+			err = nil
+		}
+		if err != nil {
+			httpapi.Write(rw, http.StatusInternalServerError, httpapi.Response{
+				Message: fmt.Sprintf("get project history logs: %s", err),
+			})
+			return
+		}
+		render.Status(r, http.StatusOK)
+		render.JSON(rw, r, logs)
+		return
+	}
+
+	// We only want to fetch messages before subscribe, so that
+	// there aren't any duplicates.
+	timeBeforeSubscribe := database.Now()
+	// Start subscribing immediately, otherwise we could miss messages
+	// that occur during the database read.
+	newLogNotify := make(chan ProjectHistoryLog, 128)
+	cancelNewLogNotify, err := p.Pubsub.Subscribe(projectHistoryLogsChannel(projectHistory.ID), func(ctx context.Context, message []byte) {
+		var logs []database.ProjectHistoryLog
+		err := json.Unmarshal(message, &logs)
+		if err != nil {
+			httpapi.Write(rw, http.StatusInternalServerError, httpapi.Response{
+				Message: fmt.Sprintf("parse logs from publish: %s", err),
+			})
+			return
+		}
+		for _, log := range logs {
+			// If many logs are sent during our database query, this channel
+			// could overflow. The Go scheduler would decide the order to send
+			// logs in at that point, which is an unfortunate (but not fatal)
+			// flaw of this approach.
+			//
+			// This is an extremely unlikely outcome given reasonable database
+			// query times.
+			newLogNotify <- convertProjectHistoryLog(log)
+		}
+	})
+	if err != nil {
+		httpapi.Write(rw, http.StatusInternalServerError, httpapi.Response{
+			Message: fmt.Sprintf("listen for new logs: %s", err),
+		})
+		return
+	}
+	defer cancelNewLogNotify()
+
+	// In-between here logs could be missed!
+	projectHistoryLogs, err := p.Database.GetProjectHistoryLogsByIDBefore(r.Context(), database.GetProjectHistoryLogsByIDBeforeParams{
+		ProjectHistoryID: projectHistory.ID,
+		CreatedAt:        timeBeforeSubscribe,
+	})
+	if errors.Is(err, sql.ErrNoRows) {
+		err = nil
+	}
+	if err != nil {
+		httpapi.Write(rw, http.StatusInternalServerError, httpapi.Response{
+			Message: fmt.Sprintf("get project history logs: %s", err),
+		})
+		return
+	}
+
+	// "follow" uses the ndjson format to stream data.
+	// See: https://canjs.com/doc/can-ndjson-stream.html
+	rw.Header().Set("Content-Type", "application/stream+json")
+	rw.WriteHeader(http.StatusOK)
+	rw.(http.Flusher).Flush()
+
+	// The Go stdlib JSON encoder appends a newline character after message write.
+	encoder := json.NewEncoder(rw)
+	for _, projectHistoryLog := range projectHistoryLogs {
+		// JSON separated by a newline
+		err = encoder.Encode(convertProjectHistoryLog(projectHistoryLog))
+		if err != nil {
+			httpapi.Write(rw, http.StatusInternalServerError, httpapi.Response{
+				Message: fmt.Sprintf("marshal: %s", err),
+			})
+			return
+		}
+	}
+
+	for {
+		select {
+		case <-r.Context().Done():
+			return
+		case log := <-newLogNotify:
+			err = encoder.Encode(log)
+			if err != nil {
+				httpapi.Write(rw, http.StatusInternalServerError, httpapi.Response{
+					Message: fmt.Sprintf("marshal follow: %s", err),
+				})
+				return
+			}
+		}
+	}
+}
+
 func convertProjectHistory(history database.ProjectHistory) ProjectHistory {
 	return ProjectHistory{
 		ID:        history.ID,
@@ -231,3 +351,17 @@ func convertProjectHistory(history database.ProjectHistory) ProjectHistory {
 		Name:      history.Name,
 	}
 }
+
+func convertProjectHistoryLog(log database.ProjectHistoryLog) ProjectHistoryLog {
+	return ProjectHistoryLog{
+		ID:        log.ID,
+		CreatedAt: log.CreatedAt,
+		Source:    log.Source,
+		Level:     log.Level,
+		Output:    log.Output,
+	}
+}
+
+func projectHistoryLogsChannel(projectHistoryID uuid.UUID) string {
+	return fmt.Sprintf("project-history-logs:%s", projectHistoryID)
+}