From 40ca2f2903c2b95aa4dd2101598f0dbdd2024d07 Mon Sep 17 00:00:00 2001
From: Chris Lu <chris.lu@gmail.com>
Date: Thu, 30 May 2019 09:47:54 -0700
Subject: [PATCH] add collection.delete

---
 weed/shell/command_collection_delete.go  | 51 ++++++++++++++++++++++++
 weed/storage/disk_location.go            | 18 ++++++++-
 weed/storage/disk_location_ec.go         | 10 +++++
 weed/storage/erasure_coding/ec_shard.go  |  4 ++
 weed/storage/erasure_coding/ec_volume.go | 11 +++++
 5 files changed, 92 insertions(+), 2 deletions(-)
 create mode 100644 weed/shell/command_collection_delete.go

diff --git a/weed/shell/command_collection_delete.go b/weed/shell/command_collection_delete.go
new file mode 100644
index 000000000..e3356aba4
--- /dev/null
+++ b/weed/shell/command_collection_delete.go
@@ -0,0 +1,51 @@
+package shell
+
+import (
+	"context"
+	"fmt"
+	"github.com/chrislusf/seaweedfs/weed/pb/master_pb"
+	"io"
+)
+
+func init() {
+	commands = append(commands, &commandCollectionDelete{})
+}
+
+type commandCollectionDelete struct {
+}
+
+func (c *commandCollectionDelete) Name() string {
+	return "collection.delete"
+}
+
+func (c *commandCollectionDelete) Help() string {
+	return `delete specified collection
+
+	collection.delete <collection_name>
+
+`
+}
+
+func (c *commandCollectionDelete) Do(args []string, commandEnv *commandEnv, writer io.Writer) (err error) {
+
+	if len(args) == 0 {
+		return nil
+	}
+
+	collectionName := args[0]
+
+	ctx := context.Background()
+	err = commandEnv.masterClient.WithClient(ctx, func(client master_pb.SeaweedClient) error {
+		_, err = client.CollectionDelete(ctx, &master_pb.CollectionDeleteRequest{
+			Name: collectionName,
+		})
+		return err
+	})
+	if err != nil {
+		return
+	}
+
+	fmt.Fprintf(writer, "collection %s is deleted.\n", collectionName)
+
+	return nil
+}
diff --git a/weed/storage/disk_location.go b/weed/storage/disk_location.go
index 0aedb0f47..234d1a5f4 100644
--- a/weed/storage/disk_location.go
+++ b/weed/storage/disk_location.go
@@ -114,17 +114,31 @@ func (l *DiskLocation) loadExistingVolumes(needleMapKind NeedleMapType) {
 }
 
 func (l *DiskLocation) DeleteCollectionFromDiskLocation(collection string) (e error) {
-	l.Lock()
-	defer l.Unlock()
 
+	l.Lock()
 	for k, v := range l.volumes {
 		if v.Collection == collection {
 			e = l.deleteVolumeById(k)
 			if e != nil {
+				l.Unlock()
 				return
 			}
 		}
 	}
+	l.Unlock()
+
+	l.ecVolumesLock.Lock()
+	for k, v := range l.ecVolumes {
+		if v.Collection == collection {
+			e = l.deleteEcVolumeById(k)
+			if e != nil {
+				l.ecVolumesLock.Unlock()
+				return
+			}
+		}
+	}
+	l.ecVolumesLock.Unlock()
+
 	return
 }
 
diff --git a/weed/storage/disk_location_ec.go b/weed/storage/disk_location_ec.go
index bcc09a1c0..ff72baf33 100644
--- a/weed/storage/disk_location_ec.go
+++ b/weed/storage/disk_location_ec.go
@@ -148,3 +148,13 @@ func (l *DiskLocation) loadAllEcShards() (err error) {
 	}
 	return nil
 }
+
+func (l *DiskLocation) deleteEcVolumeById(vid needle.VolumeId) (e error) {
+	ecVolume, ok := l.ecVolumes[vid]
+	if !ok {
+		return
+	}
+	ecVolume.Destroy()
+	delete(l.ecVolumes, vid)
+	return
+}
diff --git a/weed/storage/erasure_coding/ec_shard.go b/weed/storage/erasure_coding/ec_shard.go
index 037c4ba48..687a582a7 100644
--- a/weed/storage/erasure_coding/ec_shard.go
+++ b/weed/storage/erasure_coding/ec_shard.go
@@ -64,6 +64,10 @@ func (shard *EcVolumeShard) Close() {
 	}
 }
 
+func (shard *EcVolumeShard) Destroy() {
+	os.Remove(shard.FileName() + ToExt(int(shard.ShardId)))
+}
+
 func (shard *EcVolumeShard) ReadAt(buf []byte, offset int64) (int, error) {
 
 	return shard.ecdFile.ReadAt(buf, offset)
diff --git a/weed/storage/erasure_coding/ec_volume.go b/weed/storage/erasure_coding/ec_volume.go
index 32f66260a..28dbd8c7f 100644
--- a/weed/storage/erasure_coding/ec_volume.go
+++ b/weed/storage/erasure_coding/ec_volume.go
@@ -94,6 +94,17 @@ func (ev *EcVolume) Close() {
 	}
 }
 
+func (ev *EcVolume) Destroy() {
+
+	ev.Close()
+
+	baseFileName := EcShardFileName(ev.Collection, ev.dir, int(ev.VolumeId))
+	for _, s := range ev.Shards {
+		s.Destroy()
+	}
+	os.Remove(baseFileName + ".ecx")
+}
+
 func (ev *EcVolume) ToVolumeEcShardInformationMessage() (messages []*master_pb.VolumeEcShardInformationMessage) {
 	prevVolumeId := needle.VolumeId(math.MaxUint32)
 	var m *master_pb.VolumeEcShardInformationMessage