Merge pull request #74 from DCSO/fixes

Address reconnects and protocol stability in backend communication

Author: 0mbi

db/db_remote.go

@@ -1,15 +1,15 @@
 // balboa
-// Copyright (c) 2019, DCSO GmbH
+// Copyright (c) 2019, 2025, DCSO GmbH
 
 package db
 
 import (
-	"gopkg.in/yaml.v2"
 	"net"
 
 	obs "github.com/DCSO/balboa/observation"
 
 	log "github.com/sirupsen/logrus"
+	"gopkg.in/yaml.v2"
 )
 
 type Backend struct {
@@ -109,8 +109,22 @@ func (db *RemoteBackend) ConsumeFeed(inChan chan obs.InputObservation) {
 					wanted := w.Len()
 					n, err := w.WriteTo(*conn)
 					if err != nil {
-						log.Warnf("sending observation failed: %s", err)
-						continue
+						log.Warnf("sending observation failed: %s, reconnecting", err)
+						(*conn).Close()
+						// try to reconnect
+						newConn, err := net.Dial("tcp", (*conn).RemoteAddr().String())
+						if err != nil {
+							log.Warnf("reconnecting to backend failed: %s", err)
+							continue
+						}
+						*conn = newConn
+						// and try again
+						n, err = w.WriteTo(*conn)
+						if err != nil {
+							log.Warnf("retry sending observation failed: %s", err)
+							(*conn).Close()
+							continue
+						}
 					}
 					if n != int64(wanted) {
 						log.Warnf("short write")

observation/input_observation.go

@@ -18,8 +18,8 @@ type InputObservation struct {
 	SensorID       string                   `codec:"I"`
 	TimestampEnd   time.Time                `codec:"L"`
 	TimestampStart time.Time                `codec:"F"`
-	Tags           map[string]struct{}      `codec:"G,omitempty"`
-	Selectors      map[interface{}]struct{} `codec:"S,omitempty"`
+	Tags           map[string]struct{}      `codec:"-,omitempty"`
+	Selectors      map[interface{}]struct{} `codec:"-,omitempty"`
 }
 
 // InChan is the global input channel delivering InputObservations from