Merge remote-tracking branch 'origin/master' into feature/jobs_plugin

3 files changed, 60 insertions, 0 deletions

diff --git a/plugins/kv/drivers/boltdb/driver.go b/plugins/kv/drivers/boltdb/driver.go
index e5aac290..0f737fbd 100644
--- a/plugins/kv/drivers/boltdb/driver.go
+++ b/plugins/kv/drivers/boltdb/driver.go
@@ -18,6 +18,7 @@ import (
 )
 
 type Driver struct {
+	clearMu sync.RWMutex
 	// db instance
 	DB *bolt.DB
 	// name should be UTF-8
@@ -373,6 +374,35 @@ func (d *Driver) TTL(keys ...string) (map[string]string, error) {
 	return m, nil
 }
 
+func (d *Driver) Clear() error {
+	err := d.DB.Update(func(tx *bolt.Tx) error {
+		err := tx.DeleteBucket(d.bucket)
+		if err != nil {
+			d.log.Error("boltdb delete bucket", "error", err)
+			return err
+		}
+
+		_, err = tx.CreateBucket(d.bucket)
+		if err != nil {
+			d.log.Error("boltdb create bucket", "error", err)
+			return err
+		}
+
+		return nil
+	})
+
+	if err != nil {
+		d.log.Error("clear transaction failed", "error", err)
+		return err
+	}
+
+	d.clearMu.Lock()
+	d.gc = sync.Map{}
+	d.clearMu.Unlock()
+
+	return nil
+}
+
 // ========================= PRIVATE =================================
 
 func (d *Driver) startGCLoop() { //nolint:gocognit
@@ -382,6 +412,8 @@ func (d *Driver) startGCLoop() { //nolint:gocognit
 		for {
 			select {
 			case <-t.C:
+				d.clearMu.RLock()
+
 				// calculate current time before loop started to be fair
 				now := time.Now()
 				d.gc.Range(func(key, value interface{}) bool {
@@ -414,6 +446,8 @@ func (d *Driver) startGCLoop() { //nolint:gocognit
 					}
 					return true
 				})
+
+				d.clearMu.RUnlock()
 			case <-d.stop:
 				err := d.DB.Close()
 				if err != nil {
diff --git a/plugins/kv/drivers/memcached/driver.go b/plugins/kv/drivers/memcached/driver.go
index 520ec7d5..42e342ac 100644
--- a/plugins/kv/drivers/memcached/driver.go
+++ b/plugins/kv/drivers/memcached/driver.go
@@ -237,3 +237,13 @@ func (d *Driver) Delete(keys ...string) error {
 	}
 	return nil
 }
+
+func (d *Driver) Clear() error {
+	err := d.client.DeleteAll()
+	if err != nil {
+		d.log.Error("flush_all operation failed", "error", err)
+		return err
+	}
+
+	return nil
+}
diff --git a/plugins/kv/rpc.go b/plugins/kv/rpc.go
index b9b302fe..ad4aefa9 100644
--- a/plugins/kv/rpc.go
+++ b/plugins/kv/rpc.go
@@ -162,3 +162,19 @@ func (r *rpc) Delete(in *kvv1.Request, _ *kvv1.Response) error {
 
 	return errors.E(op, errors.Errorf("no such storage: %s", in.GetStorage()))
 }
+
+// Clear clean the storage
+func (r *rpc) Clear(in *kvv1.Request, _ *kvv1.Response) error {
+	const op = errors.Op("rcp_delete")
+
+	if st, exists := r.storages[in.GetStorage()]; exists {
+		err := st.Clear()
+		if err != nil {
+			return errors.E(op, err)
+		}
+
+		return nil
+	}
+
+	return errors.E(op, errors.Errorf("no such storage: %s", in.GetStorage()))
+}