[TASK] add database respondd for forwarding (#120)

config_example.toml

@@ -163,6 +163,16 @@ address  = "localhost:2003"
 # probably wont care much about "polluting" the namespace.
 prefix   = "freifunk"
 
+# respondd (yanic)
+# forward collected respondd package to a address
+# (e.g. to another respondd collector like a central yanic instance or hopglass)
+[[database.connection.respondd]]
+enable   = false
+# type of network to create a connection
+type     = "udp6"
+# destination address to connect/send respondd package
+address  = "stats.bremen.freifunk.net:11001"
+
 # Logging
 [[database.connection.logging]]
 enable   = false

database/all/main.go

@@ -4,4 +4,5 @@ import (
 	_ "github.com/FreifunkBremen/yanic/database/graphite"
 	_ "github.com/FreifunkBremen/yanic/database/influxdb"
 	_ "github.com/FreifunkBremen/yanic/database/logging"
+	_ "github.com/FreifunkBremen/yanic/database/respondd"
 )

database/respondd/main.go

@@ -0,0 +1,96 @@
+package respondd
+
+/**
+ * This database type is for injecting into another yanic instance.
+ */
+import (
+	"bufio"
+	"compress/flate"
+	"encoding/json"
+	"log"
+	"net"
+	"time"
+
+	"github.com/FreifunkBremen/yanic/data"
+	"github.com/FreifunkBremen/yanic/database"
+	"github.com/FreifunkBremen/yanic/runtime"
+)
+
+type Connection struct {
+	database.Connection
+	config Config
+	conn   net.Conn
+}
+
+type Config map[string]interface{}
+
+func (c Config) Type() string {
+	return c["type"].(string)
+}
+
+func (c Config) Address() string {
+	return c["address"].(string)
+}
+
+func init() {
+	database.RegisterAdapter("respondd", Connect)
+}
+
+func Connect(configuration map[string]interface{}) (database.Connection, error) {
+	var config Config
+	config = configuration
+
+	conn, err := net.Dial(config.Type(), config.Address())
+	if err != nil {
+		return nil, err
+	}
+
+	return &Connection{conn: conn, config: config}, nil
+}
+
+func (conn *Connection) InsertNode(node *runtime.Node) {
+	res := &data.ResponseData{
+		NodeInfo:   node.Nodeinfo,
+		Statistics: node.Statistics,
+		Neighbours: node.Neighbours,
+	}
+
+	writer := bufio.NewWriterSize(conn.conn, 8192)
+
+	flater, err := flate.NewWriter(writer, flate.BestCompression)
+	if err != nil {
+		log.Printf("[database-yanic] could not create flater: %s", err)
+		return
+	}
+	defer flater.Close()
+	err = json.NewEncoder(flater).Encode(res)
+	if err != nil {
+		nodeid := "unknown"
+		if node.Nodeinfo != nil && node.Nodeinfo.NodeID != "" {
+			nodeid = node.Nodeinfo.NodeID
+		}
+		log.Printf("[database-yanic] could not encode %s node: %s", nodeid, err)
+		return
+	}
+	err = flater.Flush()
+	if err != nil {
+		log.Printf("[database-yanic] could not compress: %s", err)
+	}
+	err = writer.Flush()
+	if err != nil {
+		log.Printf("[database-yanic] could not send: %s", err)
+	}
+}
+
+func (conn *Connection) InsertLink(link *runtime.Link, time time.Time) {
+}
+
+func (conn *Connection) InsertGlobals(stats *runtime.GlobalStats, time time.Time, site string) {
+}
+
+func (conn *Connection) PruneNodes(deleteAfter time.Duration) {
+}
+
+func (conn *Connection) Close() {
+	conn.conn.Close()
+}

database/respondd/main_test.go

@@ -0,0 +1,49 @@
+package respondd
+
+import (
+	"testing"
+
+	"github.com/FreifunkBremen/yanic/data"
+	"github.com/FreifunkBremen/yanic/runtime"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestStart(t *testing.T) {
+	assert := assert.New(t)
+
+	conn, err := Connect(map[string]interface{}{
+		"type":    "udp6",
+		"address": "fasfs",
+	})
+	assert.Nil(conn)
+	assert.Error(err)
+
+	conn, err = Connect(map[string]interface{}{
+		"type":    "udp",
+		"address": "localhost:11001",
+	})
+	assert.NoError(err)
+
+	conn.InsertNode(&runtime.Node{
+		Nodeinfo: &data.NodeInfo{
+			NodeID:   "73deadbeaf13",
+			Hostname: "inject-test",
+			Network: data.Network{
+				Mac:       "73:de:ad:be:af:13",
+				Addresses: []string{"a", "b"},
+			},
+		},
+		Statistics: &data.Statistics{
+			NodeID: "73deadbeaf13",
+			Clients: data.Clients{
+				Total:  1000,
+				Wifi:   500,
+				Wifi24: 100,
+				Wifi5:  300,
+			},
+		},
+	})
+
+	conn.Close()
+
+}

docs/docs_configuration.md

@@ -579,6 +579,42 @@ prefix = "freifunk"
 
 
 
+## [[database.connection.respondd]]
+{% method %}
+Forward collected respondd package to a address
+(e.g. to another respondd collector like a central yanic instance or hopglass)
+{% sample lang="toml" %}
+```toml
+enable   = false
+type     = "udp6"
+address  = "stats.bremen.freifunk.net:11001"
+```
+{% endmethod %}
+
+
+### type
+{% method %}
+Type of network to create a connection.
+
+Known networks are "tcp", "tcp4" (IPv4-only), "tcp6" (IPv6-only), "udp", "udp4" (IPv4-only), "udp6" (IPv6-only), "ip", "ip4" (IPv4-only), "ip6" (IPv6-only), "unix", "unixgram" and "unixpacket".
+{% sample lang="toml" %}
+```toml
+type     = "udp6"
+```
+{% endmethod %}
+
+
+### address
+{% method %}
+Destination address to connect/send respondd package.
+{% sample lang="toml" %}
+```toml
+address  = "stats.bremen.freifunk.net:11001"
+```
+{% endmethod %}
+
+
+
 ## [[database.connection.logging]]
 {% method %}
 This database type is just for, debugging without a real database connection.