feat: Add streaming endpoint for workspace history (#157)

* feat: Add parameter querying to the API

* feat: Add streaming endpoint for workspace history

Enables a buildlog-like flow for reading job output.

* Fix empty parameter source and destination

* Add comment for usage of workspace history logs endpoint

coderd/coderd.go

@@ -72,6 +72,7 @@ func New(options *Options) http.Handler {
 						r.Route("/{projecthistory}", func(r chi.Router) {
 							r.Use(httpmw.ExtractProjectHistoryParam(api.Database))
 							r.Get("/", api.projectHistoryByOrganizationAndName)
+							r.Get("/parameters", api.projectHistoryParametersByOrganizationAndName)
 						})
 					})
 				})
@@ -96,6 +97,7 @@ func New(options *Options) http.Handler {
 						r.Route("/{workspacehistory}", func(r chi.Router) {
 							r.Use(httpmw.ExtractWorkspaceHistoryParam(options.Database))
 							r.Get("/", api.workspaceHistoryByName)
+							r.Get("/logs", api.workspaceHistoryLogsByName)
 						})
 					})
 				})

coderd/parameters.go

@@ -16,11 +16,11 @@ import (
 
 // CreateParameterValueRequest is used to create a new parameter value for a scope.
 type CreateParameterValueRequest struct {
-	Name              string                              `json:"name"`
-	SourceValue       string                              `json:"source_value"`
-	SourceScheme      database.ParameterSourceScheme      `json:"source_scheme"`
-	DestinationScheme database.ParameterDestinationScheme `json:"destination_scheme"`
-	DestinationValue  string                              `json:"destination_value"`
+	Name              string                              `json:"name" validate:"required"`
+	SourceValue       string                              `json:"source_value" validate:"required"`
+	SourceScheme      database.ParameterSourceScheme      `json:"source_scheme" validate:"oneof=data,required"`
+	DestinationScheme database.ParameterDestinationScheme `json:"destination_scheme" validate:"oneof=environment_variable provisioner_variable,required"`
+	DestinationValue  string                              `json:"destination_value" validate:"required"`
 }
 
 // ParameterValue represents a set value for the scope.

coderd/projecthistory.go

@@ -31,6 +31,27 @@ type ProjectHistory struct {
 	Import        ProvisionerJob                `json:"import"`
 }
 
+// ProjectParameter represents a parameter parsed from project history source on creation.
+type ProjectParameter struct {
+	ID                       uuid.UUID                           `json:"id"`
+	CreatedAt                time.Time                           `json:"created_at"`
+	ProjectHistoryID         uuid.UUID                           `json:"project_history_id"`
+	Name                     string                              `json:"name"`
+	Description              string                              `json:"description,omitempty"`
+	DefaultSourceScheme      database.ParameterSourceScheme      `json:"default_source_scheme,omitempty"`
+	DefaultSourceValue       string                              `json:"default_source_value,omitempty"`
+	AllowOverrideSource      bool                                `json:"allow_override_source"`
+	DefaultDestinationScheme database.ParameterDestinationScheme `json:"default_destination_scheme,omitempty"`
+	DefaultDestinationValue  string                              `json:"default_destination_value,omitempty"`
+	AllowOverrideDestination bool                                `json:"allow_override_destination"`
+	DefaultRefresh           string                              `json:"default_refresh"`
+	RedisplayValue           bool                                `json:"redisplay_value"`
+	ValidationError          string                              `json:"validation_error,omitempty"`
+	ValidationCondition      string                              `json:"validation_condition,omitempty"`
+	ValidationTypeSystem     database.ParameterTypeSystem        `json:"validation_type_system,omitempty"`
+	ValidationValueType      string                              `json:"validation_value_type,omitempty"`
+}
+
 // CreateProjectHistoryRequest enables callers to create a new Project Version.
 type CreateProjectHistoryRequest struct {
 	StorageMethod database.ProjectStorageMethod `json:"storage_method" validate:"oneof=inline-archive,required"`
@@ -160,14 +181,81 @@ func (api *api) postProjectHistoryByOrganization(rw http.ResponseWriter, r *http
 	render.JSON(rw, r, convertProjectHistory(projectHistory, provisionerJob))
 }
 
+func (api *api) projectHistoryParametersByOrganizationAndName(rw http.ResponseWriter, r *http.Request) {
+	projectHistory := httpmw.ProjectHistoryParam(r)
+	job, err := api.Database.GetProvisionerJobByID(r.Context(), projectHistory.ImportJobID)
+	if err != nil {
+		httpapi.Write(rw, http.StatusInternalServerError, httpapi.Response{
+			Message: fmt.Sprintf("get provisioner job: %s", err),
+		})
+		return
+	}
+	apiJob := convertProvisionerJob(job)
+	if !apiJob.Status.Completed() {
+		httpapi.Write(rw, http.StatusPreconditionRequired, httpapi.Response{
+			Message: fmt.Sprintf("import job hasn't completed: %s", apiJob.Status),
+		})
+		return
+	}
+	if apiJob.Status != ProvisionerJobStatusSucceeded {
+		httpapi.Write(rw, http.StatusPreconditionFailed, httpapi.Response{
+			Message: "import job wasn't successful. no parameters were parsed",
+		})
+		return
+	}
+
+	parameters, err := api.Database.GetProjectParametersByHistoryID(r.Context(), projectHistory.ID)
+	if errors.Is(err, sql.ErrNoRows) {
+		err = nil
+		parameters = []database.ProjectParameter{}
+	}
+	if err != nil {
+		httpapi.Write(rw, http.StatusInternalServerError, httpapi.Response{
+			Message: fmt.Sprintf("get project parameters: %s", err),
+		})
+		return
+	}
+
+	apiParameters := make([]ProjectParameter, 0, len(parameters))
+	for _, parameter := range parameters {
+		apiParameters = append(apiParameters, convertProjectParameter(parameter))
+	}
+
+	render.Status(r, http.StatusOK)
+	render.JSON(rw, r, apiParameters)
+}
+
 func convertProjectHistory(history database.ProjectHistory, job database.ProvisionerJob) ProjectHistory {
 	return ProjectHistory{
-		ID:        history.ID,
-		ProjectID: history.ProjectID,
-		CreatedAt: history.CreatedAt,
-		UpdatedAt: history.UpdatedAt,
-		Name:      history.Name,
-		Import:    convertProvisionerJob(job),
+		ID:            history.ID,
+		ProjectID:     history.ProjectID,
+		CreatedAt:     history.CreatedAt,
+		UpdatedAt:     history.UpdatedAt,
+		Name:          history.Name,
+		StorageMethod: history.StorageMethod,
+		Import:        convertProvisionerJob(job),
+	}
+}
+
+func convertProjectParameter(parameter database.ProjectParameter) ProjectParameter {
+	return ProjectParameter{
+		ID:                       parameter.ID,
+		CreatedAt:                parameter.CreatedAt,
+		ProjectHistoryID:         parameter.ProjectHistoryID,
+		Name:                     parameter.Name,
+		Description:              parameter.Description,
+		DefaultSourceScheme:      parameter.DefaultSourceScheme,
+		DefaultSourceValue:       parameter.DefaultSourceValue.String,
+		AllowOverrideSource:      parameter.AllowOverrideSource,
+		DefaultDestinationScheme: parameter.DefaultDestinationScheme,
+		DefaultDestinationValue:  parameter.DefaultDestinationValue.String,
+		AllowOverrideDestination: parameter.AllowOverrideDestination,
+		DefaultRefresh:           parameter.DefaultRefresh,
+		RedisplayValue:           parameter.RedisplayValue,
+		ValidationError:          parameter.ValidationError,
+		ValidationCondition:      parameter.ValidationCondition,
+		ValidationTypeSystem:     parameter.ValidationTypeSystem,
+		ValidationValueType:      parameter.ValidationValueType,
 	}
 }
 

coderd/projects_test.go

@@ -1,8 +1,11 @@
 package coderd_test
 
 import (
+	"archive/tar"
+	"bytes"
 	"context"
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/require"
 
@@ -124,4 +127,42 @@ func TestProjects(t *testing.T) {
 		})
 		require.NoError(t, err)
 	})
+
+	t.Run("Import", func(t *testing.T) {
+		t.Parallel()
+		server := coderdtest.New(t)
+		user := server.RandomInitialUser(t)
+		_ = server.AddProvisionerd(t)
+		project, err := server.Client.CreateProject(context.Background(), user.Organization, coderd.CreateProjectRequest{
+			Name:        "someproject",
+			Provisioner: database.ProvisionerTypeTerraform,
+		})
+		require.NoError(t, err)
+		var buffer bytes.Buffer
+		writer := tar.NewWriter(&buffer)
+		content := `variable "example" {
+	default = "hi"
+}`
+		err = writer.WriteHeader(&tar.Header{
+			Name: "main.tf",
+			Size: int64(len(content)),
+		})
+		require.NoError(t, err)
+		_, err = writer.Write([]byte(content))
+		require.NoError(t, err)
+		history, err := server.Client.CreateProjectHistory(context.Background(), user.Organization, project.Name, coderd.CreateProjectHistoryRequest{
+			StorageMethod: database.ProjectStorageMethodInlineArchive,
+			StorageSource: buffer.Bytes(),
+		})
+		require.NoError(t, err)
+		require.Eventually(t, func() bool {
+			projectHistory, err := server.Client.ProjectHistory(context.Background(), user.Organization, project.Name, history.Name)
+			require.NoError(t, err)
+			return projectHistory.Import.Status.Completed()
+		}, 15*time.Second, 10*time.Millisecond)
+		params, err := server.Client.ProjectHistoryParameters(context.Background(), user.Organization, project.Name, history.Name)
+		require.NoError(t, err)
+		require.Len(t, params, 1)
+		require.Equal(t, "example", params[0].Name)
+	})
 }

coderd/provisionerdaemons.go

@@ -451,6 +451,9 @@ func (server *provisionerdServer) CompleteJob(ctx context.Context, completed *pr
 				ValidationValueType:  protoParameter.ValidationValueType,
 				ValidationTypeSystem: validationTypeSystem,
 
+				DefaultSourceScheme:      database.ParameterSourceSchemeNone,
+				DefaultDestinationScheme: database.ParameterDestinationSchemeNone,
+
 				AllowOverrideDestination: protoParameter.AllowOverrideDestination,
 				AllowOverrideSource:      protoParameter.AllowOverrideSource,
 			}
@@ -574,6 +577,8 @@ func (server *provisionerdServer) CompleteJob(ctx context.Context, completed *pr
 
 func convertValidationTypeSystem(typeSystem sdkproto.ParameterSchema_TypeSystem) (database.ParameterTypeSystem, error) {
 	switch typeSystem {
+	case sdkproto.ParameterSchema_None:
+		return database.ParameterTypeSystemNone, nil
 	case sdkproto.ParameterSchema_HCL:
 		return database.ParameterTypeSystemHCL, nil
 	default:

coderd/workspacehistory.go

@@ -10,6 +10,7 @@ import (
 
 	"github.com/go-chi/render"
 	"github.com/google/uuid"
+	"github.com/moby/moby/pkg/namesgenerator"
 	"golang.org/x/xerrors"
 
 	"github.com/coder/coder/database"
@@ -155,6 +156,7 @@ func (api *api) postWorkspaceHistoryByUser(rw http.ResponseWriter, r *http.Reque
 			WorkspaceID:      workspace.ID,
 			ProjectHistoryID: projectHistory.ID,
 			BeforeID:         priorHistoryID,
+			Name:             namesgenerator.GetRandomName(1),
 			Initiator:        user.ID,
 			Transition:       createBuild.Transition,
 			ProvisionJobID:   provisionerJob.ID,
@@ -253,7 +255,3 @@ func convertWorkspaceHistory(workspaceHistory database.WorkspaceHistory, provisi
 		Provision:        convertProvisionerJob(provisionerJob),
 	})
 }
-
-func workspaceHistoryLogsChannel(workspaceHistoryID uuid.UUID) string {
-	return fmt.Sprintf("workspace-history-logs:%s", workspaceHistoryID)
-}

coderd/workspacehistorylogs.go

@@ -0,0 +1,203 @@
+package coderd
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"net/http"
+	"strconv"
+	"time"
+
+	"github.com/go-chi/render"
+	"github.com/google/uuid"
+
+	"cdr.dev/slog"
+
+	"github.com/coder/coder/database"
+	"github.com/coder/coder/httpapi"
+	"github.com/coder/coder/httpmw"
+)
+
+// WorkspaceHistoryLog represents a single log from workspace history.
+type WorkspaceHistoryLog 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"`
+}
+
+// Returns workspace history logs based on query parameters.
+// The intended usage for a client to stream all logs (with JS API):
+// const timestamp = new Date().getTime();
+// 1. GET /logs?before=<timestamp>
+// 2. GET /logs?after=<timestamp>&follow
+// The combination of these responses should provide all current logs
+// to the consumer, and future logs are streamed in the follow request.
+func (api *api) workspaceHistoryLogsByName(rw http.ResponseWriter, r *http.Request) {
+	follow := r.URL.Query().Has("follow")
+	afterRaw := r.URL.Query().Get("after")
+	beforeRaw := r.URL.Query().Get("before")
+	if beforeRaw != "" && follow {
+		httpapi.Write(rw, http.StatusBadRequest, httpapi.Response{
+			Message: "before cannot be used with follow",
+		})
+		return
+	}
+
+	var after time.Time
+	// Only fetch logs created after the time provided.
+	if afterRaw != "" {
+		afterMS, err := strconv.ParseInt(afterRaw, 10, 64)
+		if err != nil {
+			httpapi.Write(rw, http.StatusBadRequest, httpapi.Response{
+				Message: fmt.Sprintf("unable to parse after %q: %s", afterRaw, err),
+			})
+			return
+		}
+		after = time.UnixMilli(afterMS)
+	} else {
+		if follow {
+			after = database.Now()
+		}
+	}
+	var before time.Time
+	// Only fetch logs created before the time provided.
+	if beforeRaw != "" {
+		beforeMS, err := strconv.ParseInt(beforeRaw, 10, 64)
+		if err != nil {
+			httpapi.Write(rw, http.StatusBadRequest, httpapi.Response{
+				Message: fmt.Sprintf("unable to parse before %q: %s", beforeRaw, err),
+			})
+			return
+		}
+		before = time.UnixMilli(beforeMS)
+	} else {
+		before = database.Now()
+	}
+
+	history := httpmw.WorkspaceHistoryParam(r)
+	if !follow {
+		logs, err := api.Database.GetWorkspaceHistoryLogsByIDBetween(r.Context(), database.GetWorkspaceHistoryLogsByIDBetweenParams{
+			WorkspaceHistoryID: history.ID,
+			CreatedAfter:       after,
+			CreatedBefore:      before,
+		})
+		if errors.Is(err, sql.ErrNoRows) {
+			err = nil
+			logs = []database.WorkspaceHistoryLog{}
+		}
+		if err != nil {
+			httpapi.Write(rw, http.StatusInternalServerError, httpapi.Response{
+				Message: fmt.Sprintf("get workspace history: %s", err),
+			})
+			return
+		}
+		render.Status(r, http.StatusOK)
+		render.JSON(rw, r, logs)
+		return
+	}
+
+	bufferedLogs := make(chan database.WorkspaceHistoryLog, 128)
+	closeSubscribe, err := api.Pubsub.Subscribe(workspaceHistoryLogsChannel(history.ID), func(ctx context.Context, message []byte) {
+		var logs []database.WorkspaceHistoryLog
+		err := json.Unmarshal(message, &logs)
+		if err != nil {
+			api.Logger.Warn(r.Context(), fmt.Sprintf("invalid workspace log on channel %q: %s", workspaceHistoryLogsChannel(history.ID), err.Error()))
+			return
+		}
+
+		for _, log := range logs {
+			select {
+			case bufferedLogs <- log:
+			default:
+				// This is a case that shouldn't happen, but totally could.
+				// There's no way to stream data from the database, so we'll
+				// need to maintain some level of internal buffer.
+				//
+				// If this overflows users could miss logs when streaming.
+				// We warn to make sure we know when it happens!
+				api.Logger.Warn(r.Context(), "workspace history log overflowing channel")
+			}
+		}
+	})
+	if err != nil {
+		httpapi.Write(rw, http.StatusInternalServerError, httpapi.Response{
+			Message: fmt.Sprintf("subscribe to workspace history logs: %s", err),
+		})
+		return
+	}
+	defer closeSubscribe()
+
+	workspaceHistoryLogs, err := api.Database.GetWorkspaceHistoryLogsByIDBetween(r.Context(), database.GetWorkspaceHistoryLogsByIDBetweenParams{
+		WorkspaceHistoryID: history.ID,
+		CreatedAfter:       after,
+		CreatedBefore:      before,
+	})
+	if errors.Is(err, sql.ErrNoRows) {
+		err = nil
+		workspaceHistoryLogs = []database.WorkspaceHistoryLog{}
+	}
+	if err != nil {
+		httpapi.Write(rw, http.StatusInternalServerError, httpapi.Response{
+			Message: fmt.Sprint("get workspace history logs: %w", 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 _, workspaceHistoryLog := range workspaceHistoryLogs {
+		err = encoder.Encode(convertWorkspaceHistoryLog(workspaceHistoryLog))
+		if err != nil {
+			return
+		}
+	}
+
+	ticker := time.NewTicker(250 * time.Millisecond)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-r.Context().Done():
+			return
+		case log := <-bufferedLogs:
+			err = encoder.Encode(convertWorkspaceHistoryLog(log))
+			if err != nil {
+				return
+			}
+			rw.(http.Flusher).Flush()
+		case <-ticker.C:
+			job, err := api.Database.GetProvisionerJobByID(r.Context(), history.ProvisionJobID)
+			if err != nil {
+				api.Logger.Warn(r.Context(), "streaming workspace logs; checking if job completed", slog.Error(err), slog.F("job_id", history.ProvisionJobID))
+				continue
+			}
+			if convertProvisionerJob(job).Status.Completed() {
+				return
+			}
+		}
+	}
+}
+
+func convertWorkspaceHistoryLog(workspaceHistoryLog database.WorkspaceHistoryLog) WorkspaceHistoryLog {
+	return WorkspaceHistoryLog{
+		ID:        workspaceHistoryLog.ID,
+		CreatedAt: workspaceHistoryLog.CreatedAt,
+		Source:    workspaceHistoryLog.Source,
+		Level:     workspaceHistoryLog.Level,
+		Output:    workspaceHistoryLog.Output,
+	}
+}
+
+func workspaceHistoryLogsChannel(workspaceHistoryID uuid.UUID) string {
+	return fmt.Sprintf("workspace-history-logs:%s", workspaceHistoryID)
+}

coderd/workspacehistorylogs_test.go

@@ -0,0 +1,102 @@
+package coderd_test
+
+import (
+	"archive/tar"
+	"bytes"
+	"context"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/coder/coder/coderd"
+	"github.com/coder/coder/coderd/coderdtest"
+	"github.com/coder/coder/codersdk"
+	"github.com/coder/coder/database"
+)
+
+func TestWorkspaceHistoryLogs(t *testing.T) {
+	t.Parallel()
+
+	setupProjectAndWorkspace := func(t *testing.T, client *codersdk.Client, user coderd.CreateInitialUserRequest) (coderd.Project, coderd.Workspace) {
+		project, err := client.CreateProject(context.Background(), user.Organization, coderd.CreateProjectRequest{
+			Name:        "banana",
+			Provisioner: database.ProvisionerTypeTerraform,
+		})
+		require.NoError(t, err)
+		workspace, err := client.CreateWorkspace(context.Background(), "", coderd.CreateWorkspaceRequest{
+			Name:      "example",
+			ProjectID: project.ID,
+		})
+		require.NoError(t, err)
+		return project, workspace
+	}
+
+	setupProjectHistory := func(t *testing.T, client *codersdk.Client, user coderd.CreateInitialUserRequest, project coderd.Project, files map[string]string) coderd.ProjectHistory {
+		var buffer bytes.Buffer
+		writer := tar.NewWriter(&buffer)
+		for path, content := range files {
+			err := writer.WriteHeader(&tar.Header{
+				Name: path,
+				Size: int64(len(content)),
+			})
+			require.NoError(t, err)
+			_, err = writer.Write([]byte(content))
+			require.NoError(t, err)
+		}
+		err := writer.Flush()
+		require.NoError(t, err)
+
+		projectHistory, err := client.CreateProjectHistory(context.Background(), user.Organization, project.Name, coderd.CreateProjectHistoryRequest{
+			StorageMethod: database.ProjectStorageMethodInlineArchive,
+			StorageSource: buffer.Bytes(),
+		})
+		require.NoError(t, err)
+		require.Eventually(t, func() bool {
+			hist, err := client.ProjectHistory(context.Background(), user.Organization, project.Name, projectHistory.Name)
+			require.NoError(t, err)
+			return hist.Import.Status.Completed()
+		}, 15*time.Second, 50*time.Millisecond)
+		return projectHistory
+	}
+
+	server := coderdtest.New(t)
+	user := server.RandomInitialUser(t)
+	_ = server.AddProvisionerd(t)
+	project, workspace := setupProjectAndWorkspace(t, server.Client, user)
+	projectHistory := setupProjectHistory(t, server.Client, user, project, map[string]string{
+		"main.tf": `resource "null_resource" "test" {}`,
+	})
+
+	workspaceHistory, err := server.Client.CreateWorkspaceHistory(context.Background(), "", workspace.Name, coderd.CreateWorkspaceHistoryRequest{
+		ProjectHistoryID: projectHistory.ID,
+		Transition:       database.WorkspaceTransitionCreate,
+	})
+	require.NoError(t, err)
+
+	now := database.Now()
+	logChan, err := server.Client.FollowWorkspaceHistoryLogsAfter(context.Background(), "", workspace.Name, workspaceHistory.Name, now)
+	require.NoError(t, err)
+
+	for {
+		log, more := <-logChan
+		if !more {
+			break
+		}
+		t.Logf("Output: %s", log.Output)
+	}
+
+	t.Run("ReturnAll", func(t *testing.T) {
+		t.Parallel()
+
+		_, err = server.Client.WorkspaceHistoryLogs(context.Background(), "", workspace.Name, workspaceHistory.Name)
+		require.NoError(t, err)
+	})
+
+	t.Run("Between", func(t *testing.T) {
+		t.Parallel()
+
+		_, err = server.Client.WorkspaceHistoryLogsBetween(context.Background(), "", workspace.Name, workspaceHistory.Name, time.Time{}, database.Now())
+		require.NoError(t, err)
+	})
+}