Merge branch 'master' into mq-subscribe

6 files changed, 37 insertions, 20 deletions

diff --git a/go.mod b/go.mod
index cffdca9c1..518e60a50 100644
--- a/go.mod
+++ b/go.mod
@@ -159,7 +159,7 @@ require (
 	github.com/shirou/gopsutil/v3 v3.23.11
 	github.com/tikv/client-go/v2 v2.0.7
 	github.com/ydb-platform/ydb-go-sdk-auth-environ v0.2.0
-	github.com/ydb-platform/ydb-go-sdk/v3 v3.53.3
+	github.com/ydb-platform/ydb-go-sdk/v3 v3.54.2
 	google.golang.org/grpc/security/advancedtls v0.0.0-20220622233350-5cdb09fa29c1
 )
 
diff --git a/go.sum b/go.sum
index 720b932fa..db1097f01 100644
--- a/go.sum
+++ b/go.sum
@@ -961,8 +961,8 @@ github.com/ydb-platform/ydb-go-sdk-auth-environ v0.2.0 h1:IG5bPd+Lqyc+zsw2kmxqfG
 github.com/ydb-platform/ydb-go-sdk-auth-environ v0.2.0/go.mod h1:l6lZ+osdQOjDRBgRA4PQ06BuvmXN2neYjnRw8rCfd7s=
 github.com/ydb-platform/ydb-go-sdk/v3 v3.25.3/go.mod h1:PFizF/vJsdAgEwjK3DVSBD52kdmRkWfSIS2q2pA+e88=
 github.com/ydb-platform/ydb-go-sdk/v3 v3.47.3/go.mod h1:bWnOIcUHd7+Sl7DN+yhyY1H/I61z53GczvwJgXMgvj0=
-github.com/ydb-platform/ydb-go-sdk/v3 v3.53.3 h1:yv0RqgpJUtx32CJzU0se1fi+AJZ1V+F3wD4mP0ydkls=
-github.com/ydb-platform/ydb-go-sdk/v3 v3.53.3/go.mod h1:qtQwB//TJJPH4YisuX/87ZNplf09eAP1FhrrnGJG6ME=
+github.com/ydb-platform/ydb-go-sdk/v3 v3.54.2 h1:E0yUuuX7UmPxXm92+yQCjMveLFO3zfvYFIJVuAqsVRA=
+github.com/ydb-platform/ydb-go-sdk/v3 v3.54.2/go.mod h1:fjBLQ2TdQNl4bMjuWl9adoTGBypwUTPoGC+EqYqiIcU=
 github.com/ydb-platform/ydb-go-yc v0.10.2 h1:RAHy6g7ncxk1y0N4oS2MwYXLATqRqKBI6DYXuxpV2wo=
 github.com/ydb-platform/ydb-go-yc v0.10.2/go.mod h1:U1dX3LJy6zADId2DciCXlgrU/vphK1+CQzaefKq21dQ=
 github.com/ydb-platform/ydb-go-yc-metadata v0.5.2 h1:nMtixUijP0Z7iHJNT9fOL+dbmEzZxqU6Xk87ll7hqXg=
diff --git a/k8s/charts/seaweedfs/templates/master-servicemonitor.yaml b/k8s/charts/seaweedfs/templates/master-servicemonitor.yaml
index 0332f756d..e9844fb8b 100644
--- a/k8s/charts/seaweedfs/templates/master-servicemonitor.yaml
+++ b/k8s/charts/seaweedfs/templates/master-servicemonitor.yaml
@@ -18,7 +18,7 @@ spec:
       scrapeTimeout: 5s
   selector:
     matchLabels:
-      app: {{ template "seaweedfs.name" . }}
-      component: master
+      app.kubernetes.io/name: {{ template "seaweedfs.name" . }}
+      app.kubernetes.io/component: master
 {{- end }}
 {{- end }}
\ No newline at end of file
diff --git a/k8s/charts/seaweedfs/templates/master-statefulset.yaml b/k8s/charts/seaweedfs/templates/master-statefulset.yaml
index 6e432450a..03d5d2959 100644
--- a/k8s/charts/seaweedfs/templates/master-statefulset.yaml
+++ b/k8s/charts/seaweedfs/templates/master-statefulset.yaml
@@ -127,7 +127,8 @@ spec:
               -metrics.intervalSeconds={{ .Values.master.metricsIntervalSec }} \
               {{- end }}
               {{- end }}
-              {{- else if .Values.master.metricsPort }}
+              {{- end }}
+              {{- if .Values.master.metricsPort }}
               -metricsPort={{ .Values.master.metricsPort }} \
               {{- end }}
               -volumeSizeLimitMB={{ .Values.master.volumeSizeLimitMB }} \
diff --git a/weed/server/volume_grpc_read_all.go b/weed/server/volume_grpc_read_all.go
index 3f3ecf0b5..68001ce9e 100644
--- a/weed/server/volume_grpc_read_all.go
+++ b/weed/server/volume_grpc_read_all.go
@@ -10,14 +10,14 @@ import (
 func (vs *VolumeServer) ReadAllNeedles(req *volume_server_pb.ReadAllNeedlesRequest, stream volume_server_pb.VolumeServer_ReadAllNeedlesServer) (err error) {
 
 	for _, vid := range req.VolumeIds {
-		if err := vs.streamReadOneVolume(needle.VolumeId(vid), stream, err); err != nil {
+		if err := vs.streamReadOneVolume(needle.VolumeId(vid), stream); err != nil {
 			return err
 		}
 	}
 	return nil
 }
 
-func (vs *VolumeServer) streamReadOneVolume(vid needle.VolumeId, stream volume_server_pb.VolumeServer_ReadAllNeedlesServer, err error) error {
+func (vs *VolumeServer) streamReadOneVolume(vid needle.VolumeId, stream volume_server_pb.VolumeServer_ReadAllNeedlesServer) error {
 	v := vs.store.GetVolume(vid)
 	if v == nil {
 		return fmt.Errorf("not found volume id %d", vid)
@@ -30,7 +30,5 @@ func (vs *VolumeServer) streamReadOneVolume(vid needle.VolumeId, stream volume_s
 
 	offset := int64(v.SuperBlock.BlockSize())
 
-	err = storage.ScanVolumeFileFrom(v.Version(), v.DataBackend, offset, scanner)
-
-	return err
+	return storage.ScanVolumeFileFrom(v.Version(), v.DataBackend, offset, scanner)
 }
diff --git a/weed/shell/command_fs_meta_load.go b/weed/shell/command_fs_meta_load.go
index 0cbdddb49..a2ae9401d 100644
--- a/weed/shell/command_fs_meta_load.go
+++ b/weed/shell/command_fs_meta_load.go
@@ -6,6 +6,7 @@ import (
 	"io"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	"google.golang.org/protobuf/proto"
@@ -47,6 +48,7 @@ func (c *commandFsMetaLoad) Do(args []string, commandEnv *CommandEnv, writer io.
 
 	metaLoadCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
 	c.dirPrefix = metaLoadCommand.String("dirPrefix", "", "load entries only with directories matching prefix")
+	concurrency := metaLoadCommand.Int("concurrency", 1, "number of parallel meta load to filer")
 	verbose := metaLoadCommand.Bool("v", true, "verbose mode")
 	if err = metaLoadCommand.Parse(args[0 : len(args)-1]); err != nil {
 		return nil
@@ -64,6 +66,9 @@ func (c *commandFsMetaLoad) Do(args []string, commandEnv *CommandEnv, writer io.
 	err = commandEnv.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
 
 		sizeBuf := make([]byte, 4)
+		waitChan := make(chan struct{}, *concurrency)
+		defer close(waitChan)
+		var wg sync.WaitGroup
 
 		for {
 			if n, err := dst.Read(sizeBuf); n != 4 {
@@ -105,21 +110,34 @@ func (c *commandFsMetaLoad) Do(args []string, commandEnv *CommandEnv, writer io.
 			}
 
 			fullEntry.Entry.Name = strings.ReplaceAll(fullEntry.Entry.Name, "/", "x")
-			if err := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
-				Directory: fullEntry.Dir,
-				Entry:     fullEntry.Entry,
-			}); err != nil {
-				return err
-			}
-
 			if fullEntry.Entry.IsDirectory {
+				wg.Wait()
+				if errEntry := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
+					Directory: fullEntry.Dir,
+					Entry:     fullEntry.Entry,
+				}); errEntry != nil {
+					return errEntry
+				}
 				dirCount++
 			} else {
+				wg.Add(1)
+				waitChan <- struct{}{}
+				go func(entry *filer_pb.FullEntry) {
+					if errEntry := filer_pb.CreateEntry(client, &filer_pb.CreateEntryRequest{
+						Directory: entry.Dir,
+						Entry:     entry.Entry,
+					}); errEntry != nil {
+						err = errEntry
+					}
+					defer wg.Done()
+					<-waitChan
+				}(fullEntry)
+				if err != nil {
+					return err
+				}
 				fileCount++
 			}
-
 		}
-
 	})
 
 	if err == nil {