chore: simultaneously write to the replica and disk

weed/filer/filer_notify_append.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/seaweedfs/seaweedfs/weed/operation"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/util"
 )
 
@@ -76,6 +77,7 @@ func (f *Filer) assignAndUpload(targetFile string, data []byte) (*operation.Assi
 		MimeType:          "",
 		PairMap:           nil,
 		Jwt:               assignResult.Auth,
+		HandlerCounter:    stats.FilerHandlerCounter,
 	}
 	uploadResult, err := operation.UploadData(data, uploadOption)
 	if err != nil {

weed/operation/upload_content.go

@@ -16,6 +16,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/prometheus/client_golang/prometheus"
 	"github.com/seaweedfs/seaweedfs/weed/glog"
 	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/security"
@@ -34,6 +35,7 @@ type UploadOption struct {
 	RetryForever      bool
 	Md5               string
 	BytesBuffer       *bytes.Buffer
+	HandlerCounter    *prometheus.CounterVec
 }
 
 type UploadResult struct {
@@ -170,6 +172,9 @@ func retriedUploadData(data []byte, option *UploadOption) (uploadResult *UploadR
 			return
 		}
 		glog.Warningf("uploading %d to %s: %v", i, option.UploadUrl, err)
+		if option.HandlerCounter != nil {
+			option.HandlerCounter.WithLabelValues(stats.RepeatErrorUploadData).Inc()
+		}
 	}
 	return
 }
@@ -263,7 +268,6 @@ func doUploadData(data []byte, option *UploadOption) (uploadResult *UploadResult
 			PairMap:           option.PairMap,
 			Jwt:               option.Jwt,
 			Md5:               option.Md5,
-			BytesBuffer:       option.BytesBuffer,
 		})
 		if uploadResult == nil {
 			return

weed/server/volume_grpc_remote.go

@@ -7,6 +7,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
 	"github.com/seaweedfs/seaweedfs/weed/remote_storage"
 	"github.com/seaweedfs/seaweedfs/weed/security"
+	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 	"sync"
@@ -69,6 +70,7 @@ func (vs *VolumeServer) FetchAndWriteNeedle(ctx context.Context, req *volume_ser
 					MimeType:          "",
 					PairMap:           nil,
 					Jwt:               security.EncodedJwt(req.Auth),
+					HandlerCounter:    stats.VolumeServerHandlerCounter,
 				}
 				if _, replicaWriteErr := operation.UploadData(data, uploadOption); replicaWriteErr != nil {
 					if err == nil {

weed/stats/metrics_names.go

@@ -15,6 +15,9 @@ const (
 
 	// master topology
 	ErrorWriteToLocalDisk      = "errorWriteToLocalDisk"
+	ErrorWriteVolumeNotFound   = "errorWriteVolumeNotFound"
+	ErrorWriteReplicasNotFound = "errorWriteReplicasNotFound"
+
 	ErrorUnmarshalPairs  = "errorUnmarshalPairs"
 	ErrorWriteToReplicas = "errorWriteToReplicas"
 
@@ -40,6 +43,7 @@ const (
 	ErrorReadInternal        = "read.internal.error"
 	ErrorWriteEntry          = "write.entry.failed"
 	RepeatErrorUploadContent = "upload.content.repeat.failed"
+	RepeatErrorUploadData    = "upload.data.repeat.failed"
 	ErrorChunkAssign         = "chunkAssign.failed"
 	ErrorReadCache           = "read.cache.failed"
 	ErrorReadStream          = "read.stream.failed"

weed/topology/store_replicate.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 	"net/http"
 	"net/url"
@@ -23,6 +24,11 @@ import (
 )
 
 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) {
+	if s.GetVolume(volumeId) == nil {
+		stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteVolumeNotFound).Inc()
+		err = fmt.Errorf("volume %d not found", volumeId)
+		return true, err
+	}
 
 	//check JWT
 	jwt := security.GetJwt(r)
@@ -34,7 +40,8 @@ func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOpt
 		remoteLocations, err = GetWritableRemoteReplications(s, grpcDialOption, volumeId, masterFn)
 		if err != nil {
 			glog.V(0).Infoln(err)
-			return
+			stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteReplicasNotFound).Inc()
+			return true, err
 		}
 	}
 
@@ -55,10 +62,12 @@ func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOpt
 			return
 		}
 	}
-
+	group := errgroup.Group{}
 	if len(remoteLocations) > 0 { //send to other replica locations
-		start := time.Now()
-		err = DistributedOperation(remoteLocations, func(location operation.Location) error {
+		group.Go(func() error {
+			startWriteToReplicas := time.Now()
+			defer stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToReplicas).Observe(time.Since(startWriteToReplicas).Seconds())
+			return DistributedOperation(remoteLocations, func(location operation.Location) error {
 				u := url.URL{
 					Scheme: "http",
 					Host:   location.Url,
@@ -79,15 +88,16 @@ func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOpt
 				pairMap := make(map[string]string)
 				if n.HasPairs() {
 					tmpMap := make(map[string]string)
-				err := json.Unmarshal(n.Pairs, &tmpMap)
-				if err != nil {
+					jsonErr := json.Unmarshal(n.Pairs, &tmpMap)
+					if jsonErr != nil {
 						stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorUnmarshalPairs).Inc()
-					glog.V(0).Infoln("Unmarshal pairs error:", err)
+						glog.V(0).Infoln("Unmarshal pairs error:", jsonErr)
 					}
 					for k, v := range tmpMap {
 						pairMap[needle.PairNamePrefix+k] = v
 					}
 				}
+
 				bytesBuffer := buffer_pool.SyncPoolGetBuffer()
 				defer buffer_pool.SyncPoolPutBuffer(bytesBuffer)
 
@@ -103,21 +113,33 @@ func ReplicatedWrite(masterFn operation.GetMasterFn, grpcDialOption grpc.DialOpt
 					Jwt:               jwt,
 					Md5:               contentMd5,
 					BytesBuffer:       bytesBuffer,
+					HandlerCounter:    stats.VolumeServerHandlerCounter,
 				}
 
-			_, err := operation.UploadData(n.Data, uploadOption)
-			if err != nil {
-				glog.Errorf("replication-UploadData, err:%v, url:%s", err, u.String())
+				_, errUload := operation.UploadData(n.Data, uploadOption)
+				if errUload != nil {
+					glog.Errorf("replication-UploadData, err:%v, url:%s", errUload, u.String())
 				}
-			return err
+				return errUload
+			})
 		})
-		stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToReplicas).Observe(time.Since(start).Seconds())
+	}
+
+	start := time.Now()
+	isUnchanged, err = s.WriteVolumeNeedle(volumeId, n, true, fsync)
+	stats.VolumeServerRequestHistogram.WithLabelValues(stats.WriteToLocalDisk).Observe(time.Since(start).Seconds())
 	if err != nil {
+		stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToLocalDisk).Inc()
+		err = fmt.Errorf("failed to write to local disk: %v", err)
+		glog.Errorln(err)
+		return true, err
+	}
+
+	if err = group.Wait(); err != nil {
 		stats.VolumeServerHandlerCounter.WithLabelValues(stats.ErrorWriteToReplicas).Inc()
 		err = fmt.Errorf("failed to write to replicas for volume %d: %v", volumeId, err)
-			glog.V(0).Infoln(err)
-			return false, err
-		}
+		glog.Errorln(err)
+		return
 	}
 	return
 }