aboutsummaryrefslogtreecommitdiffstats
path: root/dashboard/peers.go
blob: 334302cc120e1b44e45200d753e1c8fc8f507d17 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
// Copyright 2019 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.

package dashboard

import (
    "container/list"
    "strings"
    "time"

    "github.com/ethereum/go-ethereum/metrics"

    "github.com/ethereum/go-ethereum/log"
    "github.com/ethereum/go-ethereum/p2p"
)

const (
    eventBufferLimit = 128 // Maximum number of buffered peer events.
    knownPeerLimit   = 100 // Maximum number of stored peers, which successfully made the handshake.
    attemptLimit     = 200 // Maximum number of stored peers, which failed to make the handshake.

    // eventLimit is the maximum number of the dashboard's custom peer events,
    // that are collected between two metering period and sent to the clients
    // as one message.
    // TODO (kurkomisi): Limit the number of events.
    eventLimit = knownPeerLimit << 2
)

// peerContainer contains information about the node's peers. This data structure
// maintains 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
// (known peers) have node IDs too. There can appear more peers with the same IP,
// therefore the peer container 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 known peers can be active if their connection is still
// open, or inactive otherwise. The peers failing before the handshake (unknown
// peers) only have IP addresses, so their connection attempts are stored as part
// of the value of the outer map.
//
// Another criteria is to limit the number of metered peers so that
// they don't fill the memory. The selection order is based on the
// peers 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 in a way that every activity
// of the peer pushes the peer to the end of the list, so the inactive
// ones come to the front. When a peer has some activity, it is removed
// from and reinserted into the list. When the length of the list reaches
// the limit, the first element is removed from the list, as well as from
// the tree.
//
// The active peers have priority over the inactive ones, therefore
// they have their own list. The separation makes it sure that the
// inactive peers are always removed before the active ones.
//
// The peers that don't manage to make handshake are not inserted into the list,
// 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 overall count reaches the limit, the earliest attempt is
// removed from the beginning of its array.
//
// This data structure makes it possible to marshal the peer
// history simply by passing it to the JSON marshaler.
type peerContainer struct {
    // Bundles is the outer map using the peer's IP address as key.
    Bundles map[string]*peerBundle `json:"bundles,omitempty"`

    activeCount int // Number of the still connected peers

    // inactivePeers contains the peers with closed connection in chronological order.
    inactivePeers *list.List

    // attemptOrder is the super array containing the IP addresses, from which
    // the peers attempted to connect then failed before/during the handshake.
    // Its values are appended 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 related bundle's attempt array is
    // removed too, ensuring that always the latest attempts are stored.
    attemptOrder []string

    // geodb is the geoip database used to retrieve the peers' geographical location.
    geodb *geoDB
}

// newPeerContainer returns a new instance of the peer container.
func newPeerContainer(geodb *geoDB) *peerContainer {
    return &peerContainer{
        Bundles:       make(map[string]*peerBundle),
        inactivePeers: list.New(),
        attemptOrder:  make([]string, 0, attemptLimit),
        geodb:         geodb,
    }
}

// bundle inserts a new peer bundle into the map, if the peer belonging
// to the given IP wasn't metered so far. In this case retrieves the location of
// the IP address from the database and creates a corresponding peer event.
// Returns the bundle belonging to the given IP and the events occurring during
// the initialization.
func (pc *peerContainer) bundle(ip string) (*peerBundle, []*peerEvent) {
    var events []*peerEvent
    if _, ok := pc.Bundles[ip]; !ok {
        location := pc.geodb.location(ip)
        events = append(events, &peerEvent{
            IP:       ip,
            Location: location,
        })
        pc.Bundles[ip] = &peerBundle{
            Location:   location,
            KnownPeers: make(map[string]*knownPeer),
        }
    }
    return pc.Bundles[ip], events
}

// extendKnown handles the events of the successfully connected peers.
// Returns the events occurring during the extension.
func (pc *peerContainer) extendKnown(event *peerEvent) []*peerEvent {
    bundle, events := pc.bundle(event.IP)
    peer, peerEvents := bundle.knownPeer(event.IP, event.ID)
    events = append(events, peerEvents...)
    // Append the connect and the disconnect events to
    // the corresponding arrays keeping the limit.
    switch {
    case event.Connected != nil:
        peer.Connected = append(peer.Connected, event.Connected)
        if first := len(peer.Connected) - sampleLimit; first > 0 {
            peer.Connected = peer.Connected[first:]
        }
        peer.Active = true
        events = append(events, &peerEvent{
            Activity: Active,
            IP:       peer.ip,
            ID:       peer.id,
        })
        pc.activeCount++
        if peer.listElement != nil {
            _ = pc.inactivePeers.Remove(peer.listElement)
            peer.listElement = nil
        }
    case event.Disconnected != nil:
        peer.Disconnected = append(peer.Disconnected, event.Disconnected)
        if first := len(peer.Disconnected) - sampleLimit; first > 0 {
            peer.Disconnected = peer.Disconnected[first:]
        }
        peer.Active = false
        events = append(events, &peerEvent{
            Activity: Inactive,
            IP:       peer.ip,
            ID:       peer.id,
        })
        pc.activeCount--
        if peer.listElement != nil {
            // If the peer is already in the list, remove and reinsert it.
            _ = pc.inactivePeers.Remove(peer.listElement)
        }
        // Insert the peer into the list.
        peer.listElement = pc.inactivePeers.PushBack(peer)
    }
    for pc.inactivePeers.Len() > 0 && pc.activeCount+pc.inactivePeers.Len() > knownPeerLimit {
        // While the count of the known peers is greater than the limit,
        // remove the first element from the inactive peer list and from the map.
        if removedPeer, ok := pc.inactivePeers.Remove(pc.inactivePeers.Front()).(*knownPeer); ok {
            events = append(events, pc.removeKnown(removedPeer.ip, removedPeer.id)...)
        } else {
            log.Warn("Failed to parse the removed peer")
        }
    }
    if pc.activeCount > knownPeerLimit {
        log.Warn("Number of active peers is greater than the limit")
    }
    return events
}

// handleAttempt handles the events of the peers failing before/during the handshake.
// Returns the events occurring during the extension.
func (pc *peerContainer) handleAttempt(event *peerEvent) []*peerEvent {
    bundle, events := pc.bundle(event.IP)
    bundle.Attempts = append(bundle.Attempts, &peerAttempt{
        Connected:    *event.Connected,
        Disconnected: *event.Disconnected,
    })
    pc.attemptOrder = append(pc.attemptOrder, event.IP)
    for len(pc.attemptOrder) > attemptLimit {
        // While the length of the connection attempt order array is greater
        // than the limit, remove the first element from the involved peer's
        // array and also from the super array.
        events = append(events, pc.removeAttempt(pc.attemptOrder[0])...)
        pc.attemptOrder = pc.attemptOrder[1:]
    }
    return events
}

// peerBundle contains the peers belonging to a given IP address.
type peerBundle struct {
    // Location contains the geographical location based on the bundle's IP address.
    Location *geoLocation `json:"location,omitempty"`

    // 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"`

    // Attempts contains the failed connection attempts of the
    // peers belonging to a given IP address in chronological order.
    Attempts []*peerAttempt `json:"attempts,omitempty"`
}

// removeKnown removes the known peer belonging to the
// given IP address and node ID from the peer tree.
func (pc *peerContainer) removeKnown(ip, id string) (events []*peerEvent) {
    // TODO (kurkomisi): Remove peers that don't have traffic samples anymore.
    if bundle, ok := pc.Bundles[ip]; ok {
        if _, ok := bundle.KnownPeers[id]; ok {
            events = append(events, &peerEvent{
                Remove: RemoveKnown,
                IP:     ip,
                ID:     id,
            })
            delete(bundle.KnownPeers, id)
        } else {
            log.Warn("No peer to remove", "ip", ip, "id", id)
        }
        if len(bundle.KnownPeers) < 1 && len(bundle.Attempts) < 1 {
            events = append(events, &peerEvent{
                Remove: RemoveBundle,
                IP:     ip,
            })
            delete(pc.Bundles, ip)
        }
    } else {
        log.Warn("No bundle to remove", "ip", ip)
    }
    return events
}

// removeAttempt removes the peer attempt belonging to the
// given IP address and node ID from the peer tree.
func (pc *peerContainer) removeAttempt(ip string) (events []*peerEvent) {
    if bundle, ok := pc.Bundles[ip]; ok {
        if len(bundle.Attempts) > 0 {
            events = append(events, &peerEvent{
                Remove: RemoveAttempt,
                IP:     ip,
            })
            bundle.Attempts = bundle.Attempts[1:]
        }
        if len(bundle.Attempts) < 1 && len(bundle.KnownPeers) < 1 {
            events = append(events, &peerEvent{
                Remove: RemoveBundle,
                IP:     ip,
            })
            delete(pc.Bundles, ip)
        }
    }
    return events
}

// knownPeer inserts a new peer into the map, if the peer belonging
// to the given IP address and node ID wasn't metered so far. Returns the peer
// belonging to the given IP and ID as well as the events occurring during the
// initialization.
func (bundle *peerBundle) knownPeer(ip, id string) (*knownPeer, []*peerEvent) {
    var events []*peerEvent
    if _, ok := bundle.KnownPeers[id]; !ok {
        now := time.Now()
        ingress := emptyChartEntries(now, sampleLimit)
        egress := emptyChartEntries(now, sampleLimit)
        events = append(events, &peerEvent{
            IP:      ip,
            ID:      id,
            Ingress: append([]*ChartEntry{}, ingress...),
            Egress:  append([]*ChartEntry{}, egress...),
        })
        bundle.KnownPeers[id] = &knownPeer{
            ip:      ip,
            id:      id,
            Ingress: ingress,
            Egress:  egress,
        }
    }
    return bundle.KnownPeers[id], events
}

// knownPeer contains the metered data of a particular peer.
type knownPeer struct {
    // Connected contains the timestamps of the peer's connection events.
    Connected []*time.Time `json:"connected,omitempty"`

    // Disconnected contains the timestamps of the peer's disconnection events.
    Disconnected []*time.Time `json:"disconnected,omitempty"`

    // Ingress and Egress contain the peer's traffic samples, which are collected
    // periodically from the metrics registry.
    //
    // A peer can connect multiple times, and we want to visualize the time
    // passed between two connections, so after the first connection a 0 value
    // is appended to the traffic arrays even if the peer is inactive until the
    // peer is removed.
    Ingress ChartEntries `json:"ingress,omitempty"`
    Egress  ChartEntries `json:"egress,omitempty"`

    Active bool `json:"active"` // Denotes if the peer is still connected.

    listElement *list.Element // Pointer to the peer element in the list.
    ip, id      string        // The IP and the ID by which the peer can be accessed in the tree.
    prevIngress float64
    prevEgress  float64
}

// peerAttempt contains a failed peer connection attempt's attributes.
type peerAttempt struct {
    // Connected contains the timestamp of the connection attempt's moment.
    Connected time.Time `json:"connected"`

    // Disconnected contains the timestamp of the
    // moment when the connection attempt failed.
    Disconnected time.Time `json:"disconnected"`
}

type RemovedPeerType string
type ActivityType string

const (
    RemoveKnown   RemovedPeerType = "known"
    RemoveAttempt RemovedPeerType = "attempt"
    RemoveBundle  RemovedPeerType = "bundle"

    Active   ActivityType = "active"
    Inactive ActivityType = "inactive"
)

// peerEvent contains the attributes of a peer event.
type peerEvent struct {
    IP           string          `json:"ip,omitempty"`           // IP address of the peer.
    ID           string          `json:"id,omitempty"`           // Node ID of the peer.
    Remove       RemovedPeerType `json:"remove,omitempty"`       // Type of the peer that is to be removed.
    Location     *geoLocation    `json:"location,omitempty"`     // Geographical location of the peer.
    Connected    *time.Time      `json:"connected,omitempty"`    // Timestamp of the connection moment.
    Disconnected *time.Time      `json:"disconnected,omitempty"` // Timestamp of the disonnection moment.
    Ingress      ChartEntries    `json:"ingress,omitempty"`      // Ingress samples.
    Egress       ChartEntries    `json:"egress,omitempty"`       // Egress samples.
    Activity     ActivityType    `json:"activity,omitempty"`     // Connection status change.
}

// trafficMap is a container for the periodically collected peer traffic.
type trafficMap map[string]map[string]float64

// insert inserts a new value to the traffic map. Overwrites
// the value at the given ip and id if that already exists.
func (m *trafficMap) insert(ip, id string, val float64) {
    if _, ok := (*m)[ip]; !ok {
        (*m)[ip] = make(map[string]float64)
    }
    (*m)[ip][id] = val
}

// collectPeerData gathers data about the peers and sends it to the clients.
func (db *Dashboard) collectPeerData() {
    defer db.wg.Done()

    // Open the geodb database for IP to geographical information conversions.
    var err error
    db.geodb, err = openGeoDB()
    if err != nil {
        log.Warn("Failed to open geodb", "err", err)
        return
    }
    defer db.geodb.close()

    peerCh := make(chan p2p.MeteredPeerEvent, eventBufferLimit) // Peer event channel.
    subPeer := p2p.SubscribeMeteredPeerEvent(peerCh)            // Subscribe to peer events.
    defer subPeer.Unsubscribe()                                 // Unsubscribe at the end.

    ticker := time.NewTicker(db.config.Refresh)
    defer ticker.Stop()

    type registryFunc func(name string, i interface{})
    type collectorFunc func(traffic *trafficMap) registryFunc

    // trafficCollector generates a function that can be passed to
    // the prefixed peer registry in order to collect the metered
    // traffic data from each peer meter.
    trafficCollector := func(prefix string) collectorFunc {
        // This part makes is possible to collect the
        // traffic data into a map from outside.
        return func(traffic *trafficMap) registryFunc {
            // The function which can be passed to the registry.
            return func(name string, i interface{}) {
                if m, ok := i.(metrics.Meter); ok {
                    // The name of the meter has the format: <common traffic prefix><IP>/<ID>
                    if k := strings.Split(strings.TrimPrefix(name, prefix), "/"); len(k) == 2 {
                        traffic.insert(k[0], k[1], float64(m.Count()))
                    } else {
                        log.Warn("Invalid meter name", "name", name, "prefix", prefix)
                    }
                } else {
                    log.Warn("Invalid meter type", "name", name)
                }
            }
        }
    }
    collectIngress := trafficCollector(p2p.MetricsInboundTraffic + "/")
    collectEgress := trafficCollector(p2p.MetricsOutboundTraffic + "/")

    peers := newPeerContainer(db.geodb)
    db.peerLock.Lock()
    db.history.Network = &NetworkMessage{
        Peers: peers,
    }
    db.peerLock.Unlock()

    // newPeerEvents contains peer events, which trigger operations that
    // will be executed on the peer tree after a metering period.
    newPeerEvents := make([]*peerEvent, 0, eventLimit)
    ingress, egress := new(trafficMap), new(trafficMap)
    *ingress, *egress = make(trafficMap), make(trafficMap)

    for {
        select {
        case event := <-peerCh:
            now := time.Now()
            switch event.Type {
            case p2p.PeerConnected:
                connected := now.Add(-event.Elapsed)
                newPeerEvents = append(newPeerEvents, &peerEvent{
                    IP:        event.IP.String(),
                    ID:        event.ID.String(),
                    Connected: &connected,
                })
            case p2p.PeerDisconnected:
                ip, id := event.IP.String(), event.ID.String()
                newPeerEvents = append(newPeerEvents, &peerEvent{
                    IP:           ip,
                    ID:           id,
                    Disconnected: &now,
                })
                // The disconnect event comes with the last metered traffic count,
                // because after the disconnection the peer's meter is removed
                // from the registry. It can happen, that between two metering
                // period the same peer disconnects multiple times, and appending
                // all the samples to the traffic arrays would shift the metering,
                // so only the last metering is stored, overwriting the previous one.
                ingress.insert(ip, id, float64(event.Ingress))
                egress.insert(ip, id, float64(event.Egress))
            case p2p.PeerHandshakeFailed:
                connected := now.Add(-event.Elapsed)
                newPeerEvents = append(newPeerEvents, &peerEvent{
                    IP:           event.IP.String(),
                    Connected:    &connected,
                    Disconnected: &now,
                })
            default:
                log.Error("Unknown metered peer event type", "type", event.Type)
            }
        case <-ticker.C:
            // Collect the traffic samples from the registry.
            p2p.PeerIngressRegistry.Each(collectIngress(ingress))
            p2p.PeerEgressRegistry.Each(collectEgress(egress))

            // Protect 'peers', because it is part of the history.
            db.peerLock.Lock()

            var diff []*peerEvent
            for i := 0; i < len(newPeerEvents); i++ {
                if newPeerEvents[i].IP == "" {
                    log.Warn("Peer event without IP", "event", *newPeerEvents[i])
                    continue
                }
                diff = append(diff, newPeerEvents[i])
                // There are two main branches of peer events coming from the event
                // feed, one belongs to the known peers, one to the unknown peers.
                // If the event has node ID, it belongs to a known peer, otherwise
                // to an unknown one, which is considered as connection attempt.
                //
                // The extension can produce additional peer events, such
                // as remove, location and initial samples events.
                if newPeerEvents[i].ID == "" {
                    diff = append(diff, peers.handleAttempt(newPeerEvents[i])...)
                    continue
                }
                diff = append(diff, peers.extendKnown(newPeerEvents[i])...)
            }
            // Update the peer tree using the traffic maps.
            for ip, bundle := range peers.Bundles {
                for id, peer := range bundle.KnownPeers {
                    // Value is 0 if the traffic map doesn't have the
                    // entry corresponding to the given IP and ID.
                    curIngress, curEgress := (*ingress)[ip][id], (*egress)[ip][id]
                    deltaIngress, deltaEgress := curIngress, curEgress
                    if deltaIngress >= peer.prevIngress {
                        deltaIngress -= peer.prevIngress
                    }
                    if deltaEgress >= peer.prevEgress {
                        deltaEgress -= peer.prevEgress
                    }
                    peer.prevIngress, peer.prevEgress = curIngress, curEgress
                    i := &ChartEntry{
                        Value: deltaIngress,
                    }
                    e := &ChartEntry{
                        Value: deltaEgress,
                    }
                    peer.Ingress = append(peer.Ingress, i)
                    peer.Egress = append(peer.Egress, e)
                    if first := len(peer.Ingress) - sampleLimit; first > 0 {
                        peer.Ingress = peer.Ingress[first:]
                    }
                    if first := len(peer.Egress) - sampleLimit; first > 0 {
                        peer.Egress = peer.Egress[first:]
                    }
                    // Creating the traffic sample events.
                    diff = append(diff, &peerEvent{
                        IP:      ip,
                        ID:      id,
                        Ingress: ChartEntries{i},
                        Egress:  ChartEntries{e},
                    })
                }
            }
            db.peerLock.Unlock()

            if len(diff) > 0 {
                db.sendToAll(&Message{Network: &NetworkMessage{
                    Diff: diff,
                }})
            }
            // Clear the traffic maps, and the event array,
            // prepare them for the next metering.
            *ingress, *egress = make(trafficMap), make(trafficMap)
            newPeerEvents = newPeerEvents[:0]
        case err := <-subPeer.Err():
            log.Warn("Peer subscription error", "err", err)
            return
        case errc := <-db.quit:
            errc <- nil
            return
        }
    }
}