peers.go

// Copyright 2019 The Elastos.ELA.SideChain.ESC Authors
// This file is part of the Elastos.ELA.SideChain.ESC library.
//
// The Elastos.ELA.SideChain.ESC 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 Elastos.ELA.SideChain.ESC 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 Elastos.ELA.SideChain.ESC library. If not, see <http://www.gnu.org/licenses/>.

package dashboard

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

	"github.com/elastos/Elastos.ELA.SideChain.ESC/metrics"

	"github.com/elastos/Elastos.ELA.SideChain.ESC/log"
	"github.com/elastos/Elastos.ELA.SideChain.ESC/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
		}
	}
}