Clean up Collector struct
This commit is contained in:
parent
82c09ad952
commit
54cdd39eb8
|
@ -6,7 +6,6 @@ import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"log"
|
"log"
|
||||||
"net"
|
"net"
|
||||||
"reflect"
|
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/FreifunkBremen/respond-collector/data"
|
"github.com/FreifunkBremen/respond-collector/data"
|
||||||
|
@ -16,16 +15,13 @@ import (
|
||||||
|
|
||||||
//Collector for a specificle respond messages
|
//Collector for a specificle respond messages
|
||||||
type Collector struct {
|
type Collector struct {
|
||||||
CollectType string
|
|
||||||
connection *net.UDPConn // UDP socket
|
connection *net.UDPConn // UDP socket
|
||||||
queue chan *Response // received responses
|
queue chan *Response // received responses
|
||||||
msgType reflect.Type
|
|
||||||
multicastAddr string
|
multicastAddr string
|
||||||
db *database.DB
|
db *database.DB
|
||||||
nodes *models.Nodes
|
nodes *models.Nodes
|
||||||
// Ticker and stopper
|
interval time.Duration // Interval for multicast packets
|
||||||
ticker *time.Ticker
|
stop chan interface{}
|
||||||
stop chan interface{}
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewCollector creates a Collector struct
|
// NewCollector creates a Collector struct
|
||||||
|
@ -49,7 +45,7 @@ func NewCollector(db *database.DB, nodes *models.Nodes, iface string) *Collector
|
||||||
nodes: nodes,
|
nodes: nodes,
|
||||||
multicastAddr: net.JoinHostPort(multiCastGroup+"%"+iface, port),
|
multicastAddr: net.JoinHostPort(multiCastGroup+"%"+iface, port),
|
||||||
queue: make(chan *Response, 400),
|
queue: make(chan *Response, 400),
|
||||||
stop: make(chan interface{}, 1),
|
stop: make(chan interface{}),
|
||||||
}
|
}
|
||||||
|
|
||||||
go collector.receiver()
|
go collector.receiver()
|
||||||
|
@ -64,11 +60,14 @@ func NewCollector(db *database.DB, nodes *models.Nodes, iface string) *Collector
|
||||||
|
|
||||||
// Start Collector
|
// Start Collector
|
||||||
func (coll *Collector) Start(interval time.Duration) {
|
func (coll *Collector) Start(interval time.Duration) {
|
||||||
if coll.ticker != nil {
|
if coll.interval != 0 {
|
||||||
panic("already started")
|
panic("already started")
|
||||||
}
|
}
|
||||||
|
if interval <= 0 {
|
||||||
|
panic("invalid collector interval")
|
||||||
|
}
|
||||||
|
coll.interval = interval
|
||||||
|
|
||||||
coll.ticker = time.NewTicker(interval)
|
|
||||||
go func() {
|
go func() {
|
||||||
coll.sendOnce() // immediately
|
coll.sendOnce() // immediately
|
||||||
coll.sender() // periodically
|
coll.sender() // periodically
|
||||||
|
@ -77,12 +76,7 @@ func (coll *Collector) Start(interval time.Duration) {
|
||||||
|
|
||||||
// Close Collector
|
// Close Collector
|
||||||
func (coll *Collector) Close() {
|
func (coll *Collector) Close() {
|
||||||
// stop ticker
|
close(coll.stop)
|
||||||
if coll.ticker != nil {
|
|
||||||
coll.ticker.Stop()
|
|
||||||
close(coll.stop)
|
|
||||||
}
|
|
||||||
|
|
||||||
coll.connection.Close()
|
coll.connection.Close()
|
||||||
close(coll.queue)
|
close(coll.queue)
|
||||||
}
|
}
|
||||||
|
@ -91,7 +85,7 @@ func (coll *Collector) sendOnce() {
|
||||||
coll.SendPacket(coll.multicastAddr)
|
coll.SendPacket(coll.multicastAddr)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Sends a UDP request to the given unicast or multicast address
|
// SendPacket send a UDP request to the given unicast or multicast address
|
||||||
func (coll *Collector) SendPacket(address string) {
|
func (coll *Collector) SendPacket(address string) {
|
||||||
addr, err := net.ResolveUDPAddr("udp", address)
|
addr, err := net.ResolveUDPAddr("udp", address)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
@ -105,11 +99,13 @@ func (coll *Collector) SendPacket(address string) {
|
||||||
|
|
||||||
// send packets continously
|
// send packets continously
|
||||||
func (coll *Collector) sender() {
|
func (coll *Collector) sender() {
|
||||||
|
ticker := time.NewTicker(coll.interval)
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-coll.stop:
|
case <-coll.stop:
|
||||||
|
ticker.Stop()
|
||||||
return
|
return
|
||||||
case <-coll.ticker.C:
|
case <-ticker.C:
|
||||||
// send the multicast packet to request per-node statistics
|
// send the multicast packet to request per-node statistics
|
||||||
coll.sendOnce()
|
coll.sendOnce()
|
||||||
}
|
}
|
||||||
|
@ -149,13 +145,16 @@ func (coll *Collector) saveResponse(addr net.UDPAddr, res *data.ResponseData) {
|
||||||
nodeID = val.NodeID
|
nodeID = val.NodeID
|
||||||
}
|
}
|
||||||
|
|
||||||
// Updates nodes if NodeID found
|
// Check length of nodeID
|
||||||
if len(nodeID) != 12 {
|
if len(nodeID) != 12 {
|
||||||
log.Printf("invalid NodeID '%s' from %s", nodeID, addr.String())
|
log.Printf("invalid NodeID '%s' from %s", nodeID, addr.String())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Process the data
|
||||||
node := coll.nodes.Update(nodeID, res)
|
node := coll.nodes.Update(nodeID, res)
|
||||||
|
|
||||||
|
// Store statistics in InfluxDB
|
||||||
if coll.db != nil && node.Statistics != nil {
|
if coll.db != nil && node.Statistics != nil {
|
||||||
coll.db.Add(nodeID, node)
|
coll.db.Add(nodeID, node)
|
||||||
}
|
}
|
||||||
|
@ -186,6 +185,7 @@ func (coll *Collector) globalStatsWorker() {
|
||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-coll.stop:
|
case <-coll.stop:
|
||||||
|
ticker.Stop()
|
||||||
return
|
return
|
||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
coll.saveGlobalStats()
|
coll.saveGlobalStats()
|
||||||
|
|
Loading…
Reference in New Issue