refactor: server start

pkg/api/server.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/daytonaio/daytona/pkg/api/docs"
 	"github.com/daytonaio/daytona/pkg/api/middlewares"
+	"github.com/daytonaio/daytona/pkg/frpc"
 	"github.com/daytonaio/daytona/pkg/telemetry"
 	"github.com/gin-contrib/cors"
 
@@ -52,6 +53,7 @@ import (
 	log "github.com/sirupsen/logrus"
 
 	"github.com/daytonaio/daytona/internal/constants"
+	daytonaServer "github.com/daytonaio/daytona/pkg/server"
 	swaggerfiles "github.com/swaggo/files"
 	ginSwagger "github.com/swaggo/gin-swagger"
 )
@@ -60,13 +62,17 @@ type ApiServerConfig struct {
 	ApiPort          int
 	Version          string
 	TelemetryService telemetry.TelemetryService
+	Frps             *daytonaServer.FRPSConfig
+	ServerId         string
 }
 
 func NewApiServer(config ApiServerConfig) *ApiServer {
 	return &ApiServer{
 		apiPort:          config.ApiPort,
 		telemetryService: config.TelemetryService,
 		version:          config.Version,
+		frps:             config.Frps,
+		serverId:         config.ServerId,
 	}
 }
 
@@ -76,6 +82,8 @@ type ApiServer struct {
 	httpServer       *http.Server
 	router           *gin.Engine
 	version          string
+	frps             *daytonaServer.FRPSConfig
+	serverId         string
 }
 
 func (a *ApiServer) Start() error {
@@ -262,8 +270,44 @@ func (a *ApiServer) Start() error {
 		return err
 	}
 
-	log.Infof("Starting api server on port %d", a.apiPort)
-	return a.httpServer.Serve(listener)
+	errChan := make(chan error)
+	go func() {
+		errChan <- a.httpServer.Serve(listener)
+	}()
+
+	if a.frps != nil {
+		frpcHealthCheck, frpcService, err := frpc.GetService(frpc.FrpcConnectParams{
+			ServerDomain: a.frps.Domain,
+			ServerPort:   int(a.frps.Port),
+			Name:         fmt.Sprintf("daytona-server-api-%s", a.serverId),
+			Port:         int(a.apiPort),
+			SubDomain:    fmt.Sprintf("api-%s", a.serverId),
+		})
+		if err != nil {
+			return err
+		}
+
+		go func() {
+			err := frpcService.Run(context.Background())
+			if err != nil {
+				errChan <- err
+			}
+		}()
+
+		for i := 0; i < 5; i++ {
+			if err = frpcHealthCheck(); err != nil {
+				log.Debugf("Failed to connect to api frpc: %s", err)
+				time.Sleep(2 * time.Second)
+			} else {
+				break
+			}
+		}
+		if err != nil {
+			return err
+		}
+	}
+
+	return <-errChan
 }
 
 func (a *ApiServer) HealthCheck() error {

pkg/cmd/server/serve.go

@@ -75,33 +75,19 @@ var ServeCmd = &cobra.Command{
 			Version:  internal.Version,
 		})
 
-		go func() {
-			interruptChannel := make(chan os.Signal, 1)
-			signal.Notify(interruptChannel, os.Interrupt)
-
-			for range interruptChannel {
-				log.Info("Shutting down")
-			}
-		}()
-
 		apiServer := api.NewApiServer(api.ApiServerConfig{
 			ApiPort:          int(c.ApiPort),
 			TelemetryService: telemetryService,
 			Version:          internal.Version,
+			ServerId:         c.Id,
+			Frps:             c.Frps,
 		})
 
 		server, err := GetInstance(c, configDir, internal.Version, telemetryService)
 		if err != nil {
 			return err
 		}
 
-		errCh := make(chan error)
-
-		err = server.Start(errCh)
-		if err != nil {
-			return err
-		}
-
 		buildRunnerConfig, err := build.GetConfig()
 		if err != nil {
 			return err
@@ -117,22 +103,58 @@ var ServeCmd = &cobra.Command{
 			return err
 		}
 
+		apiServerErrChan := make(chan error)
+
+		go func() {
+			log.Infof("Starting api server on port %d", c.ApiPort)
+			apiServerErrChan <- apiServer.Start()
+		}()
+
+		headscaleServerStartedChan := make(chan struct{})
+		headscaleServerErrChan := make(chan error)
+
 		go func() {
-			err := apiServer.Start()
+			log.Info("Starting headscale server...")
+			err := server.TailscaleServer.Start(headscaleServerErrChan)
 			if err != nil {
-				errCh <- err
+				headscaleServerErrChan <- err
+				return
 			}
+			headscaleServerStartedChan <- struct{}{}
 		}()
 
+		localContainerRegistryErrChan := make(chan error)
+
 		go func() {
-			err := <-errCh
-			if err != nil {
-				buildRunner.Stop()
+			if server.LocalContainerRegistry != nil {
+				log.Info("Starting local container registry...")
+				localContainerRegistryErrChan <- server.LocalContainerRegistry.Start()
+			} else {
+				localContainerRegistryErrChan <- registry.RemoveRegistryContainer()
 			}
 		}()
 
-		err = waitForServerToStart(apiServer)
+		select {
+		case <-headscaleServerStartedChan:
+			log.Info("Headscale server started")
+			go func() {
+				headscaleServerErrChan <- server.TailscaleServer.Connect()
+			}()
+		case err := <-headscaleServerErrChan:
+			return err
+		}
 
+		err = server.Start()
+		if err != nil {
+			return err
+		}
+
+		err = waitForApiServerToStart(apiServer)
+		if err != nil {
+			return err
+		}
+
+		err = <-localContainerRegistryErrChan
 		if err != nil {
 			return err
 		}
@@ -144,7 +166,19 @@ var ServeCmd = &cobra.Command{
 			return err
 		}
 
-		return <-errCh
+		interruptChannel := make(chan os.Signal, 1)
+		signal.Notify(interruptChannel, os.Interrupt)
+
+		select {
+		case err := <-apiServerErrChan:
+			return err
+		case err := <-headscaleServerErrChan:
+			return err
+		case <-interruptChannel:
+			log.Info("Shutting down")
+			// Exit will be handled by command PreRun
+			select {}
+		}
 	},
 }
 
@@ -205,6 +239,7 @@ func GetInstance(c *server.Config, configDir string, version string, telemetrySe
 		FrpsProtocol:  c.Frps.Protocol,
 		HeadscalePort: c.HeadscalePort,
 		ConfigDir:     filepath.Join(configDir, "headscale"),
+		Frps:          c.Frps,
 	})
 	err = headscaleServer.Init()
 	if err != nil {
@@ -255,6 +290,8 @@ func GetInstance(c *server.Config, configDir string, version string, telemetrySe
 			Port:     c.LocalBuilderRegistryPort,
 			Image:    c.LocalBuilderRegistryImage,
 			Logger:   log.StandardLogger().Writer(),
+			Frps:     c.Frps,
+			ServerId: c.Id,
 		})
 		c.BuilderRegistryServer = util.GetFrpcRegistryDomain(c.Id, c.Frps.Domain)
 	}
@@ -311,7 +348,7 @@ func GetInstance(c *server.Config, configDir string, version string, telemetrySe
 		ProfileDataStore: profileDataStore,
 	})
 
-	return server.GetInstance(&server.ServerInstanceConfig{
+	s := server.GetInstance(&server.ServerInstanceConfig{
 		Config:                   *c,
 		Version:                  version,
 		TailscaleServer:          headscaleServer,
@@ -326,7 +363,9 @@ func GetInstance(c *server.Config, configDir string, version string, telemetrySe
 		ProviderManager:          providerManager,
 		ProfileDataService:       profileDataService,
 		TelemetryService:         telemetryService,
-	}), nil
+	})
+
+	return s, s.Initialize()
 }
 
 func GetBuildRunner(c *server.Config, buildRunnerConfig *build.Config, telemetryService telemetry.TelemetryService) (*build.BuildRunner, error) {
@@ -412,7 +451,7 @@ func GetBuildRunner(c *server.Config, buildRunnerConfig *build.Config, telemetry
 	}), nil
 }
 
-func waitForServerToStart(apiServer *api.ApiServer) error {
+func waitForApiServerToStart(apiServer *api.ApiServer) error {
 	var err error
 	for i := 0; i < 30; i++ {
 		time.Sleep(1 * time.Second)

pkg/cmd/server/server.go

@@ -56,7 +56,7 @@ var ServerCmd = &cobra.Command{
 		if err != nil {
 			return err
 		}
-		err = waitForServerToStart(apiServer)
+		err = waitForApiServerToStart(apiServer)
 		if err != nil {
 			return err
 		}

pkg/server/headscale/server.go

@@ -4,11 +4,17 @@
 package headscale
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"os"
+	"time"
 
+	"github.com/daytonaio/daytona/pkg/frpc"
+	"github.com/daytonaio/daytona/pkg/server"
 	"github.com/juanfont/headscale/hscontrol"
+
+	log "github.com/sirupsen/logrus"
 )
 
 type HeadscaleServerConfig struct {
@@ -17,6 +23,7 @@ type HeadscaleServerConfig struct {
 	FrpsProtocol  string
 	HeadscalePort uint32
 	ConfigDir     string
+	Frps          *server.FRPSConfig
 }
 
 func NewHeadscaleServer(config *HeadscaleServerConfig) *HeadscaleServer {
@@ -26,6 +33,7 @@ func NewHeadscaleServer(config *HeadscaleServerConfig) *HeadscaleServer {
 		frpsProtocol:  config.FrpsProtocol,
 		headscalePort: config.HeadscalePort,
 		configDir:     config.ConfigDir,
+		frps:          config.Frps,
 	}
 }
 
@@ -35,6 +43,7 @@ type HeadscaleServer struct {
 	frpsProtocol  string
 	headscalePort uint32
 	configDir     string
+	frps          *server.FRPSConfig
 
 	stopChan chan struct{}
 }
@@ -43,7 +52,8 @@ func (s *HeadscaleServer) Init() error {
 	return os.MkdirAll(s.configDir, 0700)
 }
 
-func (s *HeadscaleServer) Start() error {
+func (s *HeadscaleServer) Start(errChan chan error) error {
+	// Check if port is already in use
 	_, err := net.Dial("tcp", fmt.Sprintf(":%d", s.headscalePort))
 	if err == nil {
 		return fmt.Errorf("cannot start Headscale server, port %d is already in use", s.headscalePort)
@@ -60,7 +70,6 @@ func (s *HeadscaleServer) Start() error {
 	}
 
 	s.stopChan = make(chan struct{})
-	errChan := make(chan error)
 
 	go func() {
 		select {
@@ -72,7 +81,39 @@ func (s *HeadscaleServer) Start() error {
 		}
 	}()
 
-	return <-errChan
+	if s.frps != nil {
+		healthCheck, frpcService, err := frpc.GetService(frpc.FrpcConnectParams{
+			ServerDomain: s.frps.Domain,
+			ServerPort:   int(s.frps.Port),
+			Name:         fmt.Sprintf("daytona-server-%s", s.serverId),
+			Port:         int(s.headscalePort),
+			SubDomain:    s.serverId,
+		})
+		if err != nil {
+			return err
+		}
+
+		go func() {
+			err := frpcService.Run(context.Background())
+			if err != nil {
+				errChan <- err
+			}
+		}()
+
+		for i := 0; i < 5; i++ {
+			if err = healthCheck(); err != nil {
+				log.Debugf("Failed to connect to headscale frpc: %s", err)
+				time.Sleep(2 * time.Second)
+			} else {
+				break
+			}
+		}
+		if err != nil {
+			return err
+		}
+	}
+
+	return err
 }
 
 func (s *HeadscaleServer) Stop() error {

pkg/server/purge.go

@@ -30,9 +30,7 @@ func (s *Server) Purge(ctx context.Context, force bool) []error {
 
 	fmt.Println("Deleting all workspaces...")
 
-	errCh := make(chan error)
-
-	err := server.Start(errCh)
+	err := server.Start()
 	if err != nil {
 		s.trackPurgeError(ctx, force, err)
 		return []error{err}