Merge branch 'master' into support_ssd_volume

1 files changed, 33 insertions, 5 deletions

diff --git a/weed/shell/command_volume_fix_replication.go b/weed/shell/command_volume_fix_replication.go
index 9b9abd8eb..8ae8850f3 100644
--- a/weed/shell/command_volume_fix_replication.go
+++ b/weed/shell/command_volume_fix_replication.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"github.com/chrislusf/seaweedfs/weed/storage/needle"
 	"io"
+	"path/filepath"
 	"sort"
 
 	"github.com/chrislusf/seaweedfs/weed/operation"
@@ -19,6 +20,7 @@ func init() {
 }
 
 type commandVolumeFixReplication struct {
+	collectionPattern *string
 }
 
 func (c *commandVolumeFixReplication) Name() string {
@@ -33,12 +35,13 @@ func (c *commandVolumeFixReplication) Help() string {
 	This command also finds all under-replicated volumes, and finds volume servers with free slots.
 	If the free slots satisfy the replication requirement, the volume content is copied over and mounted.
 
-	volume.fix.replication -n # do not take action
-	volume.fix.replication    # actually deleting or copying the volume files and mount the volume
+	volume.fix.replication -n                             # do not take action
+	volume.fix.replication                                # actually deleting or copying the volume files and mount the volume
+	volume.fix.replication -collectionPattern=important*  # fix any collections with prefix "important"
 
 	Note:
-		* each time this will only add back one replica for one volume id. If there are multiple replicas
-		  are missing, e.g. multiple volume servers are new, you may need to run this multiple times.
+		* each time this will only add back one replica for each volume id that is under replicated.
+		  If there are multiple replicas are missing, e.g. replica count is > 2, you may need to run this multiple times.
 		* do not run this too quickly within seconds, since the new volume replica may take a few seconds 
 		  to register itself to the master.
 
@@ -52,6 +55,7 @@ func (c *commandVolumeFixReplication) Do(args []string, commandEnv *CommandEnv,
 	}
 
 	volFixReplicationCommand := flag.NewFlagSet(c.Name(), flag.ContinueOnError)
+	c.collectionPattern = volFixReplicationCommand.String("collectionPattern", "", "match with wildcard characters '*' and '?'")
 	skipChange := volFixReplicationCommand.Bool("n", false, "skip the changes")
 	if err = volFixReplicationCommand.Parse(args); err != nil {
 		return nil
@@ -127,6 +131,17 @@ func (c *commandVolumeFixReplication) fixOverReplicatedVolumes(commandEnv *Comma
 
 		replica := pickOneReplicaToDelete(replicas, replicaPlacement)
 
+		// check collection name pattern
+		if *c.collectionPattern != "" {
+			matched, err := filepath.Match(*c.collectionPattern, replica.info.Collection)
+			if err != nil {
+				return fmt.Errorf("match pattern %s with collection %s: %v", *c.collectionPattern, replica.info.Collection, err)
+			}
+			if !matched {
+				break
+			}
+		}
+
 		fmt.Fprintf(writer, "deleting volume %d from %s ...\n", replica.info.Id, replica.location.dataNode.Id)
 
 		if !takeAction {
@@ -147,9 +162,22 @@ func (c *commandVolumeFixReplication) fixUnderReplicatedVolumes(commandEnv *Comm
 		replica := pickOneReplicaToCopyFrom(replicas)
 		replicaPlacement, _ := super_block.NewReplicaPlacementFromByte(byte(replica.info.ReplicaPlacement))
 		foundNewLocation := false
+		hasSkippedCollection := false
 		for _, dst := range allLocations {
 			// check whether data nodes satisfy the constraints
 			if dst.dataNode.FreeVolumeCount > 0 && satisfyReplicaPlacement(replicaPlacement, replicas, dst) {
+				// check collection name pattern
+				if *c.collectionPattern != "" {
+					matched, err := filepath.Match(*c.collectionPattern, replica.info.Collection)
+					if err != nil {
+						return fmt.Errorf("match pattern %s with collection %s: %v", *c.collectionPattern, replica.info.Collection, err)
+					}
+					if !matched {
+						hasSkippedCollection = true
+						break
+					}
+				}
+
 				// ask the volume server to replicate the volume
 				foundNewLocation = true
 				fmt.Fprintf(writer, "replicating volume %d %s from %s to dataNode %s ...\n", replica.info.Id, replicaPlacement, replica.location.dataNode.Id, dst.dataNode.Id)
@@ -179,7 +207,7 @@ func (c *commandVolumeFixReplication) fixUnderReplicatedVolumes(commandEnv *Comm
 				break
 			}
 		}
-		if !foundNewLocation {
+		if !foundNewLocation && !hasSkippedCollection {
 			fmt.Fprintf(writer, "failed to place volume %d replica as %s, existing:%+v\n", replica.info.Id, replicaPlacement, len(replicas))
 		}