protect against data loss by not destroying pundat when btrdb is unreachable

Author: gtfierro

Makefile

@@ -1,5 +1,5 @@
 APP?=pundat
-RELEASE?=0.3.3
+RELEASE?=0.3.4
 COMMIT?=$(shell git rev-parse --short HEAD)
 PROJECT?=github.com/gtfierro/pundat
 PERSISTDIR?=/etc/pundat

archiver/archiver.go

@@ -130,7 +130,7 @@ func (a *Archiver) Serve() {
 	}()
 	for _, namespace := range a.config.BOSSWAVE.ListenNS {
 		go a.vm.subscribeNamespace(ctx, namespace)
-		time.Sleep(10 * time.Second)
+		time.Sleep(2 * time.Second)
 	}
 
 	<-a.stop

archiver/btrdb.v4.go

@@ -13,7 +13,7 @@ import (
 	"gopkg.in/btrdb.v4"
 )
 
-var timeout = time.Second * 20
+var timeout = time.Second * 60
 
 var errStreamNotExist = errors.New("Stream does not exist")
 
@@ -36,7 +36,8 @@ func newBTrDBv4(c *btrdbv4Config) *btrdbv4Iface {
 	log.Noticef("Connecting to BtrDBv4 at addresses %v...", b.addresses)
 	conn, err := btrdb.Connect(context.Background(), b.addresses...)
 	if err != nil {
-		log.Fatalf("Could not connect to btrdbv4: %v", err)
+		log.Warningf("Could not connect to btrdbv4: %v", err)
+		return nil
 	}
 	b.conn = conn
 	log.Notice("Connected to BtrDB!")

archiver/stream.go

@@ -80,20 +80,20 @@ func (s *Stream) initialize(timeseriesStore TimeseriesStore, metadataStore Metad
 			ts := s.timeseries[msg.URI]
 			s.RUnlock()
 
-			ts.Lock()
+			commitme := ts.Copy()
 			// if no readings, then we give up
-			if len(ts.Records) == 0 {
-				ts.Unlock()
+			if len(commitme.Records) == 0 {
 				continue
 			}
 			// now we can assume the stream exists and can write to it
-			if err := timeseriesStore.AddReadings(ts); err != nil {
-				log.Error(errors.Wrap(err, "Could not write timeseries reading (probably deadline exceeded)"), len(ts.Records))
-				ts.Unlock()
+			if err := timeseriesStore.AddReadings(commitme); err != nil {
+				log.Error(errors.Wrap(err, "Could not write timeseries reading (probably deadline exceeded)"), len(commitme.Records))
 				continue
 			}
-			ts.Records = []*common.TimeseriesReading{}
+			ts.Lock()
+			ts.Records = ts.Records[len(commitme.Records):]
 			ts.Unlock()
+
 			s.Lock()
 			s.timeseries[msg.URI] = ts
 			s.Unlock()
@@ -219,16 +219,17 @@ func (s *Stream) start(timeseriesStore TimeseriesStore, metadataStore MetadataSt
 				ts.Records = append(ts.Records, &common.TimeseriesReading{Time: timestamp, Value: value_f64})
 				ts.Unlock()
 			}
-			ts.Lock()
-			if len(ts.Records) > commitCount {
-				// now we can assume the stream exists and can write to it
-				if err := timeseriesStore.AddReadings(ts); err != nil {
-					log.Error(errors.Wrapf(err, "Could not write timeseries reading %+v", ts))
-				} else {
-					ts.Records = []*common.TimeseriesReading{}
-				}
-			}
-			ts.Unlock()
+			//ts.Lock()
+			//if len(ts.Records) > commitCount {
+			//	// now we can assume the stream exists and can write to it
+			//	if err := timeseriesStore.AddReadings(ts); err != nil {
+			//		//TODO: when server is degraded, need to reconnect?
+			//		log.Error(errors.Wrapf(err, "Could not write timeseries reading %+v", ts))
+			//	} else {
+			//		ts.Records = []*common.TimeseriesReading{}
+			//	}
+			//}
+			//ts.Unlock()
 			s.Lock()
 			s.timeseries[msg.URI] = ts
 			s.Unlock()

common/timeseries.go

@@ -44,6 +44,19 @@ type Timeseries struct {
 	UUID       UUID
 }
 
+func (ts Timeseries) Copy() Timeseries {
+	ts.Lock()
+	newts := Timeseries{
+		Generation: ts.Generation,
+		SrcURI:     ts.SrcURI,
+		UUID:       ts.UUID,
+		Records:    make([]*TimeseriesReading, len(ts.Records)),
+	}
+	copy(newts.Records, ts.Records)
+	ts.Unlock()
+	return newts
+}
+
 // sort by timestamp
 func (ts Timeseries) Len() int {
 	return len(ts.Records)