add collection.delete

This commit is contained in:
Chris Lu 2019-05-30 09:47:54 -07:00
parent 1d111d6ce8
commit 40ca2f2903
5 changed files with 92 additions and 2 deletions

View file

@ -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
}

View file

@ -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
}

View file

@ -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
}

View file

@ -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)

View file

@ -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