seaweedfs/weed/mount/filer_conf.go

package mount

import (
	"fmt"
	"path/filepath"
	"sync/atomic"
	"time"

	"github.com/seaweedfs/seaweedfs/weed/filer"
	"github.com/seaweedfs/seaweedfs/weed/glog"
	"github.com/seaweedfs/seaweedfs/weed/pb"
	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
	"github.com/seaweedfs/seaweedfs/weed/util"
)

func (wfs *WFS) subscribeFilerConfEvents() (func(), error) {
	now := time.Now()
	confDir := filer.DirectoryEtcSeaweedFS
	confName := filer.FilerConfName
	confFullName := filepath.Join(filer.DirectoryEtcSeaweedFS, filer.FilerConfName)

	// read current conf
	err := wfs.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
		content, err := filer.ReadInsideFiler(client, confDir, confName)
		if err != nil {
			return err
		}

		fc := filer.NewFilerConf()
		if len(content) > 0 {
			if err := fc.LoadFromBytes(content); err != nil {
				return fmt.Errorf("parse %s: %v", confFullName, err)
			}
		}

		wfs.FilerConf = fc

		return nil
	})
	if err != nil {
		return nil, err
	}

	processEventFn := func(resp *filer_pb.SubscribeMetadataResponse) error {
		message := resp.EventNotification
		if message.NewEntry == nil {
			return nil
		}

		dir := resp.Directory
		name := resp.EventNotification.NewEntry.Name

		if dir != confDir || name != confName {
			return nil
		}

		content := message.NewEntry.Content
		fc := filer.NewFilerConf()
		if len(content) > 0 {
			if err = fc.LoadFromBytes(content); err != nil {
				return fmt.Errorf("parse %s: %v", confFullName, err)
			}
		}

		wfs.FilerConf = fc

		return nil
	}

	metadataFollowOption := &pb.MetadataFollowOption{
		ClientName:             "fuse",
		ClientId:               wfs.signature,
		ClientEpoch:            1,
		SelfSignature:          0,
		PathPrefix:             confFullName,
		AdditionalPathPrefixes: nil,
		StartTsNs:              now.UnixNano(),
		StopTsNs:               0,
		EventErrorType:         pb.FatalOnError,
	}

	return func() {
		// sync new conf changes
		util.RetryUntil("followFilerConfChanges", func() error {
			metadataFollowOption.ClientEpoch++
			i := atomic.LoadInt32(&wfs.option.filerIndex)
			n := len(wfs.option.FilerAddresses)
			err = pb.FollowMetadata(wfs.option.FilerAddresses[i], wfs.option.GrpcDialOption, metadataFollowOption, processEventFn)
			if err == nil {
				atomic.StoreInt32(&wfs.option.filerIndex, i)
				return nil
			}

			i++
			if i >= int32(n) {
				i = 0
			}

			return err
		}, func(err error) bool {
			glog.V(0).Infof("fuse follow filer conf changes: %v", err)
			return true
		})
	}, nil
}

func (wfs *WFS) wormEnabledForEntry(path util.FullPath, entry *filer_pb.Entry) bool {
	if entry == nil || entry.Attributes == nil {
		return false
	}
	if wfs.FilerConf == nil {
		return false
	}

	rule := wfs.FilerConf.MatchStorageRule(string(path))
	if !rule.Worm {
		return false
	}

	return entry.Attributes.FileSize > 0 || entry.Attributes.Crtime != entry.Attributes.Mtime
}