2016-02-26 09:28:31 +01:00
|
|
|
package respond
|
2015-12-29 04:08:03 +01:00
|
|
|
|
|
|
|
import (
|
2016-03-19 01:50:23 +01:00
|
|
|
"bytes"
|
|
|
|
"compress/flate"
|
2016-03-12 00:59:36 +01:00
|
|
|
"encoding/json"
|
2017-04-10 18:54:12 +02:00
|
|
|
"fmt"
|
2015-12-29 04:08:03 +01:00
|
|
|
"log"
|
|
|
|
"net"
|
|
|
|
"time"
|
2017-02-01 18:37:39 +01:00
|
|
|
|
2017-03-03 16:19:35 +01:00
|
|
|
"github.com/FreifunkBremen/yanic/data"
|
|
|
|
"github.com/FreifunkBremen/yanic/database"
|
2018-01-07 21:00:56 +01:00
|
|
|
"github.com/FreifunkBremen/yanic/lib/jsontime"
|
2017-04-10 18:54:12 +02:00
|
|
|
"github.com/FreifunkBremen/yanic/runtime"
|
2015-12-29 04:08:03 +01:00
|
|
|
)
|
|
|
|
|
2017-01-29 22:14:40 +01:00
|
|
|
// Collector for a specificle respond messages
|
2015-12-29 04:08:03 +01:00
|
|
|
type Collector struct {
|
2017-12-05 23:17:49 +01:00
|
|
|
connections []multicastConn // UDP sockets
|
2017-10-26 15:24:04 +02:00
|
|
|
port int
|
2017-10-18 18:22:14 +02:00
|
|
|
|
2018-01-17 20:20:35 +01:00
|
|
|
queue chan *Response // received responses
|
|
|
|
db database.Connection
|
|
|
|
nodes *runtime.Nodes
|
|
|
|
sitesDomains map[string][]string
|
|
|
|
interval time.Duration // Interval for multicast packets
|
|
|
|
stop chan interface{}
|
2015-12-29 04:08:03 +01:00
|
|
|
}
|
|
|
|
|
2017-12-05 23:17:49 +01:00
|
|
|
type multicastConn struct {
|
|
|
|
Conn *net.UDPConn
|
|
|
|
MulticastAddress net.IP
|
|
|
|
}
|
|
|
|
|
2017-01-20 22:27:44 +01:00
|
|
|
// NewCollector creates a Collector struct
|
2017-12-05 23:17:49 +01:00
|
|
|
func NewCollector(db database.Connection, nodes *runtime.Nodes, sitesDomains map[string][]string, ifaces []InterfaceConfig) *Collector {
|
2017-10-18 18:22:14 +02:00
|
|
|
|
|
|
|
coll := &Collector{
|
2018-01-17 20:20:35 +01:00
|
|
|
db: db,
|
|
|
|
nodes: nodes,
|
|
|
|
sitesDomains: sitesDomains,
|
|
|
|
queue: make(chan *Response, 400),
|
|
|
|
stop: make(chan interface{}),
|
2017-10-18 18:22:14 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
for _, iface := range ifaces {
|
|
|
|
coll.listenUDP(iface)
|
|
|
|
}
|
|
|
|
|
|
|
|
go coll.parser()
|
|
|
|
|
|
|
|
if coll.db != nil {
|
|
|
|
go coll.globalStatsWorker()
|
|
|
|
}
|
|
|
|
|
|
|
|
return coll
|
|
|
|
}
|
|
|
|
|
2017-12-05 23:17:49 +01:00
|
|
|
func (coll *Collector) listenUDP(iface InterfaceConfig) {
|
|
|
|
|
|
|
|
var addr net.IP
|
|
|
|
|
|
|
|
var err error
|
|
|
|
if iface.IPAddress != "" {
|
|
|
|
addr = net.ParseIP(iface.IPAddress)
|
|
|
|
} else {
|
|
|
|
addr, err = getUnicastAddr(iface.InterfaceName)
|
|
|
|
if err != nil {
|
|
|
|
log.Panic(err)
|
|
|
|
}
|
2017-10-18 18:22:14 +02:00
|
|
|
}
|
2017-12-05 23:17:49 +01:00
|
|
|
|
|
|
|
multicastAddress := multicastAddressDefault
|
|
|
|
if iface.MulticastAddress != "" {
|
|
|
|
multicastAddress = iface.MulticastAddress
|
2015-12-29 04:08:03 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
// Open socket
|
2017-02-01 18:37:39 +01:00
|
|
|
conn, err := net.ListenUDP("udp", &net.UDPAddr{
|
2017-12-05 23:17:49 +01:00
|
|
|
IP: addr,
|
|
|
|
Port: iface.Port,
|
|
|
|
Zone: iface.InterfaceName,
|
2017-02-01 18:37:39 +01:00
|
|
|
})
|
2015-12-29 04:08:03 +01:00
|
|
|
if err != nil {
|
|
|
|
log.Panic(err)
|
|
|
|
}
|
2016-02-25 21:06:37 +01:00
|
|
|
conn.SetReadBuffer(maxDataGramSize)
|
2015-12-29 04:08:03 +01:00
|
|
|
|
2017-12-05 23:17:49 +01:00
|
|
|
coll.connections = append(coll.connections, multicastConn{
|
|
|
|
Conn: conn,
|
|
|
|
MulticastAddress: net.ParseIP(multicastAddress),
|
|
|
|
})
|
2016-03-12 01:04:22 +01:00
|
|
|
|
2017-10-18 18:22:14 +02:00
|
|
|
// Start receiver
|
|
|
|
go coll.receiver(conn)
|
2016-12-22 03:06:46 +01:00
|
|
|
}
|
|
|
|
|
2017-12-05 23:17:49 +01:00
|
|
|
// Returns a unicast address of given interface (prefer global unicast address over link local address)
|
|
|
|
func getUnicastAddr(ifname string) (net.IP, error) {
|
2017-02-01 18:37:39 +01:00
|
|
|
iface, err := net.InterfaceByName(ifname)
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
|
|
|
|
|
|
|
addresses, err := iface.Addrs()
|
|
|
|
if err != nil {
|
|
|
|
return nil, err
|
|
|
|
}
|
2017-12-05 23:17:49 +01:00
|
|
|
var ip net.IP
|
2017-02-01 18:37:39 +01:00
|
|
|
|
|
|
|
for _, addr := range addresses {
|
2017-12-05 23:17:49 +01:00
|
|
|
ipnet, ok := addr.(*net.IPNet)
|
|
|
|
if !ok {
|
|
|
|
continue
|
2017-02-01 18:37:39 +01:00
|
|
|
}
|
2017-12-05 23:17:49 +01:00
|
|
|
if ipnet.IP.IsGlobalUnicast() {
|
|
|
|
ip = ipnet.IP
|
|
|
|
} else if ipnet.IP.IsLinkLocalUnicast() && ip == nil {
|
|
|
|
ip = ipnet.IP
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if ip != nil {
|
|
|
|
return ip, nil
|
2017-02-01 18:37:39 +01:00
|
|
|
}
|
2017-12-05 23:17:49 +01:00
|
|
|
return nil, fmt.Errorf("unable to find a unicast address for %s", ifname)
|
2017-02-01 18:37:39 +01:00
|
|
|
}
|
|
|
|
|
2016-12-22 03:06:46 +01:00
|
|
|
// Start Collector
|
|
|
|
func (coll *Collector) Start(interval time.Duration) {
|
2017-01-29 22:26:16 +01:00
|
|
|
if coll.interval != 0 {
|
2016-12-22 03:06:46 +01:00
|
|
|
panic("already started")
|
|
|
|
}
|
2017-01-29 22:26:16 +01:00
|
|
|
if interval <= 0 {
|
|
|
|
panic("invalid collector interval")
|
|
|
|
}
|
|
|
|
coll.interval = interval
|
2016-12-22 03:06:46 +01:00
|
|
|
|
2016-03-12 18:26:51 +01:00
|
|
|
go func() {
|
2016-12-22 03:06:46 +01:00
|
|
|
coll.sendOnce() // immediately
|
|
|
|
coll.sender() // periodically
|
2016-03-12 18:26:51 +01:00
|
|
|
}()
|
2015-12-29 04:08:03 +01:00
|
|
|
}
|
|
|
|
|
2016-03-11 23:56:23 +01:00
|
|
|
// Close Collector
|
2015-12-29 04:08:03 +01:00
|
|
|
func (coll *Collector) Close() {
|
2017-01-29 22:26:16 +01:00
|
|
|
close(coll.stop)
|
2017-10-18 18:22:14 +02:00
|
|
|
for _, conn := range coll.connections {
|
2017-12-05 23:17:49 +01:00
|
|
|
conn.Conn.Close()
|
2017-10-18 18:22:14 +02:00
|
|
|
}
|
2015-12-29 04:08:03 +01:00
|
|
|
close(coll.queue)
|
|
|
|
}
|
|
|
|
|
2015-12-29 14:05:47 +01:00
|
|
|
func (coll *Collector) sendOnce() {
|
2017-01-29 22:14:40 +01:00
|
|
|
now := jsontime.Now()
|
|
|
|
coll.sendMulticast()
|
|
|
|
|
|
|
|
// Wait for the multicast responses to be processed and send unicasts
|
|
|
|
time.Sleep(coll.interval / 2)
|
|
|
|
coll.sendUnicasts(now)
|
2015-12-29 14:05:47 +01:00
|
|
|
}
|
|
|
|
|
2017-01-29 22:14:40 +01:00
|
|
|
func (coll *Collector) sendMulticast() {
|
2017-10-18 18:22:14 +02:00
|
|
|
log.Println("sending multicasts")
|
|
|
|
for _, conn := range coll.connections {
|
2017-12-05 23:17:49 +01:00
|
|
|
coll.sendPacket(conn.Conn, conn.MulticastAddress)
|
2017-10-18 18:22:14 +02:00
|
|
|
}
|
2017-01-29 22:14:40 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
// Send unicast packets to nodes that did not answer the multicast
|
|
|
|
func (coll *Collector) sendUnicasts(seenBefore jsontime.Time) {
|
|
|
|
seenAfter := seenBefore.Add(-time.Minute * 10)
|
|
|
|
|
|
|
|
// Select online nodes that has not been seen recently
|
2017-04-10 18:54:12 +02:00
|
|
|
nodes := coll.nodes.Select(func(n *runtime.Node) bool {
|
2017-01-29 22:14:40 +01:00
|
|
|
return n.Lastseen.After(seenAfter) && n.Lastseen.Before(seenBefore) && n.Address != nil
|
|
|
|
})
|
|
|
|
|
|
|
|
// Send unicast packets
|
2017-12-05 23:17:49 +01:00
|
|
|
count := 0
|
2017-01-29 22:14:40 +01:00
|
|
|
for _, node := range nodes {
|
2017-12-05 23:17:49 +01:00
|
|
|
send := 0
|
|
|
|
for _, conn := range coll.connections {
|
|
|
|
if node.Address.Zone != "" && conn.Conn.LocalAddr().(*net.UDPAddr).Zone != node.Address.Zone {
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
coll.sendPacket(conn.Conn, node.Address.IP)
|
|
|
|
send++
|
|
|
|
}
|
|
|
|
if send == 0 {
|
2017-10-18 18:22:14 +02:00
|
|
|
log.Printf("unable to find connection for %s", node.Address.Zone)
|
2017-12-05 23:17:49 +01:00
|
|
|
} else {
|
|
|
|
time.Sleep(10 * time.Millisecond)
|
|
|
|
count += send
|
2017-10-18 18:22:14 +02:00
|
|
|
}
|
2017-01-29 22:14:40 +01:00
|
|
|
}
|
2017-12-05 23:17:49 +01:00
|
|
|
log.Printf("sending %d unicast pkg for %d nodes", count, len(nodes))
|
2017-01-29 22:14:40 +01:00
|
|
|
}
|
|
|
|
|
2017-10-18 18:22:14 +02:00
|
|
|
// SendPacket sends a UDP request to the given unicast or multicast address on the first UDP socket
|
|
|
|
func (coll *Collector) SendPacket(destination net.IP) {
|
2017-12-05 23:17:49 +01:00
|
|
|
coll.sendPacket(coll.connections[0].Conn, destination)
|
2017-10-18 18:22:14 +02:00
|
|
|
}
|
|
|
|
|
|
|
|
// sendPacket sends a UDP request to the given unicast or multicast address on the given UDP socket
|
|
|
|
func (coll *Collector) sendPacket(conn *net.UDPConn, destination net.IP) {
|
2017-01-29 22:14:40 +01:00
|
|
|
addr := net.UDPAddr{
|
2017-10-18 18:22:14 +02:00
|
|
|
IP: destination,
|
2017-01-29 22:14:40 +01:00
|
|
|
Port: port,
|
2017-10-18 18:22:14 +02:00
|
|
|
Zone: conn.LocalAddr().(*net.UDPAddr).Zone,
|
2016-02-19 11:13:30 +01:00
|
|
|
}
|
2015-12-29 14:05:47 +01:00
|
|
|
|
2017-10-18 18:22:14 +02:00
|
|
|
if _, err := conn.WriteToUDP([]byte("GET nodeinfo statistics neighbours"), &addr); err != nil {
|
2016-03-07 01:37:07 +01:00
|
|
|
log.Println("WriteToUDP failed:", err)
|
|
|
|
}
|
2015-12-29 04:08:03 +01:00
|
|
|
}
|
|
|
|
|
2018-01-13 16:54:48 +01:00
|
|
|
// send packets continuously
|
2016-03-11 23:56:23 +01:00
|
|
|
func (coll *Collector) sender() {
|
2017-01-29 22:26:16 +01:00
|
|
|
ticker := time.NewTicker(coll.interval)
|
2016-03-11 23:56:23 +01:00
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-coll.stop:
|
2017-01-29 22:26:16 +01:00
|
|
|
ticker.Stop()
|
2016-03-11 23:56:23 +01:00
|
|
|
return
|
2017-01-29 22:26:16 +01:00
|
|
|
case <-ticker.C:
|
2016-10-04 01:05:18 +02:00
|
|
|
// send the multicast packet to request per-node statistics
|
2016-03-11 23:56:23 +01:00
|
|
|
coll.sendOnce()
|
|
|
|
}
|
2015-12-29 04:08:03 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (coll *Collector) parser() {
|
2016-01-04 02:07:09 +01:00
|
|
|
for obj := range coll.queue {
|
2016-10-03 19:55:37 +02:00
|
|
|
if data, err := obj.parse(); err != nil {
|
2017-12-05 23:17:49 +01:00
|
|
|
log.Println("unable to decode response from", obj.Address.String(), err)
|
2016-10-03 19:55:37 +02:00
|
|
|
} else {
|
|
|
|
coll.saveResponse(obj.Address, data)
|
2016-03-12 00:59:36 +01:00
|
|
|
}
|
2015-12-29 04:08:03 +01:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-10-03 19:55:37 +02:00
|
|
|
func (res *Response) parse() (*data.ResponseData, error) {
|
2016-03-19 23:27:07 +01:00
|
|
|
// Deflate
|
2016-10-03 19:55:37 +02:00
|
|
|
deflater := flate.NewReader(bytes.NewReader(res.Raw))
|
2016-03-19 23:27:07 +01:00
|
|
|
defer deflater.Close()
|
2016-03-19 15:07:44 +01:00
|
|
|
|
2016-03-19 23:27:07 +01:00
|
|
|
// Unmarshal
|
2016-10-03 19:55:37 +02:00
|
|
|
rdata := &data.ResponseData{}
|
|
|
|
err := json.NewDecoder(deflater).Decode(rdata)
|
|
|
|
|
|
|
|
return rdata, err
|
|
|
|
}
|
|
|
|
|
2017-10-18 18:22:14 +02:00
|
|
|
func (coll *Collector) saveResponse(addr *net.UDPAddr, res *data.ResponseData) {
|
2016-10-03 19:55:37 +02:00
|
|
|
// Search for NodeID
|
2017-01-20 22:27:44 +01:00
|
|
|
var nodeID string
|
2016-10-03 19:55:37 +02:00
|
|
|
if val := res.NodeInfo; val != nil {
|
2017-01-20 22:27:44 +01:00
|
|
|
nodeID = val.NodeID
|
2016-10-03 19:55:37 +02:00
|
|
|
} else if val := res.Neighbours; val != nil {
|
2017-01-20 22:27:44 +01:00
|
|
|
nodeID = val.NodeID
|
2016-10-03 19:55:37 +02:00
|
|
|
} else if val := res.Statistics; val != nil {
|
2017-01-20 22:27:44 +01:00
|
|
|
nodeID = val.NodeID
|
2016-10-03 19:55:37 +02:00
|
|
|
}
|
|
|
|
|
2017-01-29 22:26:16 +01:00
|
|
|
// Check length of nodeID
|
2017-01-20 22:27:44 +01:00
|
|
|
if len(nodeID) != 12 {
|
|
|
|
log.Printf("invalid NodeID '%s' from %s", nodeID, addr.String())
|
2016-10-03 19:55:37 +02:00
|
|
|
return
|
2016-03-19 15:07:44 +01:00
|
|
|
}
|
2017-01-29 22:26:16 +01:00
|
|
|
|
2017-09-27 13:55:02 +02:00
|
|
|
// Set fields to nil if nodeID is inconsistent
|
|
|
|
if res.Statistics != nil && res.Statistics.NodeID != nodeID {
|
|
|
|
res.Statistics = nil
|
|
|
|
}
|
|
|
|
if res.Neighbours != nil && res.Neighbours.NodeID != nodeID {
|
|
|
|
res.Neighbours = nil
|
|
|
|
}
|
|
|
|
if res.NodeInfo != nil && res.NodeInfo.NodeID != nodeID {
|
|
|
|
res.NodeInfo = nil
|
|
|
|
}
|
|
|
|
|
2017-01-29 22:14:40 +01:00
|
|
|
// Process the data and update IP address
|
2017-01-20 22:27:44 +01:00
|
|
|
node := coll.nodes.Update(nodeID, res)
|
2017-10-18 18:22:14 +02:00
|
|
|
node.Address = addr
|
2016-03-19 15:07:44 +01:00
|
|
|
|
2017-04-18 01:48:38 +02:00
|
|
|
// Store statistics in database
|
2017-09-27 13:55:02 +02:00
|
|
|
if db := coll.db; db != nil {
|
|
|
|
db.InsertNode(node)
|
|
|
|
|
|
|
|
// Store link data
|
|
|
|
if neighbours := node.Neighbours; neighbours != nil {
|
2017-05-20 14:46:29 +02:00
|
|
|
coll.nodes.RLock()
|
2017-09-27 13:55:02 +02:00
|
|
|
for _, link := range coll.nodes.NodeLinks(node) {
|
|
|
|
db.InsertLink(&link, node.Lastseen.GetTime())
|
|
|
|
}
|
2017-05-20 14:46:29 +02:00
|
|
|
coll.nodes.RUnlock()
|
2017-09-27 13:55:02 +02:00
|
|
|
}
|
2016-10-03 19:55:37 +02:00
|
|
|
}
|
2016-03-19 01:50:23 +01:00
|
|
|
}
|
|
|
|
|
2017-10-18 18:22:14 +02:00
|
|
|
func (coll *Collector) receiver(conn *net.UDPConn) {
|
2016-02-25 21:06:37 +01:00
|
|
|
buf := make([]byte, maxDataGramSize)
|
2015-12-29 04:08:03 +01:00
|
|
|
for {
|
2017-10-18 18:22:14 +02:00
|
|
|
n, src, err := conn.ReadFromUDP(buf)
|
2015-12-29 04:08:03 +01:00
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
log.Println("ReadFromUDP failed:", err)
|
2015-12-29 14:05:47 +01:00
|
|
|
return
|
2015-12-29 04:08:03 +01:00
|
|
|
}
|
2016-01-04 02:07:09 +01:00
|
|
|
|
|
|
|
raw := make([]byte, n)
|
|
|
|
copy(raw, buf)
|
|
|
|
|
|
|
|
coll.queue <- &Response{
|
2017-10-18 18:22:14 +02:00
|
|
|
Address: src,
|
2016-01-04 02:07:09 +01:00
|
|
|
Raw: raw,
|
|
|
|
}
|
2015-12-29 04:08:03 +01:00
|
|
|
}
|
|
|
|
}
|
2016-12-15 11:10:09 +01:00
|
|
|
|
|
|
|
func (coll *Collector) globalStatsWorker() {
|
|
|
|
ticker := time.NewTicker(time.Minute)
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case <-coll.stop:
|
2017-01-29 22:26:16 +01:00
|
|
|
ticker.Stop()
|
2016-12-15 11:10:09 +01:00
|
|
|
return
|
|
|
|
case <-ticker.C:
|
|
|
|
coll.saveGlobalStats()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// saves global statistics
|
|
|
|
func (coll *Collector) saveGlobalStats() {
|
2018-01-17 20:20:35 +01:00
|
|
|
stats := runtime.NewGlobalStats(coll.nodes, coll.sitesDomains)
|
2016-12-15 11:10:09 +01:00
|
|
|
|
2018-01-17 20:20:35 +01:00
|
|
|
for site, domains := range stats {
|
|
|
|
for domain, stat := range domains {
|
|
|
|
coll.db.InsertGlobals(stat, time.Now(), site, domain)
|
|
|
|
}
|
2017-11-21 15:12:06 +01:00
|
|
|
}
|
2016-12-15 11:10:09 +01:00
|
|
|
}
|