Added UserAllocations support for add

add_test.go

@@ -58,6 +58,53 @@ func TestAdd(t *testing.T) {
 	})
 }
 
+func TestAddWithUserAllocations(t *testing.T) {
+	ctx := context.Background()
+	clusters, mock := createClusters(t)
+	defer shutdownClusters(t, clusters, mock)
+	sth := test.NewShardingTestHelper()
+	defer sth.Clean(t)
+
+	waitForLeaderAndMetrics(t, clusters)
+
+	t.Run("local", func(t *testing.T) {
+		params := api.DefaultAddParams()
+		params.ReplicationFactorMin = 2
+		params.ReplicationFactorMax = 2
+		params.UserAllocations = []peer.ID{clusters[0].id, clusters[1].id}
+		params.Shard = false
+		params.Name = "testlocal"
+		mfr, closer := sth.GetTreeMultiReader(t)
+		defer closer.Close()
+		r := multipart.NewReader(mfr, mfr.Boundary())
+		ci, err := clusters[0].AddFile(r, params)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		pinDelay()
+
+		f := func(t *testing.T, c *Cluster) {
+			if c == clusters[0] || c == clusters[1] {
+				pin := c.StatusLocal(ctx, ci)
+				if pin.Error != "" {
+					t.Error(pin.Error)
+				}
+				if pin.Status != api.TrackerStatusPinned {
+					t.Error("item should be pinned and is", pin.Status)
+				}
+			} else {
+				pin := c.StatusLocal(ctx, ci)
+				if pin.Status != api.TrackerStatusRemote {
+					t.Error("expected tracker status remote")
+				}
+			}
+		}
+
+		runF(t, clusters, f)
+	})
+}
+
 func TestAddPeerDown(t *testing.T) {
 	ctx := context.Background()
 	clusters, mock := createClusters(t)

api/add.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/url"
 	"strconv"
+	"strings"
 
 	cid "github.com/ipfs/go-cid"
 )
@@ -101,9 +102,14 @@ func AddParamsFromQuery(query url.Values) (*AddParams, error) {
 	params.Layout = layout
 
 	chunker := query.Get("chunker")
-	params.Chunker = chunker
+	if chunker != "" {
+		params.Chunker = chunker
+	}
+
 	name := query.Get("name")
-	params.Name = name
+	if name != "" {
+		params.Name = name
+	}
 
 	hashF := query.Get("hash")
 	if hashF != "" {
@@ -146,6 +152,11 @@ func AddParamsFromQuery(query url.Values) (*AddParams, error) {
 		return nil, err
 	}
 
+	allocs := query.Get("user-allocations")
+	if allocs != "" {
+		params.UserAllocations = StringsToPeers(strings.Split(allocs, ","))
+	}
+
 	err = parseIntParam(query, "cid-version", &params.CidVersion)
 	if err != nil {
 		return nil, err
@@ -178,6 +189,7 @@ func (p *AddParams) ToQueryString() string {
 	query.Set("replication-min", fmt.Sprintf("%d", p.ReplicationFactorMin))
 	query.Set("replication-max", fmt.Sprintf("%d", p.ReplicationFactorMax))
 	query.Set("name", p.Name)
+	query.Set("user-allocations", strings.Join(PeersToStrings(p.UserAllocations), ","))
 	query.Set("shard", fmt.Sprintf("%t", p.Shard))
 	query.Set("shard-size", fmt.Sprintf("%d", p.ShardSize))
 	query.Set("recursive", fmt.Sprintf("%t", p.Recursive))

cmd/ipfs-cluster-ctl/main.go

@@ -349,6 +349,10 @@ cluster "pin add".
 					Value: defaultAddParams.ReplicationFactorMax,
 					Usage: "Sets the maximum replication factor for pinning this file",
 				},
+				cli.StringFlag{
+					Name:  "allocations, allocs",
+					Usage: "Optional comma-separated list of peer IDs",
+				},
 				cli.BoolFlag{
 					Name:  "nocopy",
 					Usage: "Add the URL using filestore. Implies raw-leaves. (experimental)",
@@ -397,6 +401,9 @@ cluster "pin add".
 				p.ReplicationFactorMin = c.Int("replication-min")
 				p.ReplicationFactorMax = c.Int("replication-max")
 				p.Name = name
+				if c.String("allocations") != "" {
+					p.UserAllocations = api.StringsToPeers(strings.Split(c.String("allocations"), ","))
+				}
 				//p.Shard = shard
 				//p.ShardSize = c.Uint64("shard-size")
 				p.Shard = false