809b7fbda5
Fixes #1554 Fixes: peer names unset for remote peers This adds an IPFS field to pin status information (PinInfoShort). It has not been easy to add this, given that the IPFS ID is something that comes from outside of cluster (unlike the peer name). After several tries I have settled in the following things: - Use the ping metric to send out peer names and IPFS IDs to the peers in the cluster. - Cache the latest known IPFS ID (if IPFS dies we should still be setting the ID). - Provide an RPC method for the Pintracker to obtain IPFS ID from the cache. - Given we now know information for peernames and IPFS IDs from other peers, we can use that information even if the requests to them error or we are not contacting (i.e. peers allocated as remote are not queried for status). We can use the information from the last received ping metric. - This means we should keep metrics around even if peers go away, at least for a while rather than deleting them as soon as we detect that they have expired. Puting it all together we now have a system to gossip peer information around on top of the ping metrics.
183 lines
4.2 KiB
Go
183 lines
4.2 KiB
Go
package ipfscluster
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
|
|
blake2b "golang.org/x/crypto/blake2b"
|
|
|
|
cid "github.com/ipfs/go-cid"
|
|
"github.com/ipfs/ipfs-cluster/api"
|
|
peer "github.com/libp2p/go-libp2p-core/peer"
|
|
ma "github.com/multiformats/go-multiaddr"
|
|
)
|
|
|
|
// PeersFromMultiaddrs returns all the different peers in the given addresses.
|
|
// each peer only will appear once in the result, even if several
|
|
// multiaddresses for it are provided.
|
|
func PeersFromMultiaddrs(addrs []ma.Multiaddr) []peer.ID {
|
|
var pids []peer.ID
|
|
pm := make(map[peer.ID]struct{})
|
|
for _, addr := range addrs {
|
|
pinfo, err := peer.AddrInfoFromP2pAddr(addr)
|
|
if err != nil {
|
|
continue
|
|
}
|
|
_, ok := pm[pinfo.ID]
|
|
if !ok {
|
|
pm[pinfo.ID] = struct{}{}
|
|
pids = append(pids, pinfo.ID)
|
|
}
|
|
}
|
|
return pids
|
|
}
|
|
|
|
// // connect to a peer ID.
|
|
// func connectToPeer(ctx context.Context, h host.Host, id peer.ID, addr ma.Multiaddr) error {
|
|
// err := h.Connect(ctx, peerstore.PeerInfo{
|
|
// ID: id,
|
|
// Addrs: []ma.Multiaddr{addr},
|
|
// })
|
|
// return err
|
|
// }
|
|
|
|
// // return the local multiaddresses used to communicate to a peer.
|
|
// func localMultiaddrsTo(h host.Host, pid peer.ID) []ma.Multiaddr {
|
|
// var addrs []ma.Multiaddr
|
|
// conns := h.Network().ConnsToPeer(pid)
|
|
// logger.Debugf("conns to %s are: %s", pid, conns)
|
|
// for _, conn := range conns {
|
|
// addrs = append(addrs, multiaddrJoin(conn.LocalMultiaddr(), h.ID()))
|
|
// }
|
|
// return addrs
|
|
// }
|
|
|
|
func logError(fmtstr string, args ...interface{}) error {
|
|
msg := fmt.Sprintf(fmtstr, args...)
|
|
logger.Error(msg)
|
|
return errors.New(msg)
|
|
}
|
|
|
|
func containsPeer(list []peer.ID, peer peer.ID) bool {
|
|
for _, p := range list {
|
|
if p == peer {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func minInt(x, y int) int {
|
|
if x < y {
|
|
return x
|
|
}
|
|
return y
|
|
}
|
|
|
|
// // updatePinParents modifies the api.Pin input to give it the correct parents
|
|
// // so that previous additions to the pins parents are maintained after this
|
|
// // pin is committed to consensus. If this pin carries new parents they are
|
|
// // merged with those already existing for this CID.
|
|
// func updatePinParents(pin *api.Pin, existing *api.Pin) {
|
|
// // no existing parents this pin is up to date
|
|
// if existing.Parents == nil || len(existing.Parents.Keys()) == 0 {
|
|
// return
|
|
// }
|
|
// for _, c := range existing.Parents.Keys() {
|
|
// pin.Parents.Add(c)
|
|
// }
|
|
// }
|
|
|
|
type distance [blake2b.Size256]byte
|
|
|
|
type distanceChecker struct {
|
|
local peer.ID
|
|
otherPeers []peer.ID
|
|
cache map[peer.ID]distance
|
|
}
|
|
|
|
func (dc distanceChecker) isClosest(ci cid.Cid) bool {
|
|
ciHash := convertKey(ci.KeyString())
|
|
localPeerHash := dc.convertPeerID(dc.local)
|
|
myDistance := xor(ciHash, localPeerHash)
|
|
|
|
for _, p := range dc.otherPeers {
|
|
peerHash := dc.convertPeerID(p)
|
|
distance := xor(peerHash, ciHash)
|
|
|
|
// if myDistance is larger than for other peers...
|
|
if bytes.Compare(myDistance[:], distance[:]) > 0 {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// convertPeerID hashes a Peer ID (Multihash).
|
|
func (dc distanceChecker) convertPeerID(id peer.ID) distance {
|
|
hash, ok := dc.cache[id]
|
|
if ok {
|
|
return hash
|
|
}
|
|
|
|
hashBytes := convertKey(string(id))
|
|
dc.cache[id] = hashBytes
|
|
return hashBytes
|
|
}
|
|
|
|
// convertKey hashes a key.
|
|
func convertKey(id string) distance {
|
|
return blake2b.Sum256([]byte(id))
|
|
}
|
|
|
|
func xor(a, b distance) distance {
|
|
var c distance
|
|
for i := 0; i < len(c); i++ {
|
|
c[i] = a[i] ^ b[i]
|
|
}
|
|
return c
|
|
}
|
|
|
|
// peersSubtract subtracts peers ID slice b from peers ID slice a.
|
|
func peersSubtract(a []peer.ID, b []peer.ID) []peer.ID {
|
|
var result []peer.ID
|
|
bMap := make(map[peer.ID]struct{}, len(b))
|
|
|
|
for _, p := range b {
|
|
bMap[p] = struct{}{}
|
|
}
|
|
|
|
for _, p := range a {
|
|
_, ok := bMap[p]
|
|
if ok {
|
|
continue
|
|
}
|
|
result = append(result, p)
|
|
}
|
|
|
|
return result
|
|
}
|
|
|
|
// pingValue describes the value carried by ping metrics
|
|
type pingValue struct {
|
|
Peername string `json:"peer_name,omitempty"`
|
|
IPFSID peer.ID `json:"ipfs_id,omitempty"`
|
|
}
|
|
|
|
// Valid returns true if the PingValue has IPFSID set.
|
|
func (pv pingValue) Valid() bool {
|
|
return pv.IPFSID != ""
|
|
}
|
|
|
|
// PingValue from metric parses a ping value from the value of a given metric,
|
|
// if possible.
|
|
func pingValueFromMetric(m *api.Metric) (pv pingValue) {
|
|
if m == nil {
|
|
return
|
|
}
|
|
json.Unmarshal([]byte(m.Value), &pv)
|
|
return
|
|
}
|