Repository: helm/helm (go)

Issue:
v3: Can't add very first repo
The acceptance-tests started failing after #6492 (had to do a git bisect to figure that out).

The reason is that it is no longer possible to add the very first repo when starting from an empty environment as shown below:
```
> export XDG_CONFIG_HOME=/tmp/test/config
> mkdir -p ${XDG_CONFIG_HOME}

> $GOPATH/src/helm.sh/helm/bin/helm repo add gitlab https://charts.gitlab.io
Error: open /tmp/test/config/helm/repositories.yaml: no such file or directory

> /bin/rm -rf ${XDG_CONFIG_HOME}
```

The reason is that the locking of repository file expects the directory of the `repositories.yaml` file to exist.

I first thought of ignoring the error if it was a missing file error, but then I thought that the race the locking logic is fixing could also happen if two `helm repo add` try to create the `repositories.yaml` file concurrently.  I don't know how to fix that...



Pull request:
Create file locking directory if it does not exist
As Helm v3 uses lazy creation for configuration, directories and files are not created until required. File locking when doing repo add was introduced in v2 and ported to v3 in #6492. It locks on the config directory where the repo file resides and therefore needs the directory to exist. This fix adds the directory if it doesn't exist.

Closes #6521 

**What this PR does / why we need it**:
This fixes the issue where user cannot add the first repository. Bug was introduced in #6492.

**Special notes for your reviewer**:

**If applicable**:
- [ ] this PR contains documentation
- [ ] this PR contains unit tests
- [ ] this PR has been tested for backwards compatibility


Patch:
diff --git a/cmd/helm/repo_add.go b/cmd/helm/repo_add.go
index 96162499f..a28bb82a1 100644
--- a/cmd/helm/repo_add.go
+++ b/cmd/helm/repo_add.go
@@ -22,6 +22,7 @@ import (
 	"io"
 	"io/ioutil"
 	"os"
+	"path/filepath"
 	"time"
 
 	"github.com/gofrs/flock"
@@ -78,6 +79,12 @@ func newRepoAddCmd(out io.Writer) *cobra.Command {
 }
 
 func (o *repoAddOptions) run(out io.Writer) error {
+	//Ensure the file directory exists as it is required for file locking
+	err := os.MkdirAll(filepath.Dir(o.repoFile), os.ModePerm)
+	if err != nil && !os.IsExist(err) {
+		return err
+	}
+
 	// Lock the repository file for concurrent goroutines or processes synchronization
 	fileLock := flock.New(o.repoFile)
 	lockCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
diff --git a/pkg/kube/client.go b/pkg/kube/client.go
index 6f5ce44d2..3f228a583 100644
--- a/pkg/kube/client.go
+++ b/pkg/kube/client.go
@@ -23,6 +23,7 @@ import (
 	"io"
 	"log"
 	"strings"
+	"sync"
 	"time"
 
 	jsonpatch "github.com/evanphx/json-patch"
@@ -273,14 +274,35 @@ func perform(infos ResourceList, fn func(*resource.Info) error) error {
 		return ErrNoObjectsVisited
 	}
 
-	for _, info := range infos {
-		if err := fn(info); err != nil {
+	errs := make(chan error)
+	go batchPerform(infos, fn, errs)
+
+	for range infos {
+		err := <-errs
+		if err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
+func batchPerform(infos ResourceList, fn func(*resource.Info) error, errs chan<- error) {
+	var kind string
+	var wg sync.WaitGroup
+	for _, info := range infos {
+		currentKind := info.Object.GetObjectKind().GroupVersionKind().Kind
+		if kind != currentKind {
+			wg.Wait()
+			kind = currentKind
+		}
+		wg.Add(1)
+		go func(i *resource.Info) {
+			errs <- fn(i)
+			wg.Done()
+		}(info)
+	}
+}
+
 func createResource(info *resource.Info) error {
 	obj, err := resource.NewHelper(info.Client, info.Mapping).Create(info.Namespace, true, info.Object, nil)
 	if err != nil {


Test patch:
diff --git a/cmd/helm/repo_add_test.go b/cmd/helm/repo_add_test.go
index c53f3bd65..1ac61fba8 100644
--- a/cmd/helm/repo_add_test.go
+++ b/cmd/helm/repo_add_test.go
@@ -92,15 +92,23 @@ func TestRepoAdd(t *testing.T) {
 
 func TestRepoAddConcurrentGoRoutines(t *testing.T) {
 	const testName = "test-name"
+	repoFile := filepath.Join(ensure.TempDir(t), "repositories.yaml")
+	repoAddConcurrent(t, testName, repoFile)
+}
+
+func TestRepoAddConcurrentDirNotExist(t *testing.T) {
+	const testName = "test-name-2"
+	repoFile := filepath.Join(ensure.TempDir(t), "foo", "repositories.yaml")
+	repoAddConcurrent(t, testName, repoFile)
+}
 
+func repoAddConcurrent(t *testing.T, testName, repoFile string) {
 	ts, err := repotest.NewTempServer("testdata/testserver/*.*")
 	if err != nil {
 		t.Fatal(err)
 	}
 	defer ts.Stop()
 
-	repoFile := filepath.Join(ensure.TempDir(t), "repositories.yaml")
-
 	var wg sync.WaitGroup
 	wg.Add(3)
 	for i := 0; i < 3; i++ {