fe3decc42f
Fixes go1.11 vet warnings. Cancel on WithTimeout must always be called to avoid memory leak: pkg/provider/provider.go:73: the cancel function returned by context.WithTimeout should be called, not discarded, to avoid a context leak Range over non-copyable things: pkg/pool/connection_pool_test.go:32: range var v copies lock: struct{pool pool.ConnectionPool; key string; expected pool.TestFoo; expectedError error} contains pool.ConnectionPool contains sync.RWMutex pkg/pool/connection_pool_test.go:56: range var v copies lock: struct{pool pool.ConnectionPool; key string; value pool.TestFoo; expected pool.TestFoo; expectedError error} contains pool.ConnectionPool contains sync.RWMutex pkg/pool/connection_pool_test.go:83: range var v copies lock: struct{pool pool.ConnectionPool; key string; value pool.TestFoo; expected interface{}; expectedError error} contains pool.ConnectionPool contains sync.RWMutex zeebo/errs package always requires formatting directives: pkg/peertls/peertls.go:50: Class.New call has arguments but no formatting directives pkg/peertls/utils.go:47: Class.New call has arguments but no formatting directives pkg/peertls/utils.go:87: Class.New call has arguments but no formatting directives pkg/overlay/cache.go:94: Class.New call has arguments but no formatting directives pkg/provider/certificate_authority.go:98: New call has arguments but no formatting directives pkg/provider/identity.go:96: New call has arguments but no formatting directives pkg/provider/utils.go:124: New call needs 1 arg but has 2 args pkg/provider/utils.go:136: New call needs 1 arg but has 2 args storage/redis/client.go:44: Class.New call has arguments but no formatting directives storage/redis/client.go:64: Class.New call has arguments but no formatting directives storage/redis/client.go:75: Class.New call has arguments but no formatting directives storage/redis/client.go:80: Class.New call has arguments but no formatting directives storage/redis/client.go:92: Class.New call has arguments but no formatting directives storage/redis/client.go:96: Class.New call has arguments but no formatting directives storage/redis/client.go:102: Class.New call has arguments but no formatting directives storage/redis/client.go:126: Class.New call has arguments but no formatting directives
176 lines
3.9 KiB
Go
176 lines
3.9 KiB
Go
// Copyright (C) 2018 Storj Labs, Inc.
|
|
// See LICENSE for copying information.
|
|
|
|
package overlay
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"log"
|
|
|
|
"github.com/gogo/protobuf/proto"
|
|
"github.com/zeebo/errs"
|
|
"go.uber.org/zap"
|
|
|
|
"storj.io/storj/pkg/dht"
|
|
"storj.io/storj/pkg/kademlia"
|
|
"storj.io/storj/protos/overlay"
|
|
"storj.io/storj/storage"
|
|
"storj.io/storj/storage/boltdb"
|
|
"storj.io/storj/storage/redis"
|
|
)
|
|
|
|
// ErrNodeNotFound error standardization
|
|
var ErrNodeNotFound = errs.New("Node not found")
|
|
|
|
// OverlayError creates class of errors for stack traces
|
|
var OverlayError = errs.Class("Overlay Error")
|
|
|
|
// Cache is used to store overlay data in Redis
|
|
type Cache struct {
|
|
DB storage.KeyValueStore
|
|
DHT dht.DHT
|
|
}
|
|
|
|
// NewRedisOverlayCache returns a pointer to a new Cache instance with an initalized connection to Redis.
|
|
func NewRedisOverlayCache(address, password string, db int, DHT dht.DHT) (*Cache, error) {
|
|
rc, err := redis.NewClient(address, password, db)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &Cache{
|
|
DB: rc,
|
|
DHT: DHT,
|
|
}, nil
|
|
}
|
|
|
|
// NewBoltOverlayCache returns a pointer to a new Cache instance with an initalized connection to a Bolt db.
|
|
func NewBoltOverlayCache(dbPath string, DHT dht.DHT) (*Cache, error) {
|
|
bc, err := boltdb.NewClient(zap.L(), dbPath, boltdb.OverlayBucket)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return &Cache{
|
|
DB: bc,
|
|
DHT: DHT,
|
|
}, nil
|
|
}
|
|
|
|
// Get looks up the provided nodeID from the redis cache
|
|
func (o *Cache) Get(ctx context.Context, key string) (*overlay.Node, error) {
|
|
b, err := o.DB.Get([]byte(key))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if b.IsZero() {
|
|
// TODO: log? return an error?
|
|
return nil, nil
|
|
}
|
|
|
|
na := &overlay.Node{}
|
|
if err := proto.Unmarshal(b, na); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return na, nil
|
|
}
|
|
|
|
// Put adds a nodeID to the redis cache with a binary representation of proto defined Node
|
|
func (o *Cache) Put(nodeID string, value overlay.Node) error {
|
|
data, err := proto.Marshal(&value)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
return o.DB.Put(kademlia.StringToNodeID(nodeID).Bytes(), []byte(data))
|
|
}
|
|
|
|
// Bootstrap walks the initialized network and populates the cache
|
|
func (o *Cache) Bootstrap(ctx context.Context) error {
|
|
nodes, err := o.DHT.GetNodes(ctx, "", 1280)
|
|
if err != nil {
|
|
zap.Error(OverlayError.New("Error getting nodes from DHT: %v", err))
|
|
}
|
|
|
|
for _, v := range nodes {
|
|
found, err := o.DHT.FindNode(ctx, kademlia.StringToNodeID(v.Id))
|
|
if err != nil {
|
|
zap.Error(ErrNodeNotFound)
|
|
}
|
|
|
|
node, err := proto.Marshal(&found)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if err := o.DB.Put(kademlia.StringToNodeID(found.Id).Bytes(), node); err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
// Refresh updates the cache db with the current DHT.
|
|
// We currently do not penalize nodes that are unresponsive,
|
|
// but should in the future.
|
|
func (o *Cache) Refresh(ctx context.Context) error {
|
|
log.Print("starting cache refresh")
|
|
r, err := randomID()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
rid := kademlia.NodeID(r)
|
|
near, err := o.DHT.GetNodes(ctx, rid.String(), 128)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
for _, node := range near {
|
|
pinged, err := o.DHT.Ping(ctx, *node)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
err = o.DB.Put([]byte(pinged.Id), []byte(pinged.Address.Address))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
// TODO: Kademlia hooks to do this automatically rather than at interval
|
|
nodes, err := o.DHT.GetNodes(ctx, "", 128)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
for _, node := range nodes {
|
|
pinged, err := o.DHT.Ping(ctx, *node)
|
|
if err != nil {
|
|
zap.Error(ErrNodeNotFound)
|
|
return err
|
|
}
|
|
err = o.DB.Put([]byte(pinged.Id), []byte(pinged.Address.Address))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
}
|
|
|
|
return err
|
|
}
|
|
|
|
// Walk iterates over each node in each bucket to traverse the network
|
|
func (o *Cache) Walk(ctx context.Context) error {
|
|
// TODO: This should walk the cache, rather than be a duplicate of refresh
|
|
return nil
|
|
}
|
|
|
|
func randomID() ([]byte, error) {
|
|
result := make([]byte, 64)
|
|
_, err := rand.Read(result)
|
|
return result, err
|
|
}
|