volume server: added write blob grpc api similar to http

weed/pb/volume_server.proto

@@ -12,6 +12,9 @@ service VolumeServer {
     rpc BatchDelete (BatchDeleteRequest) returns (BatchDeleteResponse) {
     }
 
+    rpc WriteBlob (WriteBlobRequest) returns (WriteBlobResponse) {
+    }
+
     rpc VacuumVolumeCheck (VacuumVolumeCheckRequest) returns (VacuumVolumeCheckResponse) {
     }
     rpc VacuumVolumeCompact (VacuumVolumeCompactRequest) returns (stream VacuumVolumeCompactResponse) {
@@ -136,6 +139,16 @@ message DeleteResult {
 message Empty {
 }
 
+message WriteBlobRequest {
+    uint32 volume_id = 1;
+    string file_id = 2;
+    bytes data = 3;
+}
+
+message WriteBlobResponse {
+    uint32 size = 1;
+}
+
 message VacuumVolumeCheckRequest {
     uint32 volume_id = 1;
 }

weed/server/volume_grpc_write.go

@@ -0,0 +1,37 @@
+package weed_server
+
+import (
+	"context"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/topology"
+)
+
+// VolumeTierMoveDatToRemote copy dat file to a remote tier
+func (vs *VolumeServer) WriteBlob(ctx context.Context, req *volume_server_pb.WriteBlobRequest) (res *volume_server_pb.WriteBlobResponse, err error) {
+	res = &volume_server_pb.WriteBlobResponse{}
+
+	volumeId := needle.VolumeId(req.VolumeId)
+	needleId, cookie, _ := needle.ParseNeedleIdCookie(req.FileId)
+
+	n, contentMd5 := needle.CreateNeedleSimple(volumeId, needleId, cookie, req.Data)
+
+	params := topology.ReplicatedWriteParams{
+		VolumeId: volumeId,
+		Needle: n,
+		Jwt: "",
+		Replicate: false,
+		Fsync: false,
+		ContentMd5: contentMd5,
+	}
+	_, writeError := topology.ReplicatedWrite(vs.GetMaster, vs.grpcDialOption, vs.store, params)
+	if writeError != nil {
+		err = writeError
+		return nil, err
+	}
+
+	res.Size = uint32(n.Size)
+
+	return
+}

weed/server/volume_server_handlers_write.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/security"
 	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/topology"
@@ -45,7 +46,15 @@ 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, contentMd5)
+	params := topology.ReplicatedWriteParams{
+		VolumeId: volumeId,
+		Needle: reqNeedle,
+		Jwt: security.GetJwt(r),
+		Replicate: r.FormValue("type") == "replicate",
+		Fsync: r.FormValue("fsync") == "true",
+		ContentMd5: contentMd5,
+	}
+	isUnchanged, writeError := topology.ReplicatedWrite(vs.GetMaster, vs.grpcDialOption, vs.store, params)
 	if writeError != nil {
 		writeJsonError(w, r, http.StatusInternalServerError, writeError)
 		return
@@ -131,7 +140,13 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
 		}
 	}
 
-	_, err := topology.ReplicatedDelete(vs.GetMaster, vs.grpcDialOption, vs.store, volumeId, n, r)
+	params := topology.ReplicatedDeleteParams{
+		VolumeId: volumeId,
+		Needle: n,
+		Jwt: security.GetJwt(r),
+		Replicate: r.FormValue("type") == "replicate",
+	}
+	_, err := topology.ReplicatedDelete(vs.GetMaster, vs.grpcDialOption, vs.store, params)
 
 	writeDeleteResult(err, count, w, r)
 

weed/storage/needle/needle.go

@@ -2,6 +2,8 @@ package needle
 
 import (
 	"bytes"
+	"crypto/md5"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"net/http"
@@ -11,6 +13,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/images"
 	. "github.com/seaweedfs/seaweedfs/weed/storage/types"
+
 )
 
 const (
@@ -49,6 +52,29 @@ func (n *Needle) String() (str string) {
 	return
 }
 
+func CreateNeedleSimple(vid VolumeId, id NeedleId, cookie Cookie, data []byte) (n *Needle, contentMd5 string) {
+	n = new(Needle)
+
+	checksum := NewCRC(data)
+
+	n.Cookie = cookie
+	n.Id = id
+
+	n.Data = data
+	n.DataSize = uint32(len(data))
+	n.Checksum = checksum
+	n.Size = Size(len(data))
+
+	n.LastModified = uint64(time.Now().Unix())
+	n.SetHasLastModifiedDate()
+
+	h := md5.New()
+	h.Write(data)
+	contentMd5 = base64.StdEncoding.EncodeToString(h.Sum(nil))
+
+	return
+}
+
 func CreateNeedleFromRequest(r *http.Request, fixJpgOrientation bool, sizeLimit int64, bytesBuffer *bytes.Buffer) (n *Needle, originalSize int, contentMd5 string, e error) {
 	n = new(Needle)
 	pu, e := ParseUpload(r, sizeLimit, bytesBuffer)

weed/topology/store_replicate.go

@@ -5,7 +5,6 @@ import (
 	"errors"
 	"fmt"
 	"google.golang.org/grpc"
-	"net/http"
 	"net/url"
 	"strconv"
 	"strings"
@@ -23,16 +22,32 @@ import (
 	util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
 )
 
-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) {
+type ReplicatedWriteParams struct {
+	VolumeId needle.VolumeId
+	Needle        *needle.Needle
+	Jwt 	 security.EncodedJwt
+	Replicate bool
+	Fsync    bool
+	ContentMd5 string
+}
+
+type ReplicatedDeleteParams struct {
+	VolumeId needle.VolumeId
+	Needle        *needle.Needle
+	Jwt 	 security.EncodedJwt
+	Replicate bool
+}
+
+func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOption, s *storage.Store, params ReplicatedWriteParams) (isUnchanged bool, err error) {
 
 	//check JWT
-	jwt := security.GetJwt(r)
+	jwt := params.Jwt
 
 	// check whether this is a replicated write request
 	var remoteLocations []operation.Location
-	if r.FormValue("type") != "replicate" {
+	if !params.Replicate {
 		// this is the initial request
-		remoteLocations, err = GetWritableRemoteReplications(s, grpcDialOption, volumeId, masterFn)
+		remoteLocations, err = GetWritableRemoteReplications(s, grpcDialOption, params.VolumeId, masterFn)
 		if err != nil {
 			glog.V(0).Infoln(err)
 			return
@@ -40,19 +55,16 @@ func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOpt
 	}
 
 	// read fsync value
-	fsync := false
-	if r.FormValue("fsync") == "true" {
-		fsync = true
-	}
+	fsync := params.Fsync
 
-	if s.GetVolume(volumeId) != nil {
+	if s.GetVolume(params.VolumeId) != nil {
 		start := time.Now()
 
 		inFlightGauge := stats.VolumeServerInFlightRequestsGauge.WithLabelValues(stats.WriteToLocalDisk)
 		inFlightGauge.Inc()
 		defer inFlightGauge.Dec()
 
-		isUnchanged, err = s.WriteVolumeNeedle(volumeId, n, true, fsync)
+		isUnchanged, err = s.WriteVolumeNeedle(params.VolumeId, params.Needle, true, fsync)
 		stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToLocalDisk).Observe(time.Since(start).Seconds())
 		if err != nil {
 			stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToLocalDisk).Inc()
@@ -70,27 +82,29 @@ func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOpt
 		defer inFlightGauge.Dec()
 
 		err = DistributedOperation(remoteLocations, func(location operation.Location) error {
+			fileId := needle.NewFileIdFromNeedle(params.VolumeId, params.Needle)
+
 			u := url.URL{
 				Scheme: "http",
 				Host:   location.Url,
-				Path:   r.URL.Path,
+				Path:   fileId.String(),
 			}
 			q := url.Values{
 				"type": {"replicate"},
-				"ttl":  {n.Ttl.String()},
+				"ttl":  {params.Needle.Ttl.String()},
 			}
-			if n.LastModified > 0 {
-				q.Set("ts", strconv.FormatUint(n.LastModified, 10))
+			if params.Needle.LastModified > 0 {
+				q.Set("ts", strconv.FormatUint(params.Needle.LastModified, 10))
 			}
-			if n.IsChunkedManifest() {
+			if params.Needle.IsChunkedManifest() {
 				q.Set("cm", "true")
 			}
 			u.RawQuery = q.Encode()
 
 			pairMap := make(map[string]string)
-			if n.HasPairs() {
+			if params.Needle.HasPairs() {
 				tmpMap := make(map[string]string)
-				err := json.Unmarshal(n.Pairs, &tmpMap)
+				err := json.Unmarshal(params.Needle.Pairs, &tmpMap)
 				if err != nil {
 					stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorUnmarshalPairs).Inc()
 					glog.V(0).Infoln("Unmarshal pairs error:", err)
@@ -106,13 +120,13 @@ func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOpt
 			// TODO optimize here to compress data only once
 			uploadOption := &operation.UploadOption{
 				UploadUrl:         u.String(),
-				Filename:          string(n.Name),
+				Filename:          string(params.Needle.Name),
 				Cipher:            false,
-				IsInputCompressed: n.IsCompressed(),
-				MimeType:          string(n.Mime),
+				IsInputCompressed: params.Needle.IsCompressed(),
+				MimeType:          string(params.Needle.Mime),
 				PairMap:           pairMap,
 				Jwt:               jwt,
-				Md5:               contentMd5,
+				Md5:               params.ContentMd5,
 				BytesBuffer:       bytesBuffer,
 			}
 
@@ -121,7 +135,7 @@ func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOpt
 				glog.Errorf("replication-UploadData, err:%v, url:%s", err, u.String())
 				return err
 			}
-			_, err = uploader.UploadData(n.Data, uploadOption)
+			_, err = uploader.UploadData(params.Needle.Data, uploadOption)
 			if err != nil {
 				glog.Errorf("replication-UploadData, err:%v, url:%s", err, u.String())
 			}
@@ -130,7 +144,7 @@ func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOpt
 		stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToReplicas).Observe(time.Since(start).Seconds())
 		if err != nil {
 			stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToReplicas).Inc()
-			err = fmt.Errorf("failed to write to replicas for volume %d: %v", volumeId, err)
+			err = fmt.Errorf("failed to write to replicas for volume %d: %v", params.VolumeId, err)
 			glog.V(0).Infoln(err)
 			return false, err
 		}
@@ -138,21 +152,21 @@ func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOpt
 	return
 }
 
-func ReplicatedDelete(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOption, store *storage.Store, volumeId needle.VolumeId, n *needle.Needle, r *http.Request) (size types.Size, err error) {
+func ReplicatedDelete(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOption, store *storage.Store, params ReplicatedDeleteParams) (size types.Size, err error) {
 
 	//check JWT
-	jwt := security.GetJwt(r)
+	jwt := params.Jwt
 
 	var remoteLocations []operation.Location
-	if r.FormValue("type") != "replicate" {
-		remoteLocations, err = GetWritableRemoteReplications(store, grpcDialOption, volumeId, masterFn)
+	if params.Replicate {
+		remoteLocations, err = GetWritableRemoteReplications(store, grpcDialOption, params.VolumeId, masterFn)
 		if err != nil {
 			glog.V(0).Infoln(err)
 			return
 		}
 	}
 
-	size, err = store.DeleteVolumeNeedle(volumeId, n)
+	size, err = store.DeleteVolumeNeedle(params.VolumeId, params.Needle)
 	if err != nil {
 		glog.V(0).Infoln("delete error:", err)
 		return
@@ -160,7 +174,20 @@ func ReplicatedDelete(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOp
 
 	if len(remoteLocations) > 0 { //send to other replica locations
 		if err = DistributedOperation(remoteLocations, func(location operation.Location) error {
-			return util_http.Delete("http://"+location.Url+r.URL.Path+"?type=replicate", string(jwt))
+			fileId := needle.NewFileIdFromNeedle(params.VolumeId, params.Needle)
+
+			u := url.URL{
+				Scheme: "http",
+				Host:   location.Url,
+				Path:   fileId.String(),
+			}
+
+			q := url.Values{
+				"type": {"replicate"},
+			}
+			u.RawQuery = q.Encode()
+
+			return util_http.Delete(u.String(), string(jwt))
 		}); err != nil {
 			size = 0
 		}