Save multiple types of data

collector.go

@@ -4,6 +4,8 @@ import (
 	"encoding/json"
 	"log"
 	"net"
+	"reflect"
+	"strings"
 	"time"
 )
 
@@ -11,22 +13,15 @@ const (
 	maxDatagramSize = 8192
 )
 
-type Node struct {
-	Firstseen  time.Time   `json:"firstseen"`
-	Lastseen   time.Time   `json:"lastseen"`
-	Statistics interface{} `json:"statistics"`
-	Nodeinfo   interface{} `json:"nodeinfo"`
-}
-
 type Collector struct {
+	collectType string
 	connection  *net.UDPConn // UDP socket
 	queue       chan string  // received responses
-	nodes      map[string]*Node // the current nodemap
 }
 
-func NewCollector() *Collector {
+func NewCollector(collectType string) *Collector {
 	// Parse address
-	addr, err := net.ResolveUDPAddr("udp", "[::]:1001")
+	addr, err := net.ResolveUDPAddr("udp", "[::]:0")
 	if err != nil {
 		log.Panic(err)
 	}
@@ -39,11 +34,14 @@ func NewCollector() *Collector {
 	conn.SetReadBuffer(maxDatagramSize)
 
 	collector := &Collector{
+		collectType: collectType,
 		connection:  conn,
 		queue:       make(chan string, 100),
-		nodes:      make(map[string]*Node),
 	}
 
+	go collector.sendOnce()
+	go collector.sender()
+
 	go collector.receiver()
 	go collector.parser()
 
@@ -55,26 +53,29 @@ func (coll *Collector) Close() {
 	close(coll.queue)
 }
 
-func (coll *Collector) send(address string) {
+func (coll *Collector) sendOnce() {
-	addr, err := net.ResolveUDPAddr("udp", address)
+	coll.sendPacket("[2a06:8782:ffbb:1337:c24a:ff:fe2c:c7ac]:1001")
-	if err != nil {
+	coll.sendPacket("[2001:bf7:540:0:32b5:c2ff:fe6e:99d5]:1001")
-		log.Panic(err)
-	}
-	coll.connection.WriteToUDP([]byte("nodeinfo"), addr)
 }
 
-func (coll *Collector) print() {
+func (coll *Collector) sendPacket(address string) {
-	b, err := json.Marshal(coll.nodes)
+	addr, err := net.ResolveUDPAddr("udp", address)
-	if err != nil {
+	check(err)
-		log.Panic(err)
+
+	coll.connection.WriteToUDP([]byte(coll.collectType), addr)
+}
+
+func (coll *Collector) sender() {
+	c := time.Tick(collectInterval)
+
+	for range c {
+		coll.sendOnce()
 	}
-	log.Println(string(b))
 }
 
 func (coll *Collector) parser() {
 	for str := range coll.queue {
 		coll.parseSingle(str)
-		coll.print()
 	}
 }
 
@@ -90,29 +91,24 @@ func (coll *Collector) parseSingle(str string) {
 		return
 	}
 
-	now := time.Now()
+	node := nodes.get(nodeId)
-	node, _ := coll.nodes[nodeId]
 
-	if node == nil {
+	// Set result
-		node = &Node{
+	elem := reflect.ValueOf(node).Elem()
-			Firstseen: now,
+	field := elem.FieldByName(strings.Title(coll.collectType))
-		}
+	field.Set(reflect.ValueOf(result))
-		coll.nodes[nodeId] = node
-	}
-
-	node.Lastseen = now
-	node.Nodeinfo = result
 }
 
 func (coll *Collector) receiver() {
 	b := make([]byte, maxDatagramSize)
 	for {
-		n, _, err := coll.connection.ReadFromUDP(b)
+		n, src, err := coll.connection.ReadFromUDP(b)
 
 		if err != nil {
 			log.Println("ReadFromUDP failed:", err)
-		} else {
+			return
-			coll.queue <- string(b[:n])
 		}
+		coll.queue <- string(b[:n])
+		log.Println("received", coll.collectType, "from", src)
 	}
 }

main.go

@@ -1,17 +1,37 @@
 package main
 
 import (
+	"flag"
 	"log"
 	"os"
 	"os/signal"
 	"syscall"
+	"time"
+)
+
+var (
+	nodes           = NewNodes()
+	outputFile      string
+	collectInterval time.Duration
+	saveInterval    time.Duration
 )
 
 func main() {
-	collector := NewCollector()
+	var collectSeconds, saveSeconds int
-	defer collector.Close()
+
-	collector.send("[2a06:8782:ffbb:1337:c24a:ff:fe2c:c7ac]:1001")
+	flag.StringVar(&outputFile, "output", "nodes.json", "path output file")
-	collector.send("[2001:bf7:540:0:32b5:c2ff:fe6e:99d5]:1001")
+	flag.IntVar(&collectSeconds, "collectInterval", 15, "interval for data collections")
+	flag.IntVar(&saveSeconds, "saveInterval", 5, "interval for data saving")
+	flag.Parse()
+
+	collectInterval = time.Second * time.Duration(collectSeconds)
+	saveInterval = time.Second * time.Duration(saveSeconds)
+
+	collectors := []*Collector{
+		NewCollector("statistics"),
+		NewCollector("nodeinfo"),
+		NewCollector("neighbours"),
+	}
 
 	// Wait for SIGINT or SIGTERM
 	sigs := make(chan os.Signal, 1)
@@ -19,4 +39,13 @@ func main() {
 	sig := <-sigs
 	log.Println("received", sig)
 
+	for _, c := range collectors {
+		c.Close()
+	}
+}
+
+func check(e error) {
+	if e != nil {
+		log.Panic(e)
+	}
 }

nodes.go

@@ -0,0 +1,79 @@
+package main
+
+import (
+	"encoding/json"
+	"io/ioutil"
+	"log"
+	"os"
+	"sync"
+	"time"
+)
+
+type Node struct {
+	Firstseen  time.Time   `json:"firstseen"`
+	Lastseen   time.Time   `json:"lastseen"`
+	Statistics interface{} `json:"statistics"`
+	Nodeinfo   interface{} `json:"nodeinfo"`
+	Neighbours interface{} `json:"neighbours"`
+}
+
+type Nodes struct {
+	Version   int              `json:"version"`
+	Timestamp time.Time        `json:"timestamp"`
+	List      map[string]*Node `json:"nodes"` // the current nodemap
+	sync.Mutex
+}
+
+func NewNodes() *Nodes {
+	nodes := &Nodes{
+		Version: 1,
+		List:    make(map[string]*Node),
+	}
+
+	go nodes.saver()
+
+	return nodes
+}
+
+func (nodes *Nodes) get(nodeId string) *Node {
+	now := time.Now()
+
+	nodes.Lock()
+	node, _ := nodes.List[nodeId]
+
+	if node == nil {
+		node = &Node{
+			Firstseen: now,
+		}
+		nodes.List[nodeId] = node
+	}
+	nodes.Unlock()
+
+	node.Lastseen = now
+
+	return node
+}
+
+func (nodes *Nodes) saver() {
+	c := time.Tick(saveInterval)
+
+	for range c {
+		nodes.save()
+	}
+}
+
+func (nodes *Nodes) save() {
+	nodes.Timestamp = time.Now()
+
+	nodes.Lock()
+	data, err := json.Marshal(nodes)
+	nodes.Unlock()
+
+	check(err)
+	log.Println("saving", len(nodes.List), "nodes")
+
+	tmpFile := outputFile + ".tmp"
+
+	check(ioutil.WriteFile(tmpFile, data, 0644))
+	check(os.Rename(tmpFile, outputFile))
+}