From dd9dd0ddc76af072ff2b426d3ba77f29290af824 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kurk=C3=B3=20Mih=C3=A1ly?= Date: Thu, 27 Sep 2018 12:25:01 +0300 Subject: [PATCH] dashboard: experiment 5 --- dashboard/dashboard.go | 5 +- dashboard/geoip.go | 8 + dashboard/message.go | 79 +------- dashboard/peers.go | 445 +++++++++++++++++++++++++++++++---------- 4 files changed, 349 insertions(+), 188 deletions(-) diff --git a/dashboard/dashboard.go b/dashboard/dashboard.go index 895c121751..ff3c602129 100644 --- a/dashboard/dashboard.go +++ b/dashboard/dashboard.go @@ -42,7 +42,7 @@ import ( ) const ( - sampleLimit = 200 // Maximum number of data samples + sampleLimit = 5 // Maximum number of data samples ) // Dashboard contains the dashboard internals. @@ -100,9 +100,6 @@ func New(config *Config, commit string, logdir string) *Dashboard { DiskRead: emptyChartEntries(now, sampleLimit, config.Refresh), DiskWrite: emptyChartEntries(now, sampleLimit, config.Refresh), }, - //Network: &NetworkMessage{ - // PeerBundles: make(map[string]*PeerBundle), - //}, }, logdir: logdir, } diff --git a/dashboard/geoip.go b/dashboard/geoip.go index 943f7c2e02..60d6b2eab7 100644 --- a/dashboard/geoip.go +++ b/dashboard/geoip.go @@ -42,6 +42,14 @@ type GeoDBInfo struct { } `maxminddb:"location" json:"location,omitempty"` } +// GeoLocation contains geographical information. +type GeoLocation struct { + Country string `json:"country,omitempty"` + City string `json:"city,omitempty"` + Latitude float64 `json:"latitude,omitempty"` + Longitude float64 `json:"longitude,omitempty"` +} + // GeoDB represents a geoip database that can be queried for IP to geographical // information conversions. type GeoDB struct { diff --git a/dashboard/message.go b/dashboard/message.go index a931f72296..1f4f6790db 100644 --- a/dashboard/message.go +++ b/dashboard/message.go @@ -26,7 +26,8 @@ type Message struct { Home *HomeMessage `json:"home,omitempty"` Chain *ChainMessage `json:"chain,omitempty"` TxPool *TxPoolMessage `json:"txpool,omitempty"` - Network *PeerContainer `json:"network,omitempty"` + Network *NetworkMessage `json:"network,omitempty"` + Peers *PeersMessage `json:"peers,omitempty"` System *SystemMessage `json:"system,omitempty"` Logs *LogsMessage `json:"logs,omitempty"` } @@ -57,83 +58,9 @@ type TxPoolMessage struct { // NetworkMessage contains information about the peers organized based on the IP address. type NetworkMessage struct { - PeerBundles map[string]*PeerBundle `json:"peerBundles,omitempty"` + /* TODO (kurkomisi) */ } -// getOrInitBundle returns the peer bundle belonging to the given IP, or -// initializes the bundle if it doesn't exist. -//func (m *NetworkMessage) getOrInitBundle(ip string) *PeerBundle { -// if _, ok := m.PeerBundles[ip]; !ok { -// m.PeerBundles[ip] = &PeerBundle{ -// Peers: make(PeerMap), -// FailedPeers: make(PeerMap), -// } -// } -// return m.PeerBundles[ip] -//} -// -//// getOrInitPeer returns the peer belonging to the given IP and node id, or -//// initializes the peer if it doesn't exist. -//func (m *NetworkMessage) getOrInitPeer(ip, id string) *Peer { -// return m.getOrInitBundle(ip).getOrInitPeer(id) -//} - -//type PeerMap map[string]*Peer - -//func (pm PeerMap) getOrInitBundle(id string) *Peer { -// if _, ok := pm[id]; !ok { -// pm[id] = new(Peer) -// } -// return pm[id] -//} -// -//func (pm PeerMap) remove(id string) { -// delete(pm, id) -//} - -//// PeerBundle contains information about the peers pertaining to an IP address. -//type PeerBundle struct { -// Location *GeoLocation `json:"location,omitempty"` // geographical information based on IP -// Peers PeerMap `json:"peers,omitempty"` // the peers' node id is used as key -// FailedPeers PeerMap `json:"failedPeers,omitempty"` -//} -// -//func (b *PeerBundle) getOrInitPeer(id string) *Peer { -// return b.Peers.getOrInitBundle(id) -//} -// -//func (b *PeerBundle) removePeer(id string) { -// b.Peers.remove(id) -//} -// -//func (b *PeerBundle) getOrInitFailedPeer(id string) *Peer { -// return b.FailedPeers.getOrInitBundle(id) -//} -// -//func (b * PeerBundle) removeFailedPeer(id string) { -// b.FailedPeers.remove(id) -//} - -// GeoLocation contains geographical information. -type GeoLocation struct { - Country string `json:"country,omitempty"` - City string `json:"city,omitempty"` - Latitude float64 `json:"latitude,omitempty"` - Longitude float64 `json:"longitude,omitempty"` -} - -// Peer contains lifecycle timestamps and traffic information of a given peer. -//type Peer struct { -// Connected []time.Time `json:"connected,omitempty"` -// Disconnected []time.Time `json:"disconnected,omitempty"` -// -// Ingress ChartEntries `json:"ingress,omitempty"` -// Egress ChartEntries `json:"egress,omitempty"` -// -// element *list.Element -// ip, id string -//} - // SystemMessage contains the metered system data samples. type SystemMessage struct { ActiveMemory ChartEntries `json:"activeMemory,omitempty"` diff --git a/dashboard/peers.go b/dashboard/peers.go index ca896180b4..490bf7d6f8 100644 --- a/dashboard/peers.go +++ b/dashboard/peers.go @@ -34,149 +34,286 @@ const ( unknownPeerLimit = p2p.MeteredPeerLimit ) -type PeerContainer struct { - Bundles map[string]*PeerBundle `json:"peerBundles,omitempty"` +// PeersMessage contains information about the node's peers. This data structure +// tries to maintain the metered peer data based on the different behaviours of +// the peers. +// +// Every peer has an IP address, and the peers that manage to make the handshake +// have node IDs. There can appear more peers with the same IP, therefore the +// peer maintainer data structure is a tree consisting of a map of maps, where +// the first key groups the peers by IP, while the second one groups them by the +// node ID. The peers failing before the handshake only have IP addresses, so +// they are stored as part of the value of the outer map. A peer can connect +// multiple times, so an array is needed to store the consecutive sessions. +// +// Another criteria is to limit the number of metered peers so that they don't +// fill the memory. The metered peers are selected based on their activity: the +// peers that are inactive for the longest time are thrown first. For the selection a fifo +// list is used which is linked to the bottom of the peer tree, and when a peer +// is removed from the list, it is also removed from the tree. The active peers +// have priority over the disconnected ones, therefore this list is extended by +// a separator, which is a pointer to a list element. The separator separates the +// active peers from the inactive ones, and it is the entry for the list. If the +// peer that is to be inserted is active, it goes before the separator, otherwise +// it goes after. This way the peers that are active for the longest time are at +// the beginning of the list, and the inactive ones move to the end. When a peer +// has some activity, it is removed from and reinserted into the list. +// +// The peers that don't manage to make handshake are not inserted into the list, +// their sessions are not stored, only their connection attempts are appended +// to the array belonging to their IP. In order to keep the fifo principle, a super +// array contains the order of the attempts, and when the peer count reaches the +// limit, the earliest attempt is removed from the beginning of its array. +type PeersMessage struct { + // Bundles is the outer map using the peer's IP address as key. + Bundles map[string]*PeerBundle `json:"bundles,omitempty"` + RemovedKnown []string `json:"removedKnown,omitempty"` + RemovedUnknown []string `json:"removedUnknown,omitempty"` + + // activeSeparator is a pointer to the last active peer element, splitting + // the list into an active and inactive part, and forming the entry for the + // peer list. activeSeparator *list.Element - knownPeers *list.List - unknownPeers []*PeerBundle - geodb *GeoDB - refresh time.Duration + // knownPeers contains the peers that managed to make handshake. + knownPeers *list.List + + // unknownPeers contains pointers to the peer bundles that belong to the + // IP addresses, from which the peers attempted to connect then failed. + // Its values are appended in the moment of the attempt, so in chronological + // order, which means that the oldest attempt is at the beginning of the array. + // When the first element is removed, the first element of the linked bundle's + // attempt array is also removed, ensuring that always the latest attempts are stored. + unknownPeers []*PeerBundle + + // geodb is used to look up the geographical data based on the IP. + geodb *GeoDB } -func NewPeerContainer(geodb *GeoDB, refresh time.Duration) *PeerContainer { - return &PeerContainer{ +// NewPeersMessage returns a new instance of the metered peer maintainer data structure. +func NewPeersMessage(geodb *GeoDB) *PeersMessage { + return &PeersMessage{ Bundles: make(map[string]*PeerBundle), knownPeers: list.New(), unknownPeers: make([]*PeerBundle, 0, unknownPeerLimit), geodb: geodb, - refresh: refresh, } } -func (pc *PeerContainer) getOrInitBundle(ip string) *PeerBundle { - if _, ok := pc.Bundles[ip]; !ok { - pc.Bundles[ip] = &PeerBundle{ - Location: pc.geodb.Location(ip), +func (m *PeersMessage) hasBundle(ip string) bool { + _, ok := m.Bundles[ip] + return ok +} + +func (m *PeersMessage) hasPeer(ip, id string) bool { + if !m.hasBundle(ip) { + return false + } + return m.Bundles[ip].has(id) +} + +// getOrInitBundle returns the bundle belonging to the given IP. +// Inserts a new bundle into the map if it doesn't already exist. +func (m *PeersMessage) getOrInitBundle(ip string) *PeerBundle { + if _, ok := m.Bundles[ip]; !ok { + m.Bundles[ip] = &PeerBundle{ + Location: m.geodb.Location(ip), KnownPeers: make(map[string]*KnownPeer), - parent: pc, + root: m, ip: ip, } } - return pc.Bundles[ip] + return m.Bundles[ip] } -func (pc *PeerContainer) updateKnown(ip, id string, session *PeerSession) (removedIP, removedID string) { - peer := pc.getOrInitBundle(ip).getOrInitKnown(id) - if peer.Sessions == nil { - peer.Sessions = []*PeerSession{{ - Ingress: emptyChartEntries(time.Now(), 5, pc.refresh), - Egress: emptyChartEntries(time.Now(), 5, pc.refresh), - }} - } +func (m *PeersMessage) getOrInitKnownPeer(ip, id string) *KnownPeer { + return m.getOrInitBundle(ip).getOrInitKnownPeer(id) +} + +// updateKnownPeer updates the last session of the peer belonging to the given IP +// and ID and returns the IP and the ID of the removed peer if there is any, +// (nil, nil) otherwise. +func (m *PeersMessage) updateKnownPeer(ip, id string, session *PeerSession) (removedKey string) { + peer := m.getOrInitKnownPeer(ip, id) if peer.listElement != nil { - if peer.listElement == pc.activeSeparator { - pc.activeSeparator = pc.activeSeparator.Prev() + // If the peer is already part of the list, remove it first. + if peer.listElement == m.activeSeparator { + m.activeSeparator = m.activeSeparator.Prev() } - pc.knownPeers.Remove(peer.listElement) + m.knownPeers.Remove(peer.listElement) } - if pc.knownPeers.Len() >= knownPeerLimit { - removed := pc.knownPeers.Remove(pc.knownPeers.Back()) + if m.knownPeers.Len() >= knownPeerLimit { + // If the peer count reached the limit, remove the last element of the + // list, which is the oldest inactive one. + removed := m.knownPeers.Remove(m.knownPeers.Back()) if p, ok := removed.(*KnownPeer); ok { - removedIP, removedID = p.parent.ip, p.id - p.delete() + removedKey = fmt.Sprintf("%s/%s", p.bundle.ip, p.id) + p.delete() // Remove the peer from the tree. } else { log.Warn("Bad value used as peer", "value", removed) } } - if pc.activeSeparator == nil { - peer.listElement = pc.knownPeers.PushFront(peer) + if m.activeSeparator == nil { + // If there isn't any active peer in the list, push the new peer to the + // front of the list. + peer.listElement = m.knownPeers.PushFront(peer) } else { - peer.listElement = pc.knownPeers.InsertAfter(peer, pc.activeSeparator) + // If there are active peers in the list, push the new peer after them. + peer.listElement = m.knownPeers.InsertAfter(peer, m.activeSeparator) } peer.update(session) if peer.Sessions[len(peer.Sessions)-1].Disconnected == nil { - pc.activeSeparator = peer.listElement + // If the new peer is active, step to it with the separator. + m.activeSeparator = peer.listElement } - return removedIP, removedID + return removedKey } -func (pc *PeerContainer) updateUnknown(ip string, peer *UnknownPeer) { - if len(pc.unknownPeers) >= unknownPeerLimit { - removed := pc.unknownPeers[0] +// updateUnknownPeer inserts a peer connection attempt into the peer tree. +func (m *PeersMessage) updateUnknownPeer(ip string, peer *UnknownPeer) (removedKey string) { + if len(m.unknownPeers) >= unknownPeerLimit { + // If the count of the metered unknown peers reached the limit, + // remove the oldest attempt, which is the first element of the + // array belonging to the peer bundle pointed by the first element + // of the super unknown peer array. + removed := m.unknownPeers[0] removed.UnknownPeers = removed.UnknownPeers[1:] - pc.unknownPeers = pc.unknownPeers[1:] + m.unknownPeers = m.unknownPeers[1:] + removedKey = removed.ip } - bundle := pc.getOrInitBundle(ip) + bundle := m.getOrInitBundle(ip) bundle.UnknownPeers = append(bundle.UnknownPeers, peer) - pc.unknownPeers = append(pc.unknownPeers, bundle) + m.unknownPeers = append(m.unknownPeers, bundle) + + return removedKey } -func (pc *PeerContainer) clear() { - for pc.knownPeers.Front() != nil { - first := pc.knownPeers.Front() - if p, ok := first.Value.(*KnownPeer); ok { +// clear removes the elements of the metered peer list, and removes the linked +// peers from the peer tree along the way. +func (m *PeersMessage) clear() { + for m.knownPeers.Front() != nil { + first := m.knownPeers.Remove(m.knownPeers.Front()) + if p, ok := first.(*KnownPeer); ok { p.delete() } else { log.Warn("Bad value used as peer", "value", first) } } - pc.activeSeparator = nil + m.activeSeparator = nil } -func (pc *PeerContainer) updateTraffic(traffic *peerTraffic) { - now := time.Now() - for e := pc.knownPeers.Front(); e != nil; e = e.Next() { - if p, ok := e.Value.(*KnownPeer); ok { - key := fmt.Sprintf("%s/%s", p.parent.ip, p.id) - p.updateTraffic( - &ChartEntry{Time: now, Value: float64(traffic.ingress[key])}, - &ChartEntry{Time: now, Value: float64(traffic.egress[key])}, - ) - } else { - log.Warn("Bad value used as peer", "value", e.Value) +func (m *PeersMessage) updateTraffic(ip, id string, ingress, egress float64) { + if !m.hasPeer(ip, id) { + m.updateKnownPeer(ip, id, &PeerSession{ + Ingress: ChartEntries{&ChartEntry{ + Value: ingress, + }}, + Egress: ChartEntries{&ChartEntry{ + Value: egress, + }}, + }) + return + } + m.getOrInitKnownPeer(ip, id).updateTraffic( + &ChartEntry{Value: ingress}, + &ChartEntry{Value: egress}, + ) +} + +//func (m *PeersMessage) updateTraffic(ingress, egress *map[string]float64) { +// now := time.Now() +// for k, v := range *ingress { +// +// } +// for e := m.knownPeers.Front(); e != nil; e = e.Next() { +// if p, ok := e.Value.(*KnownPeer); ok { +// key := fmt.Sprintf("%s/%s", p.bundle.ip, p.id) +// p.updateTraffic( +// &ChartEntry{Time: now, Value: (*ingress)[key]}, +// &ChartEntry{Time: now, Value: (*egress)[key]}, +// ) +// } else { +// log.Warn("Bad value used as peer", "value", e.Value) +// } +// } +//} + +// append +func (m *PeersMessage) append(n *PeersMessage) (removedKnown, removedUnknown []string) { + for _, bundle := range n.Bundles { + for _, peer := range bundle.KnownPeers { + for _, session := range peer.Sessions { + removedKey := m.updateKnownPeer(bundle.ip, peer.id, session) + if removedKey != "" { + removedKnown = append(removedKnown, removedKey) + } + } + } + for _, peer := range bundle.UnknownPeers { + removedKey := m.updateUnknownPeer(bundle.ip, peer) + if removedKey != "" { + removedUnknown = append(removedUnknown, removedKey) + } } } + return removedKnown, removedUnknown } +// PeerBundle contains the peers belonging to a given IP address type PeerBundle struct { - Location *GeoLocation `json:"location,omitempty"` - KnownPeers map[string]*KnownPeer `json:"knownPeers,omitempty"` - UnknownPeers []*UnknownPeer `json:"unknownPeers,omitempty"` + Location *GeoLocation `json:"location,omitempty"` // Geographical location based on IP - parent *PeerContainer - ip string - element list.Element + // KnownPeers is the inner map of the metered peer maintainer data structure + // using the node ID as key. + KnownPeers map[string]*KnownPeer `json:"knownPeers,omitempty"` + + // UnknownPeers contains the failed connection attempts of the peers + // belonging to a given IP address in chronological order. + UnknownPeers []*UnknownPeer `json:"unknownPeers,omitempty"` + + root *PeersMessage // Pointer to the outer map. + ip string // Key of the bundle in the outer map. } -func (b *PeerBundle) getOrInitKnown(id string) *KnownPeer { +func (b *PeerBundle) has(id string) bool { + _, ok := b.KnownPeers[id] + return ok +} + +// getOrInitKnownPeer returns the peer belonging to the given ID. +// Initializes it if it doesn't already exist. +func (b *PeerBundle) getOrInitKnownPeer(id string) *KnownPeer { if _, ok := b.KnownPeers[id]; !ok { b.KnownPeers[id] = &KnownPeer{ - parent: b, - id: id, + sampleCount: sampleLimit, + bundle: b, + id: id, } } return b.KnownPeers[id] } +// KnownPeer contains the metered data of a particular peer. type KnownPeer struct { + // Sessions contains the metered data in a session of a peer. Sessions []*PeerSession `json:"sessions,omitempty"` - parent *PeerBundle - id string + sampleCount int - listElement *list.Element + bundle *PeerBundle // Pointer to the inner map. + id string // Key of the peer in the inner map. + + listElement *list.Element // Pointer to the peer element in the list. } func (peer *KnownPeer) delete() { - delete(peer.parent.KnownPeers, peer.id) - fmt.Println(peer, "to make sure p is not cleared") - if len(peer.parent.KnownPeers) < 1 && len(peer.parent.UnknownPeers) < 1 { - delete(peer.parent.parent.Bundles, peer.parent.ip) + delete(peer.bundle.KnownPeers, peer.id) + if len(peer.bundle.KnownPeers) < 1 && len(peer.bundle.UnknownPeers) < 1 { + delete(peer.bundle.root.Bundles, peer.bundle.ip) } peer.listElement = nil - peer.parent = nil + peer.bundle = nil for i := range peer.Sessions { peer.Sessions[i] = nil } @@ -184,16 +321,35 @@ func (peer *KnownPeer) delete() { } func (peer *KnownPeer) update(session *PeerSession) { + if peer.Sessions == nil { + peer.Sessions = []*PeerSession{session} + if session.Connected != nil && session.Disconnected != nil && session.Connected.After(*session.Disconnected) { + peer.Sessions = append(peer.Sessions, &PeerSession{Connected: session.Connected}) + session.Connected = nil + } + return + } if session.Connected != nil && session.Disconnected != nil && session.Connected.After(*session.Disconnected) { - peer.Sessions[len(peer.Sessions)-1].Disconnected = session.Disconnected + last := peer.Sessions[len(peer.Sessions)-1] + if last.Connected == nil { + // This may happen at the beginning, if the connection was established + // before the initialization of the peer event feed. Otherwise it should + // not happen to handle a disconnect event before the connect, because + // both of them are coming on the same channel consecutively. + log.Warn("Disconnect event appeared without connect") + } + last.Disconnected = session.Disconnected + last.Ingress = append(last.Ingress, session.Ingress...) + last.Egress = append(last.Egress, session.Egress...) session.Disconnected = nil + session.Ingress = nil + session.Egress = nil } if session.Connected != nil { peer.Sessions = append(peer.Sessions, session) } - last := peer.Sessions[len(peer.Sessions)-1] if session.Disconnected != nil { - last.Disconnected = session.Disconnected + peer.Sessions[len(peer.Sessions)-1].Disconnected = session.Disconnected } for i := 0; i < len(session.Ingress) && i < len(session.Egress); i++ { peer.updateTraffic(session.Ingress[i], session.Egress[i]) @@ -201,15 +357,36 @@ func (peer *KnownPeer) update(session *PeerSession) { } func (peer *KnownPeer) updateTraffic(ingress, egress *ChartEntry) { - first, last := peer.Sessions[0], peer.Sessions[len(peer.Sessions)-1] - if len(first.Ingress) < 2 || len(first.Egress) < 2 { - peer.Sessions = peer.Sessions[1:] - } else { - first.Ingress = first.Ingress[1:] - first.Egress = first.Egress[1:] + if ingress == nil { + ingress = new(ChartEntry) } + if egress == nil { + egress = new(ChartEntry) + } + first := peer.Sessions[0] + if len(first.Ingress) < 1 || len(first.Egress) < 1 { + first.Ingress = append(first.Ingress, ingress) + first.Egress = append(first.Egress, egress) + peer.sampleCount = 1 + return + } + if peer.sampleCount >= sampleLimit { + if len(first.Ingress) < 2 || len(first.Egress) < 2 { + peer.Sessions = peer.Sessions[1:] + } else { + first.Ingress = first.Ingress[1:] + first.Egress = first.Egress[1:] + } + peer.sampleCount-- + } + last := peer.Sessions[len(peer.Sessions)-1] last.Ingress = append(last.Ingress, ingress) last.Egress = append(last.Egress, egress) + peer.sampleCount++ +} + +func (peer *KnownPeer) len() int { + return peer.sampleCount } type PeerSession struct { @@ -221,13 +398,8 @@ type PeerSession struct { } type UnknownPeer struct { - Connected time.Time `json:"connected,omitempty"` - Disconnected time.Time `json:"disconnected,omitempty"` -} - -type peerTraffic struct { - ingress map[string]int64 - egress map[string]int64 + Connected time.Time `json:"connected"` + Disconnected time.Time `json:"disconnected"` } // collectPeerData gathers data about the peers and sends it to the clients. @@ -250,19 +422,24 @@ func (db *Dashboard) collectPeerData() { ticker := time.NewTicker(db.config.Refresh) defer ticker.Stop() - pc := NewPeerContainer(db.geodb, db.config.Refresh) + db.peerLock.Lock() + db.history.Peers = NewPeersMessage(db.geodb) + db.peerLock.Unlock() + diff := NewPeersMessage(db.geodb) - trafficUpdater := func(prefix string) func(*map[string]int64) func(name string, i interface{}) { - return func(entryMap *map[string]int64) func(name string, i interface{}) { + trafficCollector := func(prefix string) func(*map[string]float64) func(name string, i interface{}) { + return func(traffic *map[string]float64) func(name string, i interface{}) { return func(name string, i interface{}) { if m, ok := i.(metrics.Meter); ok { - (*entryMap)[strings.TrimPrefix(name, prefix)] = m.Count() + (*traffic)[strings.TrimPrefix(name, prefix)] = float64(m.Count()) + } else { + log.Warn("Bad value used as meter", "name", name) } } } } - updateIngress := trafficUpdater(p2p.MetricsInboundTraffic + "/") - updateEgress := trafficUpdater(p2p.MetricsOutboundTraffic + "/") + collectIngress := trafficCollector(p2p.MetricsInboundTraffic + "/") + collectEgress := trafficCollector(p2p.MetricsOutboundTraffic + "/") for { select { @@ -271,11 +448,11 @@ func (db *Dashboard) collectPeerData() { switch event.Type { case p2p.PeerConnected: connected := now.Add(-event.Elapsed) - pc.updateKnown(event.IP.String(), event.ID, &PeerSession{ + diff.updateKnownPeer(event.IP.String(), event.ID, &PeerSession{ Connected: &connected, }) case p2p.PeerDisconnected: - pc.updateKnown(event.IP.String(), event.ID, &PeerSession{ + diff.updateKnownPeer(event.IP.String(), event.ID, &PeerSession{ Disconnected: &now, Ingress: ChartEntries{ &ChartEntry{ @@ -291,7 +468,7 @@ func (db *Dashboard) collectPeerData() { }, }) case p2p.PeerHandshakeFailed: - pc.updateUnknown(event.IP.String(), &UnknownPeer{ + diff.updateUnknownPeer(event.IP.String(), &UnknownPeer{ Connected: now.Add(-event.Elapsed), Disconnected: now, }) @@ -299,16 +476,68 @@ func (db *Dashboard) collectPeerData() { log.Error("Unknown metered peer event type", "type", event.Type) } case <-ticker.C: - traffic := peerTraffic{ - ingress: make(map[string]int64), - egress: make(map[string]int64), + ingress, egress := make(map[string]float64), make(map[string]float64) + db.peerLock.Lock() + for e := db.history.Peers.knownPeers.Front(); e != nil; e = e.Next() { + if p, ok := e.Value.(*KnownPeer); ok { + key := fmt.Sprintf("%s/%s", p.bundle.ip, p.id) + ingress[key] = 0 + egress[key] = 0 + } } - p2p.PeerIngressRegistry.Each(updateIngress(&traffic.ingress)) - p2p.PeerEgressRegistry.Each(updateEgress(&traffic.egress)) - pc.updateTraffic(&traffic) - s, _ := json.MarshalIndent(pc, "", " ") - fmt.Println(string(s)) - //db.sendToAll(&Message{Network: pc}) + p2p.PeerIngressRegistry.Each(collectIngress(&ingress)) + p2p.PeerEgressRegistry.Each(collectEgress(&egress)) + //diff.updateTraffic(&ingress, &egress) + for key := range ingress { + if k := strings.Split(key, "/"); len(k) == 2 { + diff.updateTraffic(k[0], k[1], ingress[key], egress[key]) + } else { + log.Warn("Bad key", "key", key) + } + } + for key := range egress { + if _, ok := ingress[key]; !ok { + if k := strings.Split(key, "/"); len(k) == 2 { + diff.updateTraffic(k[0], k[1], ingress[key], egress[key]) + } else { + log.Warn("Bad key", "key", key) + } + } + } + for _, bundle := range diff.Bundles { + for _, peer := range bundle.KnownPeers { + var ipExists, idExists bool + var b *PeerBundle + if b, ipExists = db.history.Peers.Bundles[bundle.ip]; ipExists { + _, idExists = b.KnownPeers[peer.id] + } + if !idExists || !ipExists { + fmt.Println("doesn't exist ", bundle.ip, peer.id) + //t, n := time.Now(), sampleLimit-peer.len() + //if len(peer.Sessions) > 0 && peer.Sessions[0].Connected != nil { + // t = *peer.Sessions[0].Connected + //} + //peer.Sessions = append([]*PeerSession{{ + // Ingress: emptyChartEntries(t, n, db.config.Refresh), + // Egress: emptyChartEntries(t, n, db.config.Refresh), + //}}, peer.Sessions...) + } else { + fmt.Println("does exist ", bundle.ip, peer.id) + s, _ := json.MarshalIndent(bundle, "", " ") + fmt.Println(string(s)) + fmt.Println(ingress[fmt.Sprintf("%s/%s", bundle.ip, peer.id)]) + fmt.Println(egress[fmt.Sprintf("%s/%s", bundle.ip, peer.id)]) + } + } + } + diff.RemovedKnown, diff.RemovedUnknown = db.history.Peers.append(diff) + //sh, _ := json.MarshalIndent(db.history.Network, "", " ") + //fmt.Println(string(sh)) + db.peerLock.Unlock() + //s, _ := json.MarshalIndent(diff, "", " ") + //fmt.Println(string(s)) + //db.sendToAll(&Message{Peers: diff}) + diff.clear() case err := <-subPeer.Err(): log.Warn("Peer subscription error", "err", err) return