Feature: add ability to delete apps #119

pkg/storage/cache/cache.go

@@ -76,6 +76,24 @@ func (cache *Cache) Flush() {
 	<-cache.cleanupDone
 }
 
+func (cache *Cache) Delete(key string) {
+	//cache.db.Lock()
+	//defer cache.db.Unlock()
+	cache.Flush()
+	lg := logrus.WithFields(logrus.Fields{
+		"prefix": cache.prefix,
+		"key":    key,
+	})
+	lg.Debug("deleting")
+	err := cache.db.Update(func(txn *badger.Txn) error {
+		return txn.Delete([]byte(cache.prefix + key))
+	})
+	if err != nil {
+		// TODO: handle
+		lg.Errorf("error happened in Delete: %v", err)
+	}
+}
+
 func (cache *Cache) Get(key string) interface{} {
 	lg := logrus.WithField("key", key)
 	if cache.lfu.UpperBound > 0 {

pkg/storage/storage.go

@@ -320,6 +320,58 @@ func (s *Storage) Get(gi *GetInput) (*GetOutput, error) {
 	}, nil
 }
 
+type DeleteInput struct {
+	StartTime time.Time
+	EndTime   time.Time
+	Key       *Key
+}
+
+func (s *Storage) Delete(di *DeleteInput) error {
+	s.closingMutex.Lock()
+	defer s.closingMutex.Unlock()
+
+	if s.closing {
+		return errClosing
+	}
+
+	logrus.WithFields(logrus.Fields{
+		"startTime": di.StartTime.String(),
+		"endTime":   di.EndTime.String(),
+		"key":       di.Key.Normalized(),
+	}).Info("storage.Delete")
+
+	dimensions := []*dimension.Dimension{}
+	for k, v := range di.Key.labels {
+		d := s.dimensions.Get(k + ":" + v).(*dimension.Dimension)
+		dimensions = append(dimensions, d)
+	}
+
+	segmentKeys := dimension.Intersection(dimensions...)
+
+	for _, sk := range segmentKeys {
+		// TODO: refactor, store `Key`s in dimensions
+		skk, _ := ParseKey(string(sk))
+		st := s.segments.Get(skk.SegmentKey()).(*segment.Segment)
+		if st == nil {
+			continue
+		}
+
+		st.Get(di.StartTime, di.EndTime, func(depth int, samples, writes uint64, t time.Time, r *big.Rat) {
+			k := skk.TreeKey(depth, t)
+			s.trees.Delete(k)
+			s.dicts.Delete(FromTreeToMainKey(k))
+		})
+
+		s.segments.Delete(skk.SegmentKey())
+	}
+
+	for k, v := range di.Key.labels {
+		s.dimensions.Delete(k + ":" + v)
+	}
+
+	return nil
+}
+
 func (s *Storage) Close() error {
 	s.closingMutex.Lock()
 	s.closing = true

pkg/storage/storage_test.go

@@ -30,6 +30,47 @@ var _ = Describe("storage package", func() {
 			Expect(err).ToNot(HaveOccurred())
 		})
 
+		Context("delete tests", func() {
+			Context("simple deletion", func() {
+				It("works correctly", func() {
+					tree := tree.New()
+					tree.Insert([]byte("a;b"), uint64(1))
+					tree.Insert([]byte("a;c"), uint64(2))
+					st := testing.SimpleTime(10)
+					et := testing.SimpleTime(19)
+					st2 := testing.SimpleTime(0)
+					et2 := testing.SimpleTime(30)
+					key, _ := ParseKey("foo")
+
+					err := s.Put(&PutInput{
+						StartTime:  st,
+						EndTime:    et,
+						Key:        key,
+						Val:        tree,
+						SpyName:    "testspy",
+						SampleRate: 100,
+					})
+					Expect(err).ToNot(HaveOccurred())
+
+					err = s.Delete(&DeleteInput{
+						StartTime: st,
+						EndTime:   et,
+						Key:       key,
+					})
+					Expect(err).ToNot(HaveOccurred())
+
+					gOut, err := s.Get(&GetInput{
+						StartTime: st2,
+						EndTime:   et2,
+						Key:       key,
+					})
+
+					Expect(err).ToNot(HaveOccurred())
+					Expect(gOut).To(BeNil())
+				})
+			})
+		})
+
 		Context("smoke tests", func() {
 			Context("simple 10 second write", func() {
 				It("works correctly", func() {