refactor: use global target selector

FabianKramm · FabianKramm · commit d399dfccaa1f · 2022-02-10T17:16:55.000+01:00
diff --git a/cmd/restart.go b/cmd/restart.go
@@ -178,7 +178,7 @@ func (cmd *RestartCmd) Run(f factory.Factory) error {
 
 func restartContainer(client kubectl.Client, options targetselector.Options, log log.Logger) error {
 	options = options.WithWait(false)
-	container, err := targetselector.NewTargetSelector(client).SelectSingleContainer(context.TODO(), options, log)
+	container, err := targetselector.GlobalTargetSelector.SelectSingleContainer(context.TODO(), client, options, log)
 	if err != nil {
 		return errors.Errorf("Error selecting pod: %v", err)
 	}
diff --git a/e2e/kube/kube.go b/e2e/kube/kube.go
@@ -44,7 +44,7 @@ func (k *KubeHelper) ExecByImageSelector(imageSelector, namespace string, comman
 		WithTimeout(120).
 		WithWaitingStrategy(targetselector.NewUntilNewestRunningWaitingStrategy(time.Second * 2))
 
-	container, err := targetselector.NewTargetSelector(k.client).SelectSingleContainer(context.TODO(), targetOptions, log.Discard)
+	container, err := targetselector.GlobalTargetSelector.SelectSingleContainer(context.TODO(), k.client, targetOptions, log.Discard)
 	if err != nil {
 		return "", err
 	}
@@ -62,7 +62,7 @@ func (k *KubeHelper) ExecByContainer(labelSelector, containerName, namespace str
 		WithTimeout(120).
 		WithWaitingStrategy(targetselector.NewUntilNewestRunningWaitingStrategy(time.Second * 2))
 
-	container, err := targetselector.NewTargetSelector(k.client).SelectSingleContainer(context.TODO(), targetOptions, log.Discard)
+	container, err := targetselector.GlobalTargetSelector.SelectSingleContainer(context.TODO(), k.client, targetOptions, log.Discard)
 	if err != nil {
 		return "", err
 	}
diff --git a/pkg/devspace/hook/remote_hook.go b/pkg/devspace/hook/remote_hook.go
@@ -93,17 +93,14 @@ func (r *remoteHook) execute(hook *latest.HookConfig, imageSelector []imageselec
 		wait = true
 	}
 
-	// select the container
-	targetSelector := targetselector.NewTargetSelector(client)
-
 	// build target selector
 	targetSelectorOptions := targetselector.NewOptionsFromFlags(hook.Container.ContainerName, labelSelector, targetselector.ToStringImageSelector(imageSelector), hook.Container.Namespace, hook.Container.Pod).
 		WithTimeout(timeout).
 		WithWait(wait).
 		WithWaitingStrategy(r.WaitingStrategy)
 
 	// select container
-	podContainer, err := targetSelector.SelectSingleContainer(context.TODO(), targetSelectorOptions, log)
+	podContainer, err := targetselector.GlobalTargetSelector.SelectSingleContainer(context.TODO(), client, targetSelectorOptions, log)
 	if err != nil {
 		if _, ok := err.(*targetselector.NotFoundErr); ok {
 			return false, nil
diff --git a/pkg/devspace/kubectl/selector/selector.go b/pkg/devspace/kubectl/selector/selector.go
@@ -71,14 +71,14 @@ type SelectedPodContainer struct {
 }
 
 type Selector struct {
-	ImageSelector      []string
-	LabelSelector      string
-	Pod                string
-	ContainerName      string
-	Namespace          string
-	SkipInitContainers bool
-
-	FilterContainer FilterContainer
+	ImageSelector      []string `json:"imageSelector"`
+	LabelSelector      string   `json:"labelSelector"`
+	Pod                string   `json:"pod"`
+	ContainerName      string   `json:"containerName"`
+	Namespace          string   `json:"namespace"`
+	SkipInitContainers bool     `json:"skipInitContainers"`
+
+	FilterContainer FilterContainer `json:"-"`
 }
 
 func (s Selector) String() string {
diff --git a/pkg/devspace/services/attach.go b/pkg/devspace/services/attach.go
@@ -11,10 +11,8 @@ import (
 
 // StartAttach opens a new terminal
 func (serviceClient *client) StartAttach(options targetselector.Options, interrupt chan error) error {
-	targetSelector := targetselector.NewTargetSelector(serviceClient.client)
 	options = options.WithQuestion("Which pod do you want to attach to?")
-
-	container, err := targetSelector.SelectSingleContainer(context.TODO(), options, serviceClient.log)
+	container, err := targetselector.GlobalTargetSelector.SelectSingleContainer(context.TODO(), serviceClient.client, options, serviceClient.log)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/devspace/services/logs.go b/pkg/devspace/services/logs.go
@@ -18,14 +18,13 @@ func (serviceClient *client) StartLogs(options targetselector.Options, follow bo
 
 // StartLogsWithWriter prints the logs and then attaches to the container with the given stdout and stderr
 func (serviceClient *client) StartLogsWithWriter(options targetselector.Options, follow bool, tail int64, wait bool, writer io.Writer) error {
-	targetSelector := targetselector.NewTargetSelector(serviceClient.client)
 	options = options.WithWait(wait).
 		WithContainerFilter(selector.FilterTerminatingContainers)
 	if wait {
 		options = options.WithWaitingStrategy(targetselector.NewUntilNotWaitingStrategy(time.Second * 2))
 	}
 
-	container, err := targetSelector.SelectSingleContainer(context.TODO(), options, serviceClient.log)
+	container, err := targetselector.GlobalTargetSelector.SelectSingleContainer(context.TODO(), serviceClient.client, options, serviceClient.log)
 	if err != nil {
 		return err
 	}
diff --git a/pkg/devspace/services/podreplace/persist.go b/pkg/devspace/services/podreplace/persist.go
@@ -72,7 +72,7 @@ func persistPaths(podName string, replacePod *latest.ReplacePod, copiedPod *core
 			Name:    fmt.Sprintf("path-%d-init", i),
 			Image:   container.Image,
 			Command: []string{"sh"},
-			Args:    []string{"-c", fmt.Sprintf("if [ ! -d \"/devspace-persistence/.devspace/\" ] && [ -d \"%s\" ]; then\n cp --verbose -a %s/. /devspace-persistence/ && mkdir /devspace-persistence/.devspace\n sleep 20\nfi", path.Clean(p.Path), path.Clean(p.Path))},
+			Args:    []string{"-c", fmt.Sprintf("if [ ! -d \"/devspace-persistence/.devspace/\" ] && [ -d \"%s\" ]; then\n echo 'DevSpace is initializing the sync volume...'\n cp -a %s/. /devspace-persistence/ && mkdir /devspace-persistence/.devspace\nfi", path.Clean(p.Path), path.Clean(p.Path))},
 			VolumeMounts: []corev1.VolumeMount{
 				{
 					Name:      "devspace-persistence",
diff --git a/pkg/devspace/services/podreplace/replace.go b/pkg/devspace/services/podreplace/replace.go
@@ -833,7 +833,7 @@ func findSingleReplacedPod(ctx context.Context, client kubectl.Client, replacePo
 		WithSkipInitContainers(true)
 
 	// get container
-	selected, err := targetselector.NewTargetSelector(client).SelectSingleContainer(ctx, targetOptions, log)
+	selected, err := targetselector.GlobalTargetSelector.SelectSingleContainer(ctx, client, targetOptions, log)
 	if err != nil {
 		return nil, err
 	}
@@ -918,7 +918,7 @@ func findSingleReplaceablePodParent(ctx context.Context, client kubectl.Client,
 		WithWaitingStrategy(targetselector.NewUntilNotTerminatingStrategy(time.Second * 2)).
 		WithSkipInitContainers(true)
 
-	container, err := targetselector.NewTargetSelector(client).SelectSingleContainer(ctx, targetOptions, log)
+	container, err := targetselector.GlobalTargetSelector.SelectSingleContainer(ctx, client, targetOptions, log)
 	if err != nil {
 		return nil, nil, err
 	}
diff --git a/pkg/devspace/services/port_forwarding.go b/pkg/devspace/services/port_forwarding.go
@@ -128,7 +128,7 @@ func (serviceClient *client) startForwarding(portForwarding *latest.PortForwardi
 
 	// start port forwarding
 	log.Info("Port-Forwarding: Waiting for containers to start...")
-	pod, err := targetselector.NewTargetSelector(serviceClient.client).SelectSinglePod(context.TODO(), options, log)
+	pod, err := targetselector.GlobalTargetSelector.SelectSinglePod(context.TODO(), serviceClient.client, options, log)
 	if err != nil {
 		return errors.Errorf("%s: %s", message.SelectorErrorPod, err.Error())
 	} else if pod == nil {
diff --git a/pkg/devspace/services/reverse_port_forwarding.go b/pkg/devspace/services/reverse_port_forwarding.go
@@ -39,7 +39,7 @@ func (serviceClient *client) startReversePortForwarding(portForwarding *latest.P
 		WithSkipInitContainers(true)
 
 	log.Info("Reverse-Port-Forwarding: Waiting for containers to start...")
-	container, err := targetselector.NewTargetSelector(serviceClient.client).SelectSingleContainer(context.TODO(), options, log)
+	container, err := targetselector.GlobalTargetSelector.SelectSingleContainer(context.TODO(), serviceClient.client, options, log)
 	if err != nil {
 		return errors.Errorf("%s: %s", message.SelectorErrorPod, err.Error())
 	}
diff --git a/pkg/devspace/services/synccontroller/controller.go b/pkg/devspace/services/synccontroller/controller.go
@@ -289,7 +289,7 @@ func (c *controller) startSync(options *Options, onInitUploadDone chan struct{},
 	}
 
 	log.Info("Waiting for containers to start...")
-	container, err := targetselector.NewTargetSelector(c.client).SelectSingleContainer(context.TODO(), options.TargetOptions, log)
+	container, err := targetselector.GlobalTargetSelector.SelectSingleContainer(context.TODO(), c.client, options.TargetOptions, log)
 	if err != nil {
 		return nil, nil, errors.Errorf("Error selecting pod: %v", err)
 	}
diff --git a/pkg/devspace/services/targetselector/target_selector.go b/pkg/devspace/services/targetselector/target_selector.go
@@ -2,8 +2,11 @@ package targetselector
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"github.com/loft-sh/devspace/pkg/devspace/imageselector"
+	"github.com/loft-sh/devspace/pkg/util/lockfactory"
+	"sync"
 	"time"
 
 	"github.com/loft-sh/devspace/pkg/devspace/kubectl/selector"
@@ -150,24 +153,34 @@ func ToStringImageSelector(imageSelector []imageselector.ImageSelector) []string
 }
 
 type TargetSelector interface {
-	SelectSinglePod(ctx context.Context, options Options, log log.Logger) (*v1.Pod, error)
-	SelectSingleContainer(ctx context.Context, options Options, log log.Logger) (*selector.SelectedPodContainer, error)
+	SelectSinglePod(ctx context.Context, client kubectl.Client, options Options, log log.Logger) (*v1.Pod, error)
+	SelectSingleContainer(ctx context.Context, client kubectl.Client, options Options, log log.Logger) (*selector.SelectedPodContainer, error)
 }
 
+// GlobalTargetSelector is used to make sure we don't wait for the same pod
+// multiple times concurrently.
+var GlobalTargetSelector = NewTargetSelector()
+
 // targetSelector is the struct that will select a target
 type targetSelector struct {
-	client kubectl.Client
+	// lock factory is used to make sure we don't wait for the same pod
+	// multiple times.
+	lockFactory lockfactory.LockFactory
 }
 
 // NewTargetSelector creates a new target selector for selecting a target pod or container
-func NewTargetSelector(client kubectl.Client) TargetSelector {
+func NewTargetSelector() TargetSelector {
 	return &targetSelector{
-		client: client,
+		lockFactory: lockfactory.NewDefaultLockFactory(),
 	}
 }
 
-func (t *targetSelector) SelectSingleContainer(ctx context.Context, options Options, log log.Logger) (*selector.SelectedPodContainer, error) {
-	container, err := t.selectSingle(ctx, options, log, t.selectSingleContainer)
+func (t *targetSelector) SelectSingleContainer(ctx context.Context, client kubectl.Client, options Options, log log.Logger) (*selector.SelectedPodContainer, error) {
+	lock := t.getSelectorLock(options.selector)
+	lock.Lock()
+	defer lock.Unlock()
+
+	container, err := t.selectSingle(ctx, client, options, log, t.selectSingleContainer)
 	if err != nil {
 		return nil, err
 	} else if container == nil {
@@ -177,8 +190,12 @@ func (t *targetSelector) SelectSingleContainer(ctx context.Context, options Opti
 	return container.(*selector.SelectedPodContainer), nil
 }
 
-func (t *targetSelector) SelectSinglePod(ctx context.Context, options Options, log log.Logger) (*v1.Pod, error) {
-	pod, err := t.selectSingle(ctx, options, log, t.selectSinglePod)
+func (t *targetSelector) SelectSinglePod(ctx context.Context, client kubectl.Client, options Options, log log.Logger) (*v1.Pod, error) {
+	lock := t.getSelectorLock(options.selector)
+	lock.Lock()
+	defer lock.Unlock()
+
+	pod, err := t.selectSingle(ctx, client, options, log, t.selectSinglePod)
 	if err != nil {
 		return nil, err
 	} else if pod == nil {
@@ -188,7 +205,12 @@ func (t *targetSelector) SelectSinglePod(ctx context.Context, options Options, l
 	return pod.(*v1.Pod), nil
 }
 
-func (t *targetSelector) selectSingle(ctx context.Context, options Options, log log.Logger, selectFn func(ctx context.Context, options Options, log log.Logger) (bool, interface{}, error)) (interface{}, error) {
+func (t *targetSelector) getSelectorLock(s selector.Selector) sync.Locker {
+	out, _ := json.Marshal(s)
+	return t.lockFactory.GetLock(string(out))
+}
+
+func (t *targetSelector) selectSingle(ctx context.Context, client kubectl.Client, options Options, log log.Logger, selectFn func(ctx context.Context, client kubectl.Client, options Options, log log.Logger) (bool, interface{}, error)) (interface{}, error) {
 	if options.wait == nil || *options.wait {
 		timeout := time.Minute * 10
 		if options.timeout > 0 {
@@ -197,7 +219,7 @@ func (t *targetSelector) selectSingle(ctx context.Context, options Options, log
 
 		var out interface{}
 		err := wait.PollImmediate(time.Second, timeout, func() (done bool, err error) {
-			done, o, err := selectFn(ctx, options, log)
+			done, o, err := selectFn(ctx, client, options, log)
 			if err != nil {
 				return false, err
 			} else if !done {
@@ -222,7 +244,7 @@ func (t *targetSelector) selectSingle(ctx context.Context, options Options, log
 	}
 
 	// we try to select a pod
-	done, out, err := selectFn(ctx, options, log)
+	done, out, err := selectFn(ctx, client, options, log)
 	if err != nil {
 		return nil, err
 	} else if !done {
@@ -235,12 +257,12 @@ func (t *targetSelector) selectSingle(ctx context.Context, options Options, log
 	return out, nil
 }
 
-func (t *targetSelector) selectSingleContainer(ctx context.Context, options Options, log log.Logger) (bool, interface{}, error) {
-	containers, err := selector.NewFilterWithSort(t.client, options.sortContainers).SelectContainers(ctx, options.selector)
+func (t *targetSelector) selectSingleContainer(ctx context.Context, client kubectl.Client, options Options, log log.Logger) (bool, interface{}, error) {
+	containers, err := selector.NewFilterWithSort(client, options.sortContainers).SelectContainers(ctx, options.selector)
 	if err != nil {
 		return false, nil, err
 	} else if options.waitingStrategy != nil {
-		return options.waitingStrategy.SelectContainer(ctx, t.client, options.selector.Namespace, containers, log)
+		return options.waitingStrategy.SelectContainer(ctx, client, options.selector.Namespace, containers, log)
 	}
 
 	if len(containers) == 0 {
@@ -285,16 +307,16 @@ func (t *targetSelector) selectSingleContainer(ctx context.Context, options Opti
 	return true, containers[0], nil
 }
 
-func (t *targetSelector) selectSinglePod(ctx context.Context, options Options, log log.Logger) (bool, interface{}, error) {
-	stack, err := selector.NewFilterWithSort(t.client, options.sortContainers).SelectContainers(ctx, options.selector)
+func (t *targetSelector) selectSinglePod(ctx context.Context, client kubectl.Client, options Options, log log.Logger) (bool, interface{}, error) {
+	stack, err := selector.NewFilterWithSort(client, options.sortContainers).SelectContainers(ctx, options.selector)
 	if err != nil {
 		return false, nil, err
 	}
 
 	// transform stack
 	pods := selector.PodsFromPodContainer(stack)
 	if options.waitingStrategy != nil {
-		return options.waitingStrategy.SelectPod(ctx, t.client, options.selector.Namespace, pods, log)
+		return options.waitingStrategy.SelectPod(ctx, client, options.selector.Namespace, pods, log)
 	}
 
 	if len(pods) == 0 {
diff --git a/pkg/devspace/services/terminal.go b/pkg/devspace/services/terminal.go
@@ -36,16 +36,14 @@ func (serviceClient *client) StartTerminal(
 	stdin io.Reader,
 ) (int, error) {
 	command := serviceClient.getCommand(args, workDir)
-	targetSelector := targetselector.NewTargetSelector(serviceClient.client)
-
 	options = options.WithWait(wait).
 		WithQuestion("Which pod do you want to open the terminal for?")
 	if wait {
 		options = options.WithContainerFilter(selector.FilterTerminatingContainers)
 		options = options.WithWaitingStrategy(targetselector.NewUntilNewestRunningWaitingStrategy(time.Second))
 	}
 
-	container, err := targetSelector.SelectSingleContainer(context.TODO(), options, serviceClient.log)
+	container, err := targetselector.GlobalTargetSelector.SelectSingleContainer(context.TODO(), serviceClient.client, options, serviceClient.log)
 	if err != nil {
 		return 0, err
 	}
diff --git a/pkg/util/lockfactory/lockfactory.go b/pkg/util/lockfactory/lockfactory.go
@@ -0,0 +1,63 @@
+/*
+Copyright 2020 DevSpace Technologies Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package lockfactory
+
+import (
+	"sync"
+)
+
+// LockFactory is the interface to retrieve named locks from
+type LockFactory interface {
+	GetLock(string) sync.Locker
+}
+
+type defaultLockFactory struct {
+	lock  sync.RWMutex
+	locks map[string]sync.Locker
+}
+
+// NewDefaultLockFactory creates a new lock factory
+func NewDefaultLockFactory() LockFactory {
+	return &defaultLockFactory{locks: map[string]sync.Locker{}}
+}
+
+func (f *defaultLockFactory) GetLock(key string) sync.Locker {
+	lock, exists := f.getExistingLock(key)
+	if exists {
+		return lock
+	}
+
+	f.lock.Lock()
+	defer f.lock.Unlock()
+
+	lock, exists = f.locks[key]
+	if exists {
+		return lock
+	}
+
+	lock = &sync.Mutex{}
+	f.locks[key] = lock
+	return lock
+}
+
+func (f *defaultLockFactory) getExistingLock(key string) (sync.Locker, bool) {
+	f.lock.RLock()
+	defer f.lock.RUnlock()
+
+	lock, exists := f.locks[key]
+	return lock, exists
+}

Original file line number	Diff line number	Diff line change
`@@ -178,7 +178,7 @@ func (cmd *RestartCmd) Run(f factory.Factory) error {`
`178`	`178`
`179`	`179`	`func restartContainer(client kubectl.Client, options targetselector.Options, log log.Logger) error {`
`180`	`180`	`options = options.WithWait(false)`
`181`		`- container, err := targetselector.NewTargetSelector(client).SelectSingleContainer(context.TODO(), options, log)`
	`181`	`+ container, err := targetselector.GlobalTargetSelector.SelectSingleContainer(context.TODO(), client, options, log)`
`182`	`182`	`if err != nil {`
`183`	`183`	`return errors.Errorf("Error selecting pod: %v", err)`
`184`	`184`	`}`
Original file line number	Diff line number	Diff line change
`@@ -18,14 +18,13 @@ func (serviceClient *client) StartLogs(options targetselector.Options, follow bo`
`18`	`18`
`19`	`19`	`// StartLogsWithWriter prints the logs and then attaches to the container with the given stdout and stderr`
`20`	`20`	`func (serviceClient *client) StartLogsWithWriter(options targetselector.Options, follow bool, tail int64, wait bool, writer io.Writer) error {`
`21`		`- targetSelector := targetselector.NewTargetSelector(serviceClient.client)`
`22`	`21`	`options = options.WithWait(wait).`
`23`	`22`	`WithContainerFilter(selector.FilterTerminatingContainers)`
`24`	`23`	`if wait {`
`25`	`24`	`options = options.WithWaitingStrategy(targetselector.NewUntilNotWaitingStrategy(time.Second * 2))`
`26`	`25`	`}`
`27`	`26`
`28`		`- container, err := targetSelector.SelectSingleContainer(context.TODO(), options, serviceClient.log)`
	`27`	`+ container, err := targetselector.GlobalTargetSelector.SelectSingleContainer(context.TODO(), serviceClient.client, options, serviceClient.log)`
`29`	`28`	`if err != nil {`
`30`	`29`	`return err`
`31`	`30`	`}`
Original file line number	Diff line number	Diff line change
`@@ -72,7 +72,7 @@ func persistPaths(podName string, replacePod latest.ReplacePod, copiedPod core`
`72`	`72`	`Name: fmt.Sprintf("path-%d-init", i),`
`73`	`73`	`Image: container.Image,`
`74`	`74`	`Command: []string{"sh"},`
`75`		`- Args: []string{"-c", fmt.Sprintf("if [ ! -d \"/devspace-persistence/.devspace/\" ] && [ -d \"%s\" ]; then\n cp --verbose -a %s/. /devspace-persistence/ && mkdir /devspace-persistence/.devspace\n sleep 20\nfi", path.Clean(p.Path), path.Clean(p.Path))},`
	`75`	`+ Args: []string{"-c", fmt.Sprintf("if [ ! -d \"/devspace-persistence/.devspace/\" ] && [ -d \"%s\" ]; then\n echo 'DevSpace is initializing the sync volume...'\n cp -a %s/. /devspace-persistence/ && mkdir /devspace-persistence/.devspace\nfi", path.Clean(p.Path), path.Clean(p.Path))},`
`76`	`76`	`VolumeMounts: []corev1.VolumeMount{`
`77`	`77`	`{`
`78`	`78`	`Name: "devspace-persistence",`
Original file line number	Diff line number	Diff line change
`@@ -39,7 +39,7 @@ func (serviceClient client) startReversePortForwarding(portForwarding latest.P`
`39`	`39`	`WithSkipInitContainers(true)`
`40`	`40`
`41`	`41`	`log.Info("Reverse-Port-Forwarding: Waiting for containers to start...")`
`42`		`- container, err := targetselector.NewTargetSelector(serviceClient.client).SelectSingleContainer(context.TODO(), options, log)`
	`42`	`+ container, err := targetselector.GlobalTargetSelector.SelectSingleContainer(context.TODO(), serviceClient.client, options, log)`
`43`	`43`	`if err != nil {`
`44`	`44`	`return errors.Errorf("%s: %s", message.SelectorErrorPod, err.Error())`
`45`	`45`	`}`