shell: support configure volume ttl

Signed-off-by: Lei Liu <liul.stone@gmail.com>

unmaintained/see_dat/see_dat_gzip.go

@@ -9,6 +9,7 @@ import (
 	"io/ioutil"
 	"net/http"
 	"time"
+
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/storage"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"

unmaintained/see_log_entry/see_log_entry.go

@@ -15,7 +15,7 @@ import (
 )
 
 var (
-	logdataFile = flag.String("logdata", "", "log data file saved under "+ filer2.SystemLogDir)
+	logdataFile = flag.String("logdata", "", "log data file saved under "+filer2.SystemLogDir)
 )
 
 func main() {

weed/command/master.go

@@ -26,13 +26,13 @@ var (
 )
 
 type MasterOptions struct {
-	port               *int
-	ip                 *string
-	ipBind             *string
-	metaFolder         *string
-	peers              *string
-	volumeSizeLimitMB  *uint
-	volumePreallocate  *bool
+	port              *int
+	ip                *string
+	ipBind            *string
+	metaFolder        *string
+	peers             *string
+	volumeSizeLimitMB *uint
+	volumePreallocate *bool
 	// pulseSeconds       *int
 	defaultReplication *string
 	garbageThreshold   *float64
@@ -173,11 +173,11 @@ func checkPeers(masterIp string, masterPort int, peers string) (masterAddress st
 
 func (m *MasterOptions) toMasterOption(whiteList []string) *weed_server.MasterOption {
 	return &weed_server.MasterOption{
-		Host:                    *m.ip,
-		Port:                    *m.port,
-		MetaFolder:              *m.metaFolder,
-		VolumeSizeLimitMB:       *m.volumeSizeLimitMB,
-		VolumePreallocate:       *m.volumePreallocate,
+		Host:              *m.ip,
+		Port:              *m.port,
+		MetaFolder:        *m.metaFolder,
+		VolumeSizeLimitMB: *m.volumeSizeLimitMB,
+		VolumePreallocate: *m.volumePreallocate,
 		// PulseSeconds:            *m.pulseSeconds,
 		DefaultReplicaPlacement: *m.defaultReplication,
 		GarbageThreshold:        *m.garbageThreshold,

weed/filer2/meta_aggregator.go

@@ -98,7 +98,7 @@ func (ma *MetaAggregator) subscribeToOneFiler(f *Filer, self string, filer strin
 	for {
 		err := pb.WithFilerClient(filer, ma.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
 			stream, err := client.SubscribeLocalMetadata(context.Background(), &filer_pb.SubscribeMetadataRequest{
-				ClientName: "filer:"+self,
+				ClientName: "filer:" + self,
 				PathPrefix: "/",
 				SinceNs:    lastTsNs,
 			})

weed/filesys/meta_cache/meta_cache.go

@@ -60,7 +60,7 @@ func (mc *MetaCache) AtomicUpdateEntry(ctx context.Context, oldPath util.FullPat
 				return err
 			}
 		}
-	}else{
+	} else {
 		// println("unknown old directory:", oldDir)
 	}
 

weed/filesys/unimplemented.go

@@ -10,11 +10,13 @@ import (
 // https://github.com/bazil/fuse/issues/130
 
 var _ = fs.NodeAccesser(&Dir{})
+
 func (dir *Dir) Access(ctx context.Context, req *fuse.AccessRequest) error {
 	return fuse.ENOSYS
 }
 
 var _ = fs.NodeAccesser(&File{})
+
 func (file *File) Access(ctx context.Context, req *fuse.AccessRequest) error {
 	return fuse.ENOSYS
 }

weed/operation/needle_parse_test.go

@@ -101,7 +101,6 @@ func TestCreateNeedleFromRequest(t *testing.T) {
 
 }
 
-
 var textContent = `Redistribution and use in source and binary forms, with or without
 modification, are permitted provided that the following conditions are
 met:
@@ -127,4 +126,4 @@ DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-`
+`

weed/pb/filer_pb/filer.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.24.0
-// 	protoc        v3.12.3
+// 	protoc-gen-go v1.25.0-devel
+// 	protoc        v3.12.0
 // source: filer.proto
 
 package filer_pb

weed/pb/iam_pb/iam.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.24.0
-// 	protoc        v3.12.3
+// 	protoc-gen-go v1.25.0-devel
+// 	protoc        v3.12.0
 // source: iam.proto
 
 package iam_pb

weed/pb/master_pb/master.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.24.0
-// 	protoc        v3.12.3
+// 	protoc-gen-go v1.25.0-devel
+// 	protoc        v3.12.0
 // source: master.proto
 
 package master_pb

weed/pb/messaging_pb/messaging.pb.go

@@ -1,7 +1,7 @@
 // Code generated by protoc-gen-go. DO NOT EDIT.
 // versions:
-// 	protoc-gen-go v1.24.0
-// 	protoc        v3.12.3
+// 	protoc-gen-go v1.25.0-devel
+// 	protoc        v3.12.0
 // source: messaging.proto
 
 package messaging_pb

weed/pb/volume_server.proto

@@ -201,6 +201,7 @@ message VolumeMarkReadonlyResponse {
 message VolumeConfigureRequest {
     uint32 volume_id = 1;
     string replication = 2;
+    string ttl = 3;
 }
 message VolumeConfigureResponse {
     string error = 1;
@@ -375,6 +376,7 @@ message VolumeInfo {
     repeated RemoteFile files = 1;
     uint32 version = 2;
     string replication = 3;
+    string ttl = 4;
 }
 
 message VolumeTierMoveDatToRemoteRequest {

weed/server/master_server.go

@@ -32,11 +32,11 @@ const (
 )
 
 type MasterOption struct {
-	Host                    string
-	Port                    int
-	MetaFolder              string
-	VolumeSizeLimitMB       uint
-	VolumePreallocate       bool
+	Host              string
+	Port              int
+	MetaFolder        string
+	VolumeSizeLimitMB uint
+	VolumePreallocate bool
 	// PulseSeconds            int
 	DefaultReplicaPlacement string
 	GarbageThreshold        float64
@@ -66,7 +66,7 @@ type MasterServer struct {
 
 	MasterClient *wdclient.MasterClient
 
-	adminLocks          *AdminLocks
+	adminLocks *AdminLocks
 }
 
 func NewMasterServer(r *mux.Router, option *MasterOption, peers []string) *MasterServer {

weed/server/volume_grpc_admin.go

@@ -118,7 +118,7 @@ func (vs *VolumeServer) VolumeConfigure(ctx context.Context, req *volume_server_
 	}
 
 	// modify the volume info file
-	if err := vs.store.ConfigureVolume(needle.VolumeId(req.VolumeId), req.Replication); err != nil {
+	if err := vs.store.ConfigureVolume(needle.VolumeId(req.VolumeId), req.Replication, req.Ttl); err != nil {
 		glog.Errorf("volume configure %v: %v", req, err)
 		resp.Error = fmt.Sprintf("volume configure %v: %v", req, err)
 		return resp, nil

weed/shell/command_volume_configure_replication.go → weed/shell/command_volume_configure.go

@@ -15,34 +15,36 @@ import (
 )
 
 func init() {
-	Commands = append(Commands, &commandVolumeConfigureReplication{})
+	Commands = append(Commands, &commandVolumeConfigure{})
 }
 
-type commandVolumeConfigureReplication struct {
+type commandVolumeConfigure struct {
 }
 
-func (c *commandVolumeConfigureReplication) Name() string {
-	return "volume.configure.replication"
+func (c *commandVolumeConfigure) Name() string {
+	return "volume.configure"
 }
 
-func (c *commandVolumeConfigureReplication) Help() string {
-	return `change volume replication value
+func (c *commandVolumeConfigure) Help() string {
+	return `change volume replication or ttl value
 
-	This command changes a volume replication value. It should be followed by volume.fix.replication.
+	This command changes a volume replication or ttl value. It should be followed by volume.fix.replication.
 
 `
 }
 
-func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
+func (c *commandVolumeConfigure) Do(args []string, commandEnv *CommandEnv, writer io.Writer) (err error) {
 
 	if err = commandEnv.confirmIsLocked(); err != nil {
 		return
 	}
 
-	configureReplicationCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
-	volumeIdInt := configureReplicationCommand.Int("volumeId", 0, "the volume id")
-	replicationString := configureReplicationCommand.String("replication", "", "the intended replication value")
-	if err = configureReplicationCommand.Parse(args); err != nil {
+	configureVolumeCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	volumeIdInt := configureVolumeCommand.Int("volumeId", 0, "the volume id")
+	replicationString := configureVolumeCommand.String("replication", "", "the intended replication value")
+	ttlString := configureVolumeCommand.String("ttl", "", "the intended ttl value")
+
+	if err = configureVolumeCommand.Parse(args); err != nil {
 		return nil
 	}
 
@@ -50,12 +52,21 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
 		return fmt.Errorf("empty replication value")
 	}
 
+	if *ttlString == "" {
+		return fmt.Errorf("empty ttl value")
+	}
+
 	replicaPlacement, err := super_block.NewReplicaPlacementFromString(*replicationString)
 	if err != nil {
 		return fmt.Errorf("replication format: %v", err)
 	}
 	replicaPlacementInt32 := uint32(replicaPlacement.Byte())
 
+	ttl, err := needle.ReadTTL(*ttlString)
+	if err != nil {
+		return fmt.Errorf("wrong ttl format: %v", err)
+	}
+
 	var resp *master_pb.VolumeListResponse
 	err = commandEnv.MasterClient.WithClient(func(client master_pb.SeaweedClient) error {
 		resp, err = client.VolumeList(context.Background(), &master_pb.VolumeListRequest{})
@@ -72,7 +83,7 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
 	eachDataNode(resp.TopologyInfo, func(dc string, rack RackId, dn *master_pb.DataNodeInfo) {
 		loc := newLocation(dc, string(rack), dn)
 		for _, v := range dn.VolumeInfos {
-			if v.Id == uint32(vid) && v.ReplicaPlacement != replicaPlacementInt32 {
+			if v.Id == uint32(vid) && (v.ReplicaPlacement != replicaPlacementInt32 || v.Ttl != ttl.ToUint32()) {
 				allLocations = append(allLocations, loc)
 				continue
 			}
@@ -88,6 +99,7 @@ func (c *commandVolumeConfigureReplication) Do(args []string, commandEnv *Comman
 			resp, configureErr := volumeServerClient.VolumeConfigure(context.Background(), &volume_server_pb.VolumeConfigureRequest{
 				VolumeId:    uint32(vid),
 				Replication: replicaPlacement.String(),
+				Ttl:         ttl.String(),
 			})
 			if configureErr != nil {
 				return configureErr

weed/storage/erasure_coding/ec_decoder.go

@@ -119,7 +119,7 @@ func iterateEcxFile(baseFileName string, processNeedleFn func(key types.NeedleId
 }
 
 func iterateEcjFile(baseFileName string, processNeedleFn func(key types.NeedleId) error) error {
-	if !util.FileExists(baseFileName+".ecj") {
+	if !util.FileExists(baseFileName + ".ecj") {
 		return nil
 	}
 	ecjFile, openErr := os.OpenFile(baseFileName+".ecj", os.O_RDONLY, 0644)

weed/storage/erasure_coding/ec_volume_test.go

@@ -24,8 +24,8 @@ func TestPositioning(t *testing.T) {
 
 	tests := []struct {
 		needleId string
-		offset int64
-		size int
+		offset   int64
+		size     int
 	}{
 		{needleId: "0f0edb92", offset: 31300679656, size: 1167},
 		{needleId: "0ef7d7f8", offset: 11513014944, size: 66044},

weed/storage/store.go

@@ -372,7 +372,7 @@ func (s *Store) DeleteVolume(i needle.VolumeId) error {
 	return fmt.Errorf("volume %d not found on disk", i)
 }
 
-func (s *Store) ConfigureVolume(i needle.VolumeId, replication string) error {
+func (s *Store) ConfigureVolume(i needle.VolumeId, replication, ttl string) error {
 
 	for _, location := range s.Locations {
 		fileInfo, found := location.LocateVolume(i)
@@ -387,6 +387,7 @@ func (s *Store) ConfigureVolume(i needle.VolumeId, replication string) error {
 			return fmt.Errorf("volume %d fail to load vif", i)
 		}
 		volumeInfo.Replication = replication
+		volumeInfo.Ttl = ttl
 		err = pb.SaveVolumeInfo(vifFile, volumeInfo)
 		if err != nil {
 			return fmt.Errorf("volume %d fail to save vif", i)

weed/util/bounded_tree/bounded_tree_test.go

@@ -9,9 +9,7 @@ import (
 	"github.com/chrislusf/seaweedfs/weed/util"
 )
 
-
 var (
-
 	visitFn = func(path util.FullPath) (childDirectories []string, err error) {
 		fmt.Printf("  visit %v ...\n", path)
 		switch path {
@@ -37,14 +35,11 @@ var (
 		return nil, nil
 	}
 
-
 	printMap = func(m map[string]*Node) {
 		for k := range m {
 			println("  >", k)
 		}
 	}
-
-
 )
 
 func TestBoundedTree(t *testing.T) {

weed/util/compression.go

@@ -54,6 +54,7 @@ func ungzipData(input []byte) ([]byte, error) {
 }
 
 var decoder, _ = zstd.NewReader(nil)
+
 func unzstdData(input []byte) ([]byte, error) {
 	return decoder.DecodeAll(input, nil)
 }