Fix chunk manifest replicate upload error

Escape the redirect url in volume read handler

2 files changed, 30 insertions, 3 deletions

diff --git a/go/topology/store_replicate.go b/go/topology/store_replicate.go
index a4be4726f..89fa1f379 100644
--- a/go/topology/store_replicate.go
+++ b/go/topology/store_replicate.go
@@ -5,6 +5,8 @@ import (
 	"net/http"
 	"strconv"
 
+	"net/url"
+
 	"github.com/chrislusf/seaweedfs/go/glog"
 	"github.com/chrislusf/seaweedfs/go/operation"
 	"github.com/chrislusf/seaweedfs/go/security"
@@ -33,9 +35,24 @@ func ReplicatedWrite(masterNode string, s *storage.Store,
 	}
 	if needToReplicate { //send to other replica locations
 		if r.FormValue("type") != "replicate" {
+
 			if !distributedOperation(masterNode, s, volumeId, func(location operation.Location) bool {
-				_, err := operation.Upload(
-					"http://"+location.Url+r.URL.Path+"?type=replicate&ts="+strconv.FormatUint(needle.LastModified, 10),
+				u := url.URL{
+					Scheme: "http",
+					Host:   location.Url,
+					Path:   r.URL.Path,
+				}
+				q := url.Values{
+					"type": {"replicate"},
+				}
+				if needle.LastModified > 0 {
+					q.Set("ts", strconv.FormatUint(needle.LastModified, 10))
+				}
+				if needle.IsChunkedManifest() {
+					q.Set("cm", "true")
+				}
+				u.RawQuery = q.Encode()
+				_, err := operation.Upload(u.String(),
 					string(needle.Name), bytes.NewReader(needle.Data), needle.IsGzipped(), string(needle.Mime),
 					jwt)
 				return err == nil
diff --git a/go/weed/weed_server/volume_server_handlers_read.go b/go/weed/weed_server/volume_server_handlers_read.go
index bfa6d6a4f..52b22426f 100644
--- a/go/weed/weed_server/volume_server_handlers_read.go
+++ b/go/weed/weed_server/volume_server_handlers_read.go
@@ -11,6 +11,8 @@ import (
 	"strings"
 	"time"
 
+	"net/url"
+
 	"github.com/chrislusf/seaweedfs/go/glog"
 	"github.com/chrislusf/seaweedfs/go/images"
 	"github.com/chrislusf/seaweedfs/go/operation"
@@ -46,7 +48,15 @@ func (vs *VolumeServer) GetOrHeadHandler(w http.ResponseWriter, r *http.Request)
 		lookupResult, err := operation.Lookup(vs.GetMasterNode(), volumeId.String())
 		glog.V(2).Infoln("volume", volumeId, "found on", lookupResult, "error", err)
 		if err == nil && len(lookupResult.Locations) > 0 {
-			http.Redirect(w, r, util.NormalizeUrl(lookupResult.Locations[0].PublicUrl)+r.URL.Path, http.StatusMovedPermanently)
+			u, _ := url.Parse(util.NormalizeUrl(lookupResult.Locations[0].PublicUrl))
+			u.Path = r.URL.Path
+			arg := url.Values{}
+			if c := r.FormValue("collection"); c != "" {
+				arg.Set("collection", c)
+			}
+			u.RawQuery = arg.Encode()
+			http.Redirect(w, r, u.String(), http.StatusMovedPermanently)
+
 		} else {
 			glog.V(2).Infoln("lookup error:", err, r.URL.Path)
 			w.WriteHeader(http.StatusNotFound)