Jan 29, 2022 · Jan 29, 2022 · Jan 29, 2022 · Jan 29, 2022 · Jan 29, 2022 · Jan 29, 2022
diff --git a/coderd/coderd.go b/coderd/coderd.go
 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()
 })
 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) {
 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)
 })
 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
 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.
 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)
 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
 t.Parallel()
 server := coderdtest.New(t)
 _ = server.RandomInitialUser(t)
 _ = server.AddProvisionerd(t)
 }
diff --git a/coderd/projects.go b/coderd/projects.go
 import (
 "archive/tar"
 "bytes"
 "context"
 "database/sql"
 "encoding/json"
 "errors"
 "fmt"
 "net/http"
 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"`

 type projects struct {
 Database database.Store
 Pubsub   database.Pubsub
 }

 // Lists all projects the authenticated user has access to.
 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,
 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)
 }
Original file line number	Diff line number	Diff line change
Expand Up		@@ -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()
Expand All		@@ -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) {
Expand DownExpand Up		@@ -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)
		})
Expand All		@@ -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
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -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.
Expand DownExpand Up		@@ -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)
Expand All		@@ -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)
Expand Down
Original file line number	Diff line number	Diff line change
Expand Up		@@ -16,4 +16,5 @@ func TestNew(t *testing.T) {
		t.Parallel()
		server := coderdtest.New(t)
		_ = server.RandomInitialUser(t)
		_ = server.AddProvisionerd(t)
		}
Original file line number	Diff line number	Diff line change
Expand Up		@@ -3,7 +3,9 @@ package coderd
		import (
		"archive/tar"
		"bytes"
		"context"
		"database/sql"
		"encoding/json"
		"errors"
		"fmt"
		"net/http"
Expand DownExpand Up		@@ -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"`
Expand All		@@ -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.
Expand DownExpand Up		@@ -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,
Expand All		@@ -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)
		}