ec deletion code complete, not tested yet

weed/server/volume_server_handlers_write.go

@@ -1,6 +1,7 @@
 package weed_server
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"net/http"
@@ -89,6 +90,14 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
 
 	cookie := n.Cookie
 
+	ecVolume, hasEcVolume := vs.store.FindEcVolume(volumeId)
+
+	if hasEcVolume {
+		count, err := vs.store.DeleteEcShardNeedle(context.Background(), ecVolume, n, cookie);
+		writeDeleteResult(err, count, w, r)
+		return
+	}
+
 	_, ok := vs.store.ReadVolumeNeedle(volumeId, n)
 	if ok != nil {
 		m := make(map[string]uint32)
@@ -129,6 +138,11 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
 
 	_, err := topology.ReplicatedDelete(vs.GetMaster(), vs.store, volumeId, n, r)
 
+	writeDeleteResult(err, count, w, r)
+
+}
+
+func writeDeleteResult(err error, count int64, w http.ResponseWriter, r *http.Request) {
 	if err == nil {
 		m := make(map[string]int64)
 		m["size"] = count
@@ -136,7 +150,6 @@ func (vs *VolumeServer) DeleteHandler(w http.ResponseWriter, r *http.Request) {
 	} else {
 		writeJsonError(w, r, http.StatusInternalServerError, fmt.Errorf("Deletion Failed: %v", err))
 	}
-
 }
 
 func setEtag(w http.ResponseWriter, etag string) {

weed/storage/store_ec_delete.go

@@ -0,0 +1,105 @@
+package storage
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/operation"
+	"github.com/chrislusf/seaweedfs/weed/pb/volume_server_pb"
+	"github.com/chrislusf/seaweedfs/weed/storage/erasure_coding"
+	"github.com/chrislusf/seaweedfs/weed/storage/needle"
+	"github.com/chrislusf/seaweedfs/weed/storage/types"
+)
+
+func (s *Store) DeleteEcShardNeedle(ctx context.Context, ecVolume *erasure_coding.EcVolume, n *needle.Needle, cookie types.Cookie) (int64, error) {
+
+	count, err := s.ReadEcShardNeedle(ctx, ecVolume.VolumeId, n)
+
+	if err != nil {
+		return 0, err
+	}
+
+	if cookie != n.Cookie {
+		return 0, fmt.Errorf("unexpected cookie %x", cookie)
+	}
+
+	if err = s.doDeleteNeedleFromAtLeastOneRemoteEcShards(ctx, ecVolume, n.Id); err != nil {
+		return 0, err
+	}
+
+	return int64(count), nil
+
+}
+
+func (s *Store) doDeleteNeedleFromAtLeastOneRemoteEcShards(ctx context.Context, ecVolume *erasure_coding.EcVolume, needleId types.NeedleId) error {
+
+	_, _, intervals, err := ecVolume.LocateEcShardNeedle(needleId, ecVolume.Version)
+
+	if len(intervals) == 0 {
+		return erasure_coding.NotFoundError
+	}
+
+	shardId, _ := intervals[0].ToShardIdAndOffset(erasure_coding.ErasureCodingLargeBlockSize, erasure_coding.ErasureCodingSmallBlockSize)
+
+	hasDeletionSuccess := false
+	err = s.doDeleteNeedleFromRemoteEcShardServers(ctx, shardId, ecVolume, needleId)
+	if err == nil {
+		hasDeletionSuccess = true
+	}
+
+	for shardId = erasure_coding.DataShardsCount; shardId < erasure_coding.TotalShardsCount; shardId++ {
+		if parityDeletionError := s.doDeleteNeedleFromRemoteEcShardServers(ctx, shardId, ecVolume, needleId); parityDeletionError == nil {
+			hasDeletionSuccess = true
+		}
+	}
+
+	if hasDeletionSuccess {
+		return nil
+	}
+
+	return err
+
+}
+
+func (s *Store) doDeleteNeedleFromRemoteEcShardServers(ctx context.Context, shardId erasure_coding.ShardId, ecVolume *erasure_coding.EcVolume, needleId types.NeedleId) error {
+
+	ecVolume.ShardLocationsLock.RLock()
+	sourceDataNodes, hasShardLocations := ecVolume.ShardLocations[shardId]
+	ecVolume.ShardLocationsLock.RUnlock()
+
+	if !hasShardLocations {
+		return fmt.Errorf("ec shard %d.%d not located", ecVolume.VolumeId, shardId)
+	}
+
+	for _, sourceDataNode := range sourceDataNodes {
+		glog.V(4).Infof("delete from remote ec shard %d.%d from %s", ecVolume.VolumeId, shardId, sourceDataNode)
+		err := s.doDeleteNeedleFromRemoteEcShard(ctx, sourceDataNode, ecVolume.VolumeId, ecVolume.Collection, ecVolume.Version, needleId)
+		if err != nil {
+			return err
+		}
+		glog.V(1).Infof("delete from remote ec shard %d.%d from %s: %v", ecVolume.VolumeId, shardId, sourceDataNode, err)
+	}
+
+	return nil
+
+}
+
+func (s *Store) doDeleteNeedleFromRemoteEcShard(ctx context.Context, sourceDataNode string, vid needle.VolumeId, collection string, version needle.Version, needleId types.NeedleId) error {
+
+	return operation.WithVolumeServerClient(sourceDataNode, s.grpcDialOption, func(client volume_server_pb.VolumeServerClient) error {
+
+		// copy data slice
+		_, err := client.VolumeEcBlobDelete(ctx, &volume_server_pb.VolumeEcBlobDeleteRequest{
+			VolumeId:   uint32(vid),
+			Collection: collection,
+			FileKey:    uint64(needleId),
+			Version:    uint32(version),
+		})
+		if err != nil {
+			return fmt.Errorf("failed to delete from ec shard %d on %s: %v", vid, sourceDataNode, err)
+		}
+		return nil
+	})
+
+}