feat: Add external provisioner daemons (#4935)

* Start to port over provisioner daemons PR

* Move to Enterprise

* Begin adding tests for external registration

* Move provisioner daemons query to enterprise

* Move around provisioner daemons schema

* Add tags to provisioner daemons

* make gen

* Add user local provisioner daemons

* Add provisioner daemons

* Add feature for external daemons

* Add command to start a provisioner daemon

* Add provisioner tags to template push and create

* Rename migration files

* Fix tests

* Fix entitlements test

* PR comments

* Update migration

* Fix FE types

.vscode/settings.json

@@ -17,6 +17,7 @@
     "codersdk",
     "cronstrue",
     "databasefake",
+    "dbtype",
     "DERP",
     "derphttp",
     "derpmap",

cli/deployment/config.go

@@ -143,7 +143,7 @@ func newConfig() *codersdk.DeploymentConfig {
 			Name:    "Cache Directory",
 			Usage:   "The directory to cache temporary files. If unspecified and $CACHE_DIRECTORY is set, it will be used for compatibility with systemd.",
 			Flag:    "cache-dir",
-			Default: defaultCacheDir(),
+			Default: DefaultCacheDir(),
 		},
 		InMemoryDatabase: &codersdk.DeploymentConfigField[bool]{
 			Name:   "In Memory Database",
@@ -672,7 +672,7 @@ func formatEnv(key string) string {
 	return "CODER_" + strings.ToUpper(strings.NewReplacer("-", "_", ".", "_").Replace(key))
 }
 
-func defaultCacheDir() string {
+func DefaultCacheDir() string {
 	defaultCacheDir, err := os.UserCacheDir()
 	if err != nil {
 		defaultCacheDir = os.TempDir()

cli/gitaskpass.go

@@ -26,7 +26,7 @@ func gitAskpass() *cobra.Command {
 		RunE: func(cmd *cobra.Command, args []string) error {
 			ctx := cmd.Context()
 
-			ctx, stop := signal.NotifyContext(ctx, interruptSignals...)
+			ctx, stop := signal.NotifyContext(ctx, InterruptSignals...)
 			defer stop()
 
 			user, host, err := gitauth.ParseAskpass(args[0])

cli/gitssh.go

@@ -29,7 +29,7 @@ func gitssh() *cobra.Command {
 
 			// Catch interrupt signals to ensure the temporary private
 			// key file is cleaned up on most cases.
-			ctx, stop := signal.NotifyContext(ctx, interruptSignals...)
+			ctx, stop := signal.NotifyContext(ctx, InterruptSignals...)
 			defer stop()
 
 			// Early check so errors are reported immediately.

cli/server.go

@@ -108,7 +108,7 @@ func Server(vip *viper.Viper, newAPI func(context.Context, *coderd.Options) (*co
 			//
 			// To get out of a graceful shutdown, the user can send
 			// SIGQUIT with ctrl+\ or SIGKILL with `kill -9`.
-			notifyCtx, notifyStop := signal.NotifyContext(ctx, interruptSignals...)
+			notifyCtx, notifyStop := signal.NotifyContext(ctx, InterruptSignals...)
 			defer notifyStop()
 
 			// Clean up idle connections at the end, e.g.
@@ -946,7 +946,7 @@ func newProvisionerDaemon(
 	return provisionerd.New(func(ctx context.Context) (proto.DRPCProvisionerDaemonClient, error) {
 		// This debounces calls to listen every second. Read the comment
 		// in provisionerdserver.go to learn more!
-		return coderAPI.ListenProvisionerDaemon(ctx, time.Second)
+		return coderAPI.CreateInMemoryProvisionerDaemon(ctx, time.Second)
 	}, &provisionerd.Options{
 		Logger:              logger,
 		PollInterval:        500 * time.Millisecond,

cli/signal_unix.go

@@ -7,7 +7,7 @@ import (
 	"syscall"
 )
 
-var interruptSignals = []os.Signal{
+var InterruptSignals = []os.Signal{
 	os.Interrupt,
 	syscall.SIGTERM,
 	syscall.SIGHUP,

cli/signal_windows.go

@@ -6,4 +6,4 @@ import (
 	"os"
 )
 
-var interruptSignals = []os.Signal{os.Interrupt}
+var InterruptSignals = []os.Signal{os.Interrupt}

cli/templatecreate.go

@@ -24,10 +24,11 @@ import (
 
 func templateCreate() *cobra.Command {
 	var (
-		directory     string
-		provisioner   string
-		parameterFile string
-		defaultTTL    time.Duration
+		directory       string
+		provisioner     string
+		provisionerTags []string
+		parameterFile   string
+		defaultTTL      time.Duration
 	)
 	cmd := &cobra.Command{
 		Use:   "create [name]",
@@ -87,12 +88,18 @@ func templateCreate() *cobra.Command {
 			}
 			spin.Stop()
 
+			tags, err := ParseProvisionerTags(provisionerTags)
+			if err != nil {
+				return err
+			}
+
 			job, _, err := createValidTemplateVersion(cmd, createValidTemplateVersionArgs{
-				Client:        client,
-				Organization:  organization,
-				Provisioner:   database.ProvisionerType(provisioner),
-				FileID:        resp.ID,
-				ParameterFile: parameterFile,
+				Client:          client,
+				Organization:    organization,
+				Provisioner:     database.ProvisionerType(provisioner),
+				FileID:          resp.ID,
+				ParameterFile:   parameterFile,
+				ProvisionerTags: tags,
 			})
 			if err != nil {
 				return err
@@ -131,6 +138,7 @@ func templateCreate() *cobra.Command {
 	cmd.Flags().StringVarP(&directory, "directory", "d", currentDirectory, "Specify the directory to create from")
 	cmd.Flags().StringVarP(&provisioner, "test.provisioner", "", "terraform", "Customize the provisioner backend")
 	cmd.Flags().StringVarP(&parameterFile, "parameter-file", "", "", "Specify a file path with parameter values.")
+	cmd.Flags().StringArrayVarP(&provisionerTags, "provisioner-tag", "", []string{}, "Specify a set of tags to target provisioner daemons.")
 	cmd.Flags().DurationVarP(&defaultTTL, "default-ttl", "", 24*time.Hour, "Specify a default TTL for workspaces created from this template.")
 	// This is for testing!
 	err := cmd.Flags().MarkHidden("test.provisioner")
@@ -154,6 +162,7 @@ type createValidTemplateVersionArgs struct {
 	// before prompting the user. Set to false to always prompt for param
 	// values.
 	ReuseParameters bool
+	ProvisionerTags map[string]string
 }
 
 func createValidTemplateVersion(cmd *cobra.Command, args createValidTemplateVersionArgs, parameters ...codersdk.CreateParameterRequest) (*codersdk.TemplateVersion, []codersdk.CreateParameterRequest, error) {
@@ -165,6 +174,7 @@ func createValidTemplateVersion(cmd *cobra.Command, args createValidTemplateVers
 		FileID:          args.FileID,
 		Provisioner:     codersdk.ProvisionerType(args.Provisioner),
 		ParameterValues: parameters,
+		ProvisionerTags: args.ProvisionerTags,
 	}
 	if args.Template != nil {
 		req.TemplateID = args.Template.ID
@@ -334,3 +344,15 @@ func prettyDirectoryPath(dir string) string {
 	}
 	return pretty
 }
+
+func ParseProvisionerTags(rawTags []string) (map[string]string, error) {
+	tags := map[string]string{}
+	for _, rawTag := range rawTags {
+		parts := strings.SplitN(rawTag, "=", 2)
+		if len(parts) < 2 {
+			return nil, xerrors.Errorf("invalid tag format for %q. must be key=value", rawTag)
+		}
+		tags[parts[0]] = parts[1]
+	}
+	return tags, nil
+}

cli/templatepush.go

@@ -18,11 +18,12 @@ import (
 
 func templatePush() *cobra.Command {
 	var (
-		directory     string
-		versionName   string
-		provisioner   string
-		parameterFile string
-		alwaysPrompt  bool
+		directory       string
+		versionName     string
+		provisioner     string
+		parameterFile   string
+		alwaysPrompt    bool
+		provisionerTags []string
 	)
 
 	cmd := &cobra.Command{
@@ -75,6 +76,11 @@ func templatePush() *cobra.Command {
 			}
 			spin.Stop()
 
+			tags, err := ParseProvisionerTags(provisionerTags)
+			if err != nil {
+				return err
+			}
+
 			job, _, err := createValidTemplateVersion(cmd, createValidTemplateVersionArgs{
 				Name:            versionName,
 				Client:          client,
@@ -84,6 +90,7 @@ func templatePush() *cobra.Command {
 				ParameterFile:   parameterFile,
 				Template:        &template,
 				ReuseParameters: !alwaysPrompt,
+				ProvisionerTags: tags,
 			})
 			if err != nil {
 				return err

coderd/autobuild/executor/lifecycle_executor.go

@@ -278,6 +278,7 @@ func build(ctx context.Context, store database.Store, workspace database.Workspa
 		Type:           database.ProvisionerJobTypeWorkspaceBuild,
 		StorageMethod:  priorJob.StorageMethod,
 		FileID:         priorJob.FileID,
+		Tags:           priorJob.Tags,
 		Input:          input,
 	})
 	if err != nil {

coderd/coderd.go

@@ -1,8 +1,10 @@
 package coderd
 
 import (
+	"context"
 	"crypto/tls"
 	"crypto/x509"
+	"encoding/json"
 	"fmt"
 	"io"
 	"net/http"
@@ -18,10 +20,13 @@ import (
 	"github.com/go-chi/chi/v5/middleware"
 	"github.com/google/uuid"
 	"github.com/klauspost/compress/zstd"
+	"github.com/moby/moby/pkg/namesgenerator"
 	"github.com/prometheus/client_golang/prometheus"
 	"go.opentelemetry.io/otel/trace"
 	"golang.org/x/xerrors"
 	"google.golang.org/api/idtoken"
+	"storj.io/drpc/drpcmux"
+	"storj.io/drpc/drpcserver"
 	"tailscale.com/derp"
 	"tailscale.com/derp/derphttp"
 	"tailscale.com/tailcfg"
@@ -32,17 +37,20 @@ import (
 	"github.com/coder/coder/coderd/audit"
 	"github.com/coder/coder/coderd/awsidentity"
 	"github.com/coder/coder/coderd/database"
+	"github.com/coder/coder/coderd/database/dbtype"
 	"github.com/coder/coder/coderd/gitauth"
 	"github.com/coder/coder/coderd/gitsshkey"
 	"github.com/coder/coder/coderd/httpapi"
 	"github.com/coder/coder/coderd/httpmw"
 	"github.com/coder/coder/coderd/metricscache"
+	"github.com/coder/coder/coderd/provisionerdserver"
 	"github.com/coder/coder/coderd/rbac"
 	"github.com/coder/coder/coderd/telemetry"
 	"github.com/coder/coder/coderd/tracing"
 	"github.com/coder/coder/coderd/wsconncache"
 	"github.com/coder/coder/codersdk"
 	"github.com/coder/coder/provisionerd/proto"
+	"github.com/coder/coder/provisionersdk"
 	"github.com/coder/coder/site"
 	"github.com/coder/coder/tailnet"
 )
@@ -323,13 +331,6 @@ func New(options *Options) *API {
 			r.Get("/{fileID}", api.fileByID)
 			r.Post("/", api.postFile)
 		})
-
-		r.Route("/provisionerdaemons", func(r chi.Router) {
-			r.Use(
-				apiKeyMiddleware,
-			)
-			r.Get("/", api.provisionerDaemons)
-		})
 		r.Route("/organizations", func(r chi.Router) {
 			r.Use(
 				apiKeyMiddleware,
@@ -595,18 +596,20 @@ type API struct {
 	// RootHandler serves "/"
 	RootHandler chi.Router
 
-	metricsCache        *metricscache.Cache
-	siteHandler         http.Handler
-	websocketWaitMutex  sync.Mutex
-	websocketWaitGroup  sync.WaitGroup
+	metricsCache *metricscache.Cache
+	siteHandler  http.Handler
+
+	WebsocketWaitMutex sync.Mutex
+	WebsocketWaitGroup sync.WaitGroup
+
 	workspaceAgentCache *wsconncache.Cache
 }
 
 // Close waits for all WebSocket connections to drain before returning.
 func (api *API) Close() error {
-	api.websocketWaitMutex.Lock()
-	api.websocketWaitGroup.Wait()
-	api.websocketWaitMutex.Unlock()
+	api.WebsocketWaitMutex.Lock()
+	api.WebsocketWaitGroup.Wait()
+	api.WebsocketWaitMutex.Unlock()
 
 	api.metricsCache.Close()
 	coordinator := api.TailnetCoordinator.Load()
@@ -635,3 +638,70 @@ func compressHandler(h http.Handler) http.Handler {
 
 	return cmp.Handler(h)
 }
+
+// CreateInMemoryProvisionerDaemon is an in-memory connection to a provisionerd.  Useful when starting coderd and provisionerd
+// in the same process.
+func (api *API) CreateInMemoryProvisionerDaemon(ctx context.Context, debounce time.Duration) (client proto.DRPCProvisionerDaemonClient, err error) {
+	clientSession, serverSession := provisionersdk.TransportPipe()
+	defer func() {
+		if err != nil {
+			_ = clientSession.Close()
+			_ = serverSession.Close()
+		}
+	}()
+
+	name := namesgenerator.GetRandomName(1)
+	daemon, err := api.Database.InsertProvisionerDaemon(ctx, database.InsertProvisionerDaemonParams{
+		ID:           uuid.New(),
+		CreatedAt:    database.Now(),
+		Name:         name,
+		Provisioners: []database.ProvisionerType{database.ProvisionerTypeEcho, database.ProvisionerTypeTerraform},
+		Tags: dbtype.StringMap{
+			provisionerdserver.TagScope: provisionerdserver.ScopeOrganization,
+		},
+	})
+	if err != nil {
+		return nil, xerrors.Errorf("insert provisioner daemon %q: %w", name, err)
+	}
+
+	tags, err := json.Marshal(daemon.Tags)
+	if err != nil {
+		return nil, xerrors.Errorf("marshal tags: %w", err)
+	}
+
+	mux := drpcmux.New()
+	err = proto.DRPCRegisterProvisionerDaemon(mux, &provisionerdserver.Server{
+		AccessURL:          api.AccessURL,
+		ID:                 daemon.ID,
+		Database:           api.Database,
+		Pubsub:             api.Pubsub,
+		Provisioners:       daemon.Provisioners,
+		Telemetry:          api.Telemetry,
+		Tags:               tags,
+		QuotaCommitter:     &api.QuotaCommitter,
+		AcquireJobDebounce: debounce,
+		Logger:             api.Logger.Named(fmt.Sprintf("provisionerd-%s", daemon.Name)),
+	})
+	if err != nil {
+		return nil, err
+	}
+	server := drpcserver.NewWithOptions(mux, drpcserver.Options{
+		Log: func(err error) {
+			if xerrors.Is(err, io.EOF) {
+				return
+			}
+			api.Logger.Debug(ctx, "drpc server error", slog.Error(err))
+		},
+	})
+	go func() {
+		err := server.Serve(ctx, serverSession)
+		if err != nil && !xerrors.Is(err, io.EOF) {
+			api.Logger.Debug(ctx, "provisioner daemon disconnected", slog.Error(err))
+		}
+		// close the sessions so we don't leak goroutines serving them.
+		_ = clientSession.Close()
+		_ = serverSession.Close()
+	}()
+
+	return proto.NewDRPCProvisionerDaemonClient(provisionersdk.Conn(clientSession)), nil
+}

coderd/coderdtest/authorize.go

@@ -19,7 +19,6 @@ import (
 	"github.com/coder/coder/codersdk"
 	"github.com/coder/coder/provisioner/echo"
 	"github.com/coder/coder/provisionersdk/proto"
-	"github.com/coder/coder/testutil"
 )
 
 func AGPLRoutes(a *AuthTester) (map[string]string, map[string]RouteCheck) {
@@ -204,11 +203,6 @@ func AGPLRoutes(a *AuthTester) (map[string]string, map[string]RouteCheck) {
 			AssertAction: rbac.ActionRead,
 			AssertObject: rbac.ResourceTemplate.InOrg(a.Version.OrganizationID),
 		},
-		"GET:/api/v2/provisionerdaemons": {
-			StatusCode:   http.StatusOK,
-			AssertObject: rbac.ResourceProvisionerDaemon,
-		},
-
 		"POST:/api/v2/parameters/{scope}/{id}": {
 			AssertAction: rbac.ActionUpdate,
 			AssertObject: rbac.ResourceTemplate,
@@ -303,16 +297,6 @@ func NewAuthTester(ctx context.Context, t *testing.T, client *codersdk.Client, a
 	if !ok {
 		t.Fail()
 	}
-	// The provisioner will call to coderd and register itself. This is async,
-	// so we wait for it to occur.
-	require.Eventually(t, func() bool {
-		provisionerds, err := client.ProvisionerDaemons(ctx)
-		return assert.NoError(t, err) && len(provisionerds) > 0
-	}, testutil.WaitLong, testutil.IntervalSlow)
-
-	provisionerds, err := client.ProvisionerDaemons(ctx)
-	require.NoError(t, err, "fetch provisioners")
-	require.Len(t, provisionerds, 1)
 
 	organization, err := client.Organization(ctx, admin.OrganizationID)
 	require.NoError(t, err, "fetch org")