add md5 header when UploadData to replication in ReplicatedWrite (#3881)

weed/operation/upload_content.go

@@ -30,6 +30,7 @@ type UploadOption struct {
 	PairMap           map[string]string
 	Jwt               security.EncodedJwt
 	RetryForever      bool
+	Md5               string
 }
 
 type UploadResult struct {
@@ -254,6 +255,7 @@ func doUploadData(data []byte, option *UploadOption) (uploadResult *UploadResult
 			MimeType:          option.MimeType,
 			PairMap:           option.PairMap,
 			Jwt:               option.Jwt,
+			Md5:               option.Md5,
 		})
 		if uploadResult == nil {
 			return
@@ -284,6 +286,9 @@ func upload_content(fillBufferFunction func(w io.Writer) error, originalDataSize
 	if option.IsInputCompressed {
 		h.Set("Content-Encoding", "gzip")
 	}
+	if option.Md5 != "" {
+		h.Set("Content-MD5", option.Md5)
+	}
 
 	file_writer, cp_err := body_writer.CreatePart(h)
 	if cp_err != nil {

weed/server/volume_server_handlers_write.go

@@ -45,7 +45,7 @@ func (vs *VolumeServer) PostHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	ret := operation.UploadResult{}
-	isUnchanged, writeError := topology.ReplicatedWrite(vs.GetMaster, vs.grpcDialOption, vs.store, volumeId, reqNeedle, r)
+	isUnchanged, writeError := topology.ReplicatedWrite(vs.GetMaster, vs.grpcDialOption, vs.store, volumeId, reqNeedle, r, contentMd5)
 	if writeError != nil {
 		writeJsonError(w, r, http.StatusInternalServerError, writeError)
 	}

weed/topology/store_replicate.go

@@ -21,7 +21,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
-func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOption, s *storage.Store, volumeId needle.VolumeId, n *needle.Needle, r *http.Request) (isUnchanged bool, err error) {
+func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOption, s *storage.Store, volumeId needle.VolumeId, n *needle.Needle, r *http.Request, contentMd5 string) (isUnchanged bool, err error) {
 
 	//check JWT
 	jwt := security.GetJwt(r)
@@ -98,8 +98,13 @@ func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOpt
 				MimeType:          string(n.Mime),
 				PairMap:           pairMap,
 				Jwt:               jwt,
+				Md5:               contentMd5,
 			}
+
 			_, err := operation.UploadData(n.Data, uploadOption)
+			if err != nil {
+				glog.Errorf("replication-UploadData, err:%v, url:%s", err, u.String())
+			}
 			return err
 		})
 		stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToReplicas).Observe(time.Since(start).Seconds())