feat: implement dependency locking

Author: FabianKramm

cmd/build.go

@@ -75,7 +75,7 @@ func (cmd *BuildCmd) Run(f factory.Factory) error {
 }
 
 func (cmd *BuildCmd) runCommand(ctx *devspacecontext.Context, f factory.Factory, configOptions *loader.ConfigOptions) error {
-	return runPipeline(ctx, f, &PipelineOptions{
+	return runPipeline(ctx, f, true, &PipelineOptions{
 		Options: pipelinetypes.Options{
 			BuildOptions: build.Options{
 				Tags:                      cmd.Tags,

cmd/deploy.go

@@ -19,7 +19,6 @@ import (
 	"github.com/loft-sh/devspace/pkg/devspace/pipeline"
 	"github.com/loft-sh/devspace/pkg/devspace/pipeline/types"
 	"github.com/loft-sh/devspace/pkg/devspace/plugin"
-	"github.com/loft-sh/devspace/pkg/devspace/server"
 	"github.com/loft-sh/devspace/pkg/devspace/upgrade"
 	"github.com/loft-sh/devspace/pkg/util/factory"
 	logpkg "github.com/loft-sh/devspace/pkg/util/log"
@@ -29,6 +28,8 @@ import (
 	"github.com/spf13/cobra"
 	"gopkg.in/yaml.v3"
 	"k8s.io/client-go/kubernetes/fake"
+	"os"
+	"strings"
 )
 
 // DeployCmd holds the required data for the down cmd
@@ -124,7 +125,7 @@ func (cmd *DeployCmd) Run(f factory.Factory) error {
 }
 
 func (cmd *DeployCmd) runCommand(ctx *devspacecontext.Context, f factory.Factory, configOptions *loader.ConfigOptions) error {
-	return runPipeline(ctx, f, &PipelineOptions{
+	return runPipeline(ctx, f, true, &PipelineOptions{
 		Options: types.Options{
 			BuildOptions: build.Options{
 				SkipBuild:                 cmd.SkipBuild,
@@ -223,7 +224,7 @@ type PipelineOptions struct {
 	UIPort        int
 }
 
-func runPipeline(ctx *devspacecontext.Context, f factory.Factory, options *PipelineOptions) error {
+func runPipeline(ctx *devspacecontext.Context, f factory.Factory, forceLeader bool, options *PipelineOptions) error {
 	// create namespace if necessary
 	if !options.DeployOptions.Render {
 		err := ctx.KubeClient.EnsureNamespace(ctx.Context, ctx.KubeClient.Namespace(), ctx.Log)
@@ -242,12 +243,6 @@ func runPipeline(ctx *devspacecontext.Context, f factory.Factory, options *Pipel
 	}
 	ctx = ctx.WithDependencies(dependencies)
 
-	// start ui & open
-	serv, err := startServices(ctx, options.UIPort)
-	if err != nil {
-		return err
-	}
-
 	// execute plugin hook
 	err = hook.ExecuteHooks(ctx, nil, "deploy")
 	if err != nil {
@@ -270,29 +265,36 @@ func runPipeline(ctx *devspacecontext.Context, f factory.Factory, options *Pipel
 		}
 	}
 
-	// create dependency registry
-	dependencyRegistry := registry.NewDependencyRegistry("http://"+serv.Server.Addr, options.DeployOptions.Render)
-
-	// exclude ourselves
-	couldExclude, err := dependencyRegistry.MarkDependencyExcluded(ctx, ctx.Config.Config().Name, true)
-	if err != nil {
-		return err
-	} else if !couldExclude {
-		return fmt.Errorf("couldn't start project %s, because there is another DevSpace instance active in the current namespace right now that uses the same project", ctx.Config.Config().Name)
+	// marshal pipeline
+	configPipelineBytes, err := yaml.Marshal(configPipeline)
+	if err == nil {
+		ctx.Log.Debugf("Run pipeline:\n%s\n", string(configPipelineBytes))
 	}
 
 	// create a new base dev pod manager
 	devPodManager := devpod.NewManager(ctx.Context)
 	defer devPodManager.Close()
 
-	// marshal pipeline
-	configPipelineBytes, err := yaml.Marshal(configPipeline)
-	if err == nil {
-		ctx.Log.Debugf("Run pipeline:\n%s\n", string(configPipelineBytes))
-	}
+	// create dependency registry
+	dependencyRegistry := registry.NewDependencyRegistry(options.DeployOptions.Render)
 
 	// get deploy pipeline
-	pipe := pipeline.NewPipeline(options.Pipeline, devPodManager, dependencyRegistry, configPipeline, options.Options)
+	pipe := pipeline.NewPipeline(ctx.Config.Config().Name, devPodManager, dependencyRegistry, configPipeline, options.Options)
+
+	// start ui & open
+	serv, err := dev.UI(ctx, options.UIPort, pipe)
+	if err != nil {
+		return err
+	}
+	dependencyRegistry.SetServer("http://" + serv.Server.Addr)
+
+	// exclude ourselves
+	couldExclude, err := dependencyRegistry.MarkDependencyExcluded(ctx, ctx.Config.Config().Name, forceLeader)
+	if err != nil {
+		return err
+	} else if !couldExclude {
+		return fmt.Errorf("couldn't execute '%s', because there is another DevSpace instance active in the current namespace right now that uses the same project name (%s)", strings.Join(os.Args, " "), ctx.Config.Config().Name)
+	}
 
 	// start pipeline
 	err = pipe.Run(ctx.WithLogger(ctx.Log.WithoutPrefix()))
@@ -317,13 +319,3 @@ func runPipeline(ctx *devspacecontext.Context, f factory.Factory, options *Pipel
 
 	return nil
 }
-
-func startServices(ctx *devspacecontext.Context, uiPort int) (*server.Server, error) {
-	// Open UI if configured
-	serv, err := dev.UI(ctx, uiPort)
-	if err != nil {
-		return nil, err
-	}
-
-	return serv, nil
-}

cmd/dev.go

@@ -168,7 +168,7 @@ func (cmd *DevCmd) Run(f factory.Factory, args []string) error {
 }
 
 func (cmd *DevCmd) runCommand(ctx *devspacecontext.Context, f factory.Factory, configOptions *loader.ConfigOptions) error {
-	return runPipeline(ctx, f, &PipelineOptions{
+	return runPipeline(ctx, f, true, &PipelineOptions{
 		Options: types.Options{
 			BuildOptions: build.Options{
 				SkipBuild:                 cmd.SkipBuild,

cmd/purge.go

@@ -76,7 +76,7 @@ func (cmd *PurgeCmd) Run(f factory.Factory) error {
 }
 
 func (cmd *PurgeCmd) runCommand(ctx *devspacecontext.Context, f factory.Factory, configOptions *loader.ConfigOptions) error {
-	return runPipeline(ctx, f, &PipelineOptions{
+	return runPipeline(ctx, f, false, &PipelineOptions{
 		Options: types.Options{
 			DependencyOptions: types.DependencyOptions{
 				Exclude: cmd.SkipDependency,

cmd/render.go

@@ -91,7 +91,7 @@ func (cmd *RenderCmd) Run(f factory.Factory) error {
 }
 
 func (cmd *RenderCmd) runCommand(ctx *devspacecontext.Context, f factory.Factory, configOptions *loader.ConfigOptions) error {
-	return runPipeline(ctx, f, &PipelineOptions{
+	return runPipeline(ctx, f, true, &PipelineOptions{
 		Options: pipelinetypes.Options{
 			BuildOptions: build.Options{
 				Tags:                      cmd.Tags,

cmd/ui.go

@@ -189,7 +189,7 @@ func (cmd *UICmd) RunUI(f factory.Factory) error {
 	}
 
 	// Create server
-	server, err := server.NewServer(config, nil, cmd.Host, cmd.Dev, client.CurrentContext(), client.Namespace(), forcePort, cmd.log)
+	server, err := server.NewServer(ctx, cmd.Host, cmd.Dev, forcePort, nil)
 	if err != nil {
 		return err
 	}

examples/pipelines/dep2.yaml

@@ -0,0 +1,20 @@
+version: v2beta1
+name: pipelines2
+
+pipelines:
+  helper:
+    steps:
+      - run: |-
+          echo "Hello World!"
+  dev:
+    steps:
+      - run: |-
+          run_pipelines helper
+         
+          #build_images test --set image=${IMAGE}
+          
+          #create_deployments test2
+          
+          # Start development mode for test here
+          #start_dev test --set imageSelector=${runtime.images.test} \
+          #               --set terminal.command=sh

examples/pipelines/devspace.yaml

@@ -1,16 +1,24 @@
 version: v2beta1
 name: pipelines
 
+dependencies:
+  dep2:
+    source:
+      path: dep2.yaml
+
 pipelines:
   helper:
     steps:
       - run: |-
-          echo "Hello World!"
+          sleep 10
   dev:
     steps:
       - run: |-
-          export TEST=$(run_pipelines helper)
-          echo $TEST > swag.txt
+          sleep 30
+          #run_dependency_pipelines --all --pipeline dev
+          
+          #export TEST=$(run_pipelines helper)
+          #echo $TEST > swag.txt
           
           #build_images test --set image=${IMAGE}
           

pkg/devspace/dependency/registry/inter_process.go

@@ -1,6 +1,11 @@
 package registry
 
-import "context"
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"net/http"
+)
 
 type ExcludePayload struct {
 	// RunID is the run id of the process that owns the dependency
@@ -21,19 +26,55 @@ type InterProcess interface {
 	Ping(ctx context.Context, server string, payload *PingPayload) (bool, error)
 
 	// ExcludeDependency tells the remote server to exclude a certain dependency
-	ExcludeDependency(ctx context.Context, server string, excludePayload *ExcludePayload) error
+	ExcludeDependency(ctx context.Context, server string, excludePayload *ExcludePayload) (bool, error)
 }
 
 func NewInterProcessCommunicator() InterProcess {
-	return &dummyImplementation{}
+	return &requester{}
 }
 
-type dummyImplementation struct{}
+type requester struct{}
 
-func (d *dummyImplementation) Ping(ctx context.Context, server string, pingPayload *PingPayload) (bool, error) {
+func (d *requester) Ping(ctx context.Context, server string, pingPayload *PingPayload) (bool, error) {
+	out, err := json.Marshal(pingPayload)
+	if err != nil {
+		return false, err
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "POST", server+"/api/ping", bytes.NewReader(out))
+	if err != nil {
+		return false, err
+	}
+
+	response, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return false, err
+	}
+	defer response.Body.Close()
+	if response.StatusCode == http.StatusOK {
+		return true, nil
+	}
 	return false, nil
 }
 
-func (d *dummyImplementation) ExcludeDependency(ctx context.Context, server string, excludePayload *ExcludePayload) error {
-	return nil
+func (d *requester) ExcludeDependency(ctx context.Context, server string, excludePayload *ExcludePayload) (bool, error) {
+	out, err := json.Marshal(excludePayload)
+	if err != nil {
+		return false, err
+	}
+
+	req, err := http.NewRequestWithContext(ctx, "POST", server+"/api/exclude-dependency", bytes.NewReader(out))
+	if err != nil {
+		return false, err
+	}
+
+	response, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return false, err
+	}
+	defer response.Body.Close()
+	if response.StatusCode == http.StatusForbidden {
+		return false, nil
+	}
+	return true, nil
 }

pkg/devspace/dependency/registry/registry.go

@@ -30,13 +30,15 @@ type DependencyRegistry interface {
 
 	// MarkDependenciesExcluded same as MarkDependencyExcluded but for multiple dependencies
 	MarkDependenciesExcluded(ctx *devspacecontext.Context, dependencyNames []string, forceLeader bool) (map[string]bool, error)
+
+	// SetServer sets the target server
+	SetServer(server string)
 }
 
-func NewDependencyRegistry(server string, mock bool) DependencyRegistry {
+func NewDependencyRegistry(mock bool) DependencyRegistry {
 	return &dependencyRegistry{
-		server:               server,
 		mock:                 mock,
-		interProcess:         &dummyImplementation{},
+		interProcess:         NewInterProcessCommunicator(),
 		excludedDependencies: map[string]bool{},
 	}
 }
@@ -50,6 +52,13 @@ type dependencyRegistry struct {
 	excludedDependencies     map[string]bool
 }
 
+func (d *dependencyRegistry) SetServer(server string) {
+	d.excludedDependenciesLock.Lock()
+	defer d.excludedDependenciesLock.Unlock()
+
+	d.server = server
+}
+
 func (d *dependencyRegistry) ForceExclude(dependencyName string) {
 	d.excludedDependenciesLock.Lock()
 	defer d.excludedDependenciesLock.Unlock()
@@ -212,18 +221,22 @@ func (d *dependencyRegistry) excludeDependencies(ctx *devspacecontext.Context, d
 
 		// check if we should take over
 		if forceLeader {
-			err = d.interProcess.ExcludeDependency(ctx.Context, payload.Server, &ExcludePayload{
+			excludeCtx, excludeCancel := context.WithTimeout(ctx.Context, time.Second*10)
+			allowed, err := d.interProcess.ExcludeDependency(excludeCtx, payload.Server, &ExcludePayload{
 				RunID:          payload.RunID,
 				DependencyName: dependencyName,
 			})
-			if err != nil {
-				ctx.Log.Debugf("error taking over dependency: %v", err)
-			}
+			excludeCancel()
+			if err != nil || allowed {
+				if err != nil {
+					ctx.Log.Debugf("error taking over dependency: %v", err)
+				}
 
-			configMap.Data[dependencyName] = string(encoded)
-			retMap[dependencyName] = true
-			shouldUpdate = true
-			continue
+				configMap.Data[dependencyName] = string(encoded)
+				retMap[dependencyName] = true
+				shouldUpdate = true
+				continue
+			}
 		}
 	}