fix compilation

fix the spelling error

k8s/helm_charts2/Chart.yaml

@@ -1,5 +1,5 @@
 apiVersion: v1
 description: SeaweedFS
 name: seaweedfs
-appVersion: "3.14"
-version: "3.14"
+appVersion: "3.15"
+version: "3.15"

k8s/helm_charts2/templates/filer-statefulset.yaml

@@ -46,7 +46,7 @@ spec:
       imagePullSecrets:
         - name: {{ .Values.global.imagePullSecrets }}
       {{- end }}
-      serviceAccountName: seaweefds-rw-sa #hack for delete pod master after migration
+      serviceAccountName: seaweedfs-rw-sa #hack for delete pod master after migration
       terminationGracePeriodSeconds: 60
       {{- if .Values.filer.priorityClassName }}
       priorityClassName: {{ .Values.filer.priorityClassName | quote }}

k8s/helm_charts2/templates/service-account.yaml

@@ -3,7 +3,7 @@
 kind: ClusterRole
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
-  name: seaweefds-rw-cr
+  name: seaweedfs-rw-cr
 rules:
   - apiGroups: [""]
     resources: ["pods"]
@@ -12,18 +12,18 @@ rules:
 apiVersion: v1
 kind: ServiceAccount
 metadata:
-  name: seaweefds-rw-sa
+  name: seaweedfs-rw-sa
   namespace: {{ .Release.Namespace }}
 ---
 kind: ClusterRoleBinding
 apiVersion: rbac.authorization.k8s.io/v1
 metadata:
-  name: system:serviceaccount:seaweefds-rw-sa:default
+  name: system:serviceaccount:seaweedfs-rw-sa:default
 subjects:
 - kind: ServiceAccount
-  name: seaweefds-rw-sa
+  name: seaweedfs-rw-sa
   namespace: {{ .Release.Namespace }}
 roleRef:
   apiGroup: rbac.authorization.k8s.io
   kind: ClusterRole
-  name: seaweefds-rw-cr
+  name: seaweedfs-rw-cr

weed/command/scaffold/filer.toml

@@ -337,3 +337,11 @@ pdaddrs = "localhost:2379"
 deleterange_concurrency = 1
 # Enable 1PC
 enable_1pc = false
+# Set the CA certificate path
+ca_path=""
+# Set the certificate path
+cert_path=""
+# Set the private key path
+key_path=""
+# The name list used to verify the cn name
+verify_cn=""

weed/filer/read_write.go

@@ -2,9 +2,10 @@ package filer
 
 import (
 	"bytes"
+	"time"
+
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
-	"time"
 )
 
 func ReadEntry(masterClient *wdclient.MasterClient, filerClient filer_pb.SeaweedFilerClient, dir, name string, byteBuffer *bytes.Buffer) error {
@@ -60,7 +61,7 @@ func SaveInsideFiler(client filer_pb.SeaweedFilerClient, dir, name string, conte
 				},
 				Content: content,
 			},
-			SkipCheckParentDirectory: true,
+			SkipCheckParentDirectory: false,
 		})
 	} else if err == nil {
 		entry := resp.Entry

weed/filer/reader_cache.go

@@ -19,6 +19,7 @@ type ReaderCache struct {
 
 type SingleChunkCacher struct {
 	sync.RWMutex
+	cond          *sync.Cond
 	parent        *ReaderCache
 	chunkFileId   string
 	data          []byte
@@ -140,6 +141,7 @@ func newSingleChunkCacher(parent *ReaderCache, fileId string, cipherKey []byte,
 		chunkSize:   chunkSize,
 		shouldCache: shouldCache,
 	}
+	t.cond = sync.NewCond(t)
 	return t
 }
 
@@ -168,6 +170,7 @@ func (s *SingleChunkCacher) startCaching() {
 	if s.shouldCache {
 		s.parent.chunkCache.SetChunk(s.chunkFileId, s.data)
 	}
+	s.cond.Broadcast()
 
 	return
 }
@@ -183,6 +186,10 @@ func (s *SingleChunkCacher) readChunkAt(buf []byte, offset int64) (int, error) {
 	s.RLock()
 	defer s.RUnlock()
 
+	for s.completedTime.IsZero() {
+		s.cond.Wait()
+	}
+
 	if s.err != nil {
 		return 0, s.err
 	}

weed/filer/tikv/tikv_store.go

@@ -15,6 +15,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/util"
+	"github.com/tikv/client-go/v2/config"
 	"github.com/tikv/client-go/v2/txnkv"
 )
 
@@ -38,21 +39,25 @@ func (store *TikvStore) GetName() string {
 }
 
 func (store *TikvStore) Initialize(config util.Configuration, prefix string) error {
-	pdAddrs := []string{}
-	pdAddrsStr := config.GetString(prefix + "pdaddrs")
-	for _, item := range strings.Split(pdAddrsStr, ",") {
-		pdAddrs = append(pdAddrs, strings.TrimSpace(item))
-	}
+	ca := config.GetString(prefix + "ca_path")
+	cert := config.GetString(prefix + "cert_path")
+	key := config.GetString(prefix + "key_path")
+	verify_cn := strings.Split(config.GetString(prefix+"verify_cn"), ",")
+	pdAddrs := strings.Split(config.GetString(prefix+"pdaddrs"), ",")
+
 	drc := config.GetInt(prefix + "deleterange_concurrency")
 	if drc <= 0 {
 		drc = 1
 	}
 	store.onePC = config.GetBool(prefix + "enable_1pc")
 	store.deleteRangeConcurrency = drc
-	return store.initialize(pdAddrs)
+	return store.initialize(ca, cert, key, verify_cn, pdAddrs)
 }
 
-func (store *TikvStore) initialize(pdAddrs []string) error {
+func (store *TikvStore) initialize(ca, cert, key string, verify_cn, pdAddrs []string) error {
+	config.UpdateGlobal(func(conf *config.Config) {
+		conf.Security = config.NewSecurity(ca, cert, key, verify_cn)
+	})
 	client, err := txnkv.NewClient(pdAddrs)
 	store.client = client
 	return err

weed/iamapi/iamapi_management_handlers.go

@@ -412,14 +412,11 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	values := r.PostForm
-	var s3cfgLock sync.RWMutex
-	s3cfgLock.RLock()
 	s3cfg := &iam_pb.S3ApiConfiguration{}
 	if err := iama.s3ApiConfig.GetS3ApiConfiguration(s3cfg); err != nil {
 		s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 		return
 	}
-	s3cfgLock.RUnlock()
 
 	glog.V(4).Infof("DoActions: %+v", values)
 	var response interface{}
@@ -498,9 +495,7 @@ func (iama *IamApiServer) DoActions(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 	if changed {
-		s3cfgLock.Lock()
 		err := iama.s3ApiConfig.PutS3ApiConfiguration(s3cfg)
-		s3cfgLock.Unlock()
 		if err != nil {
 			writeIamErrorResponse(w, r, fmt.Errorf(iam.ErrCodeServiceFailureException), "", "", err)
 			return

weed/iamapi/iamapi_server.go

@@ -6,6 +6,8 @@ import (
 	"bytes"
 	"encoding/json"
 	"fmt"
+	"net/http"
+
 	"github.com/chrislusf/seaweedfs/weed/filer"
 	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
@@ -17,7 +19,6 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/wdclient"
 	"github.com/gorilla/mux"
 	"google.golang.org/grpc"
-	"net/http"
 )
 
 type IamS3ApiConfig interface {
@@ -117,10 +118,10 @@ func (iam IamS3ApiConfigure) GetPolicies(policies *Policies) (err error) {
 		}
 		return nil
 	})
-	if err != nil {
+	if err != nil && err != filer_pb.ErrNotFound {
 		return err
 	}
-	if buf.Len() == 0 {
+	if err == filer_pb.ErrNotFound || buf.Len() == 0 {
 		policies.Policies = make(map[string]PolicyDocument)
 		return nil
 	}

weed/mount/filehandle.go

@@ -98,16 +98,15 @@ func (fh *FileHandle) AddChunks(chunks []*filer_pb.FileChunk) {
 	fh.entryViewCache = nil
 }
 
-func (fh *FileHandle) SetReader(reader *filer.ChunkReadAt) {
+func (fh *FileHandle) CloseReader() {
 	if fh.reader != nil {
 		fh.reader.Close()
 	}
-	fh.reader = reader
 }
 
 func (fh *FileHandle) Release() {
 	fh.dirtyPages.Destroy()
-	fh.SetReader(nil)
+	fh.CloseReader()
 }
 
 func lessThan(a, b *filer_pb.FileChunk) bool {

weed/mount/filehandle_read.go

@@ -62,21 +62,19 @@ func (fh *FileHandle) readFromChunks(buff []byte, offset int64) (int64, error) {
 		if chunkResolveErr != nil {
 			return 0, fmt.Errorf("fail to resolve chunk manifest: %v", chunkResolveErr)
 		}
-		fh.SetReader(nil)
+		fh.CloseReader()
 	}
 
-	reader := fh.reader
-	if reader == nil {
+	if fh.reader == nil {
 		chunkViews := filer.ViewFromVisibleIntervals(fh.entryViewCache, 0, fileSize)
 		glog.V(4).Infof("file handle read %s [%d,%d) from %d views", fileFullPath, offset, offset+int64(len(buff)), len(chunkViews))
 		for _, chunkView := range chunkViews {
 			glog.V(4).Infof("  read %s [%d,%d) from chunk %+v", fileFullPath, chunkView.LogicOffset, chunkView.LogicOffset+int64(chunkView.Size), chunkView.FileId)
 		}
-		reader = filer.NewChunkReaderAtFromClient(fh.wfs.LookupFn(), chunkViews, fh.wfs.chunkCache, fileSize)
+		fh.reader = filer.NewChunkReaderAtFromClient(fh.wfs.LookupFn(), chunkViews, fh.wfs.chunkCache, fileSize)
 	}
-	fh.SetReader(reader)
 
-	totalRead, err := reader.ReadAt(buff, offset)
+	totalRead, err := fh.reader.ReadAt(buff, offset)
 
 	if err != nil && err != io.EOF {
 		glog.Errorf("file handle read %s: %v", fileFullPath, err)

weed/mount/inode_to_path.go

@@ -152,7 +152,7 @@ func (i *InodeToPath) RemovePath(path util.FullPath) {
 	}
 }
 
-func (i *InodeToPath) MovePath(sourcePath, targetPath util.FullPath) (replacedInode uint64) {
+func (i *InodeToPath) MovePath(sourcePath, targetPath util.FullPath) (sourceInode, targetInode uint64) {
 	i.Lock()
 	defer i.Unlock()
 	sourceInode, sourceFound := i.path2inode[sourcePath]
@@ -178,7 +178,7 @@ func (i *InodeToPath) MovePath(sourcePath, targetPath util.FullPath) (replacedIn
 	} else {
 		glog.Errorf("MovePath %s to %s: sourceInode %d not found", sourcePath, targetPath, sourceInode)
 	}
-	return targetInode
+	return
 }
 
 func (i *InodeToPath) Forget(inode, nlookup uint64, onForgetDir func(dir util.FullPath)) {

weed/mount/weedfs_rename.go

@@ -233,10 +233,17 @@ func (wfs *WFS) handleRenameResponse(ctx context.Context, resp *filer_pb.StreamR
 		oldPath := oldParent.Child(oldName)
 		newPath := newParent.Child(newName)
 
-		replacedInode := wfs.inodeToPath.MovePath(oldPath, newPath)
-		// invalidate attr and data
-		if replacedInode > 0 {
-			wfs.fuseServer.InodeNotify(replacedInode, 0, -1)
+		sourceInode, targetInode := wfs.inodeToPath.MovePath(oldPath, newPath)
+		if sourceInode != 0 {
+			if fh, foundFh := wfs.fhmap.inode2fh[sourceInode]; foundFh && fh.entry != nil {
+				fh.entry.Name = newName
+			}
+			// invalidate attr and data
+			wfs.fuseServer.InodeNotify(sourceInode, 0, -1)
+		}
+		if targetInode != 0 {
+			// invalidate attr and data
+			wfs.fuseServer.InodeNotify(targetInode, 0, -1)
 		}
 
 	} else if resp.EventNotification.OldEntry != nil {

weed/server/filer_grpc_server_sub_meta.go

@@ -263,6 +263,9 @@ func (fs *FilerServer) addClient(clientType string, clientAddress string, client
 	if clientId != 0 {
 		fs.knownListenersLock.Lock()
 		_, alreadyKnown = fs.knownListeners[clientId]
+		if !alreadyKnown {
+			fs.knownListeners[clientId] = struct{}{}
+		}
 		fs.knownListenersLock.Unlock()
 	}
 	return

weed/topology/topology_vacuum.go

@@ -185,10 +185,13 @@ func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float
 		for _, vl := range c.storageType2VolumeLayout.Items() {
 			if vl != nil {
 				volumeLayout := vl.(*VolumeLayout)
-				if volumeId > 0 && volumeLayout.Lookup(needle.VolumeId(volumeId)) == nil {
-					continue
+				if volumeId > 0 {
+					if volumeLayout.Lookup(needle.VolumeId(volumeId)) != nil {
+						t.vacuumOneVolumeLayout(grpcDialOption, volumeLayout, c, garbageThreshold, preallocate)
+					}
+				} else {
+					t.vacuumOneVolumeLayout(grpcDialOption, volumeLayout, c, garbageThreshold, preallocate)
 				}
-				t.vacuumOneVolumeLayout(grpcDialOption, volumeLayout, c, garbageThreshold, preallocate)
 			}
 		}
 	}

weed/util/constants.go

@@ -5,7 +5,7 @@ import (
 )
 
 var (
-	VERSION_NUMBER = fmt.Sprintf("%.02f", 3.14)
+	VERSION_NUMBER = fmt.Sprintf("%.02f", 3.15)
 	VERSION        = sizeLimit + " " + VERSION_NUMBER
 	COMMIT         = ""
 )