Discovery service (#805)

* WIP possible discovery service impl

* Adds discovery service to CaptPlanet

* Updates the config and server for discovery service

* updates testplanet to use discovery package

* update satellite imports

* Removes unnecessary cache test

* linter fixes

* adds discovery startup to captplanet

* invoke refresh

* updates to discovery refresh cycle

* Make implementation more consistent with previous implementation

* add wait before trying to upload

* sleep a bit more

* remove kademlia bootstrap

* updates

* remove comments

cmd/captplanet/run.go

@@ -20,6 +20,7 @@ import (
 	"storj.io/storj/pkg/cfgstruct"
 	"storj.io/storj/pkg/datarepair/checker"
 	"storj.io/storj/pkg/datarepair/repairer"
+	"storj.io/storj/pkg/discovery"
 	"storj.io/storj/pkg/inspector"
 	"storj.io/storj/pkg/kademlia"
 	"storj.io/storj/pkg/miniogw"
@@ -52,6 +53,7 @@ type Satellite struct {
 	BwAgreement bwagreement.Config
 	Web         satelliteweb.Config
 	Database    string `help:"satellite database connection string" default:"sqlite3://$CONFDIR/master.db"`
+	Discovery   discovery.Config
 	Tally       tally.Config
 	Rollup      rollup.Config
 }
@@ -132,6 +134,7 @@ func cmdRun(cmd *cobra.Command, args []string) (err error) {
 			runCfg.Satellite.StatDB,
 			runCfg.Satellite.Audit,
 			runCfg.Satellite.Overlay,
+			runCfg.Satellite.Discovery,
 			runCfg.Satellite.PointerDB,
 			runCfg.Satellite.Checker,
 			runCfg.Satellite.Repairer,

cmd/satellite/main.go

@@ -23,6 +23,7 @@ import (
 	"storj.io/storj/pkg/datarepair/checker"
 	"storj.io/storj/pkg/datarepair/queue"
 	"storj.io/storj/pkg/datarepair/repairer"
+	"storj.io/storj/pkg/discovery"
 	"storj.io/storj/pkg/kademlia"
 	"storj.io/storj/pkg/overlay"
 	"storj.io/storj/pkg/pb"
@@ -72,6 +73,7 @@ var (
 		Audit       audit.Config
 		BwAgreement bwagreement.Config
 		Database    string `help:"satellite database connection string" default:"sqlite3://$CONFDIR/master.db"`
+		Discovery   discovery.Config
 	}
 	setupCfg struct {
 		BasePath  string `default:"$CONFDIR" help:"base path for setup"`
@@ -129,6 +131,7 @@ func cmdRun(cmd *cobra.Command, args []string) (err error) {
 		runCfg.Repairer,
 		runCfg.Audit,
 		runCfg.BwAgreement,
+		runCfg.Discovery,
 	)
 }
 

internal/testplanet/node.go

@@ -14,6 +14,7 @@ import (
 	"google.golang.org/grpc"
 
 	"storj.io/storj/pkg/auth/grpcauth"
+	"storj.io/storj/pkg/discovery"
 	"storj.io/storj/pkg/kademlia"
 	"storj.io/storj/pkg/node"
 	"storj.io/storj/pkg/overlay"
@@ -36,6 +37,7 @@ type Node struct {
 	Listener  net.Listener
 	Provider  *provider.Provider
 	Kademlia  *kademlia.Kademlia
+	Discovery *discovery.Discovery
 	StatDB    *statdb.StatDB
 	Overlay   *overlay.Cache
 
@@ -172,6 +174,7 @@ func (node *Node) initOverlay(planet *Planet) error {
 	node.StatDB = sdb
 
 	node.Overlay = overlay.NewOverlayCache(teststore.New(), node.Kademlia, node.StatDB)
+	node.Discovery = discovery.NewDiscovery(node.Overlay, node.Kademlia, node.StatDB)
 
 	return nil
 }

internal/testplanet/planet.go

@@ -128,7 +128,7 @@ func New(t zaptest.TestingT, satelliteCount, storageNodeCount, uplinkCount int)
 			t := time.NewTicker(500 * time.Millisecond).C
 			for {
 				<-t
-				if err := n.Overlay.Refresh(context.Background()); err != nil {
+				if err := n.Discovery.Refresh(context.Background()); err != nil {
 					log.Error(err.Error())
 				}
 			}

pkg/discovery/config.go

@@ -0,0 +1,72 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package discovery
+
+import (
+	"context"
+	"time"
+
+	"github.com/zeebo/errs"
+	"go.uber.org/zap"
+	monkit "gopkg.in/spacemonkeygo/monkit.v2"
+
+	"storj.io/storj/pkg/kademlia"
+	"storj.io/storj/pkg/overlay"
+	"storj.io/storj/pkg/pb"
+	"storj.io/storj/pkg/provider"
+	"storj.io/storj/pkg/statdb"
+)
+
+var (
+	mon = monkit.Package()
+	// Error represents an overlay error
+	Error = errs.Class("discovery error")
+)
+
+// CtxKey used for assigning a key to Discovery server
+type CtxKey int
+
+const (
+	ctxKeyDiscovery CtxKey = iota
+)
+
+// Config loads on the configuration values from run flags
+type Config struct {
+	RefreshInterval time.Duration `help:"the interval at which the cache refreshes itself in seconds" default:"1s"`
+}
+
+// Run runs the Discovery boot up and initialization
+func (c Config) Run(ctx context.Context, server *provider.Provider) (err error) {
+	defer mon.Task()(&ctx)(&err)
+
+	srv := NewServer(zap.L())
+	pb.RegisterDiscoveryServer(server.GRPC(), srv)
+
+	ol := overlay.LoadFromContext(ctx)
+	kad := kademlia.LoadFromContext(ctx)
+	stat := statdb.LoadFromContext(ctx)
+	discovery := NewDiscovery(ol, kad, stat)
+
+	zap.L().Debug("Starting discovery")
+
+	ticker := time.NewTicker(c.RefreshInterval)
+	defer ticker.Stop()
+
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				zap.L().Debug("Kicking off refresh")
+				err := discovery.Refresh(ctx)
+				if err != nil {
+					zap.L().Error("Error with cache refresh: ", zap.Error(err))
+				}
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return server.Run(context.WithValue(ctx, ctxKeyDiscovery, discovery))
+}

pkg/discovery/server.go

@@ -0,0 +1,23 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package discovery
+
+import (
+	"go.uber.org/zap"
+	"gopkg.in/spacemonkeygo/monkit.v2"
+)
+
+// Server struct that loads on logging and metrics
+type Server struct {
+	log     *zap.Logger
+	metrics *monkit.Registry
+}
+
+// NewServer returns a server
+func NewServer(l *zap.Logger) *Server {
+	return &Server{
+		log:     l,
+		metrics: monkit.Default,
+	}
+}

pkg/discovery/service.go

@@ -0,0 +1,57 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package discovery
+
+import (
+	"context"
+
+	"storj.io/storj/pkg/kademlia"
+	"storj.io/storj/pkg/overlay"
+	"storj.io/storj/pkg/statdb"
+)
+
+// Discovery struct loads on cache, kad, and statdb
+type Discovery struct {
+	cache  *overlay.Cache
+	kad    *kademlia.Kademlia
+	statdb *statdb.StatDB
+}
+
+// NewDiscovery Returns a new Discovery instance with cache, kad, and statdb loaded on
+func NewDiscovery(ol *overlay.Cache, kad *kademlia.Kademlia, stat *statdb.StatDB) *Discovery {
+	return &Discovery{
+		cache:  ol,
+		kad:    kad,
+		statdb: stat,
+	}
+}
+
+// Refresh updates the cache db with the current DHT.
+// We currently do not penalize nodes that are unresponsive,
+// but should in the future.
+func (d *Discovery) Refresh(ctx context.Context) error {
+	// TODO(coyle): make refresh work by looking on the network for new ndoes
+	nodes := d.kad.Seen()
+
+	for _, v := range nodes {
+		if err := d.cache.Put(ctx, v.Id, *v); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+// Bootstrap walks the initialized network and populates the cache
+func (d *Discovery) Bootstrap(ctx context.Context) error {
+	// o := overlay.LoadFromContext(ctx)
+	// kad := kademlia.LoadFromContext(ctx)
+	// TODO(coyle): make Bootstrap work
+	// look in our routing table
+	// get every node we know about
+	// ask every node for every node they know about
+	// for each newly known node, ask those nodes for every node they know about
+	// continue until no new nodes are found
+	return nil
+}

pkg/discovery/service_test.go

@@ -0,0 +1,29 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+package discovery_test
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"storj.io/storj/internal/testcontext"
+	"storj.io/storj/internal/testplanet"
+)
+
+func TestCache_Refresh(t *testing.T) {
+	ctx := testcontext.New(t)
+	defer ctx.Cleanup()
+
+	planet, err := testplanet.New(t, 1, 30, 0)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ctx.Check(planet.Shutdown)
+
+	planet.Start(ctx)
+
+	err = planet.Satellites[0].Discovery.Refresh(ctx)
+	assert.NoError(t, err)
+}

pkg/overlay/cache.go

@@ -121,36 +121,3 @@ func (o *Cache) Put(ctx context.Context, nodeID storj.NodeID, value pb.Node) err
 
 	return o.DB.Put(nodeID.Bytes(), data)
 }
-
-// Bootstrap walks the initialized network and populates the cache
-func (o *Cache) Bootstrap(ctx context.Context) error {
-	// TODO(coyle): make Bootstrap work
-	// look in our routing table
-	// get every node we know about
-	// ask every node for every node they know about
-	// for each newly known node, ask those nodes for every node they know about
-	// continue until no new nodes are found
-	return nil
-}
-
-// 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 {
-	// TODO(coyle): make refresh work by looking on the network for new ndoes
-	nodes := o.DHT.Seen()
-
-	for _, v := range nodes {
-		if err := o.Put(ctx, v.Id, *v); err != nil {
-			return err
-		}
-	}
-
-	return nil
-}
-
-// 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
-}

pkg/overlay/cache_test.go

@@ -149,19 +149,3 @@ func TestCache_Store(t *testing.T) {
 
 	testCache(ctx, t, teststore.New(), sdb)
 }
-
-func TestCache_Refresh(t *testing.T) {
-	ctx := testcontext.New(t)
-	defer ctx.Cleanup()
-
-	planet, err := testplanet.New(t, 1, 30, 0)
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer ctx.Check(planet.Shutdown)
-
-	planet.Start(ctx)
-
-	err = planet.Satellites[0].Overlay.Refresh(ctx)
-	assert.NoError(t, err)
-}

pkg/overlay/config.go

@@ -91,31 +91,6 @@ func (c Config) Run(ctx context.Context, server *provider.Provider) (
 	}
 
 	cache := NewOverlayCache(db, kad, sdb)
-
-	go func() {
-		err = cache.Bootstrap(ctx)
-		if err != nil {
-			panic(err)
-		}
-	}()
-
-	ticker := time.NewTicker(c.RefreshInterval)
-	defer ticker.Stop()
-
-	go func() {
-		for {
-			select {
-			case <-ticker.C:
-				err := cache.Refresh(ctx)
-				if err != nil {
-					zap.L().Error("Error with cache refresh: ", zap.Error(err))
-				}
-			case <-ctx.Done():
-				return
-			}
-		}
-	}()
-
 	srv := NewServer(zap.L(), cache, kad)
 	pb.RegisterOverlayServer(server.GRPC(), srv)
 

pkg/pb/discovery.pb.go

@@ -0,0 +1,74 @@
+// Code generated by protoc-gen-gogo. DO NOT EDIT.
+// source: discovery.proto
+
+package pb
+
+import proto "github.com/gogo/protobuf/proto"
+import fmt "fmt"
+import math "math"
+import _ "github.com/gogo/protobuf/gogoproto"
+
+import (
+	context "golang.org/x/net/context"
+	grpc "google.golang.org/grpc"
+)
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ = proto.Marshal
+var _ = fmt.Errorf
+var _ = math.Inf
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the proto package it is being compiled against.
+// A compilation error at this line likely means your copy of the
+// proto package needs to be updated.
+const _ = proto.GoGoProtoPackageIsVersion2 // please upgrade the proto package
+
+// Reference imports to suppress errors if they are not otherwise used.
+var _ context.Context
+var _ grpc.ClientConn
+
+// This is a compile-time assertion to ensure that this generated file
+// is compatible with the grpc package it is being compiled against.
+const _ = grpc.SupportPackageIsVersion4
+
+// DiscoveryClient is the client API for Discovery service.
+//
+// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
+type DiscoveryClient interface {
+}
+
+type discoveryClient struct {
+	cc *grpc.ClientConn
+}
+
+func NewDiscoveryClient(cc *grpc.ClientConn) DiscoveryClient {
+	return &discoveryClient{cc}
+}
+
+// DiscoveryServer is the server API for Discovery service.
+type DiscoveryServer interface {
+}
+
+func RegisterDiscoveryServer(s *grpc.Server, srv DiscoveryServer) {
+	s.RegisterService(&_Discovery_serviceDesc, srv)
+}
+
+var _Discovery_serviceDesc = grpc.ServiceDesc{
+	ServiceName: "discovery.Discovery",
+	HandlerType: (*DiscoveryServer)(nil),
+	Methods:     []grpc.MethodDesc{},
+	Streams:     []grpc.StreamDesc{},
+	Metadata:    "discovery.proto",
+}
+
+func init() { proto.RegisterFile("discovery.proto", fileDescriptor_discovery_38baf63dfe55b3f3) }
+
+var fileDescriptor_discovery_38baf63dfe55b3f3 = []byte{
+	// 71 bytes of a gzipped FileDescriptorProto
+	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xe2, 0xe2, 0x4f, 0xc9, 0x2c, 0x4e,
+	0xce, 0x2f, 0x4b, 0x2d, 0xaa, 0xd4, 0x2b, 0x28, 0xca, 0x2f, 0xc9, 0x17, 0xe2, 0x84, 0x0b, 0x48,
+	0x71, 0xa5, 0xe7, 0xa7, 0xe7, 0x43, 0x84, 0x8d, 0xb8, 0xb9, 0x38, 0x5d, 0x60, 0x12, 0x4e, 0x2c,
+	0x51, 0x4c, 0x05, 0x49, 0x49, 0x6c, 0x60, 0x19, 0x63, 0x40, 0x00, 0x00, 0x00, 0xff, 0xff, 0x92,
+	0xc2, 0xf4, 0x67, 0x43, 0x00, 0x00, 0x00,
+}

pkg/pb/discovery.proto

@@ -0,0 +1,11 @@
+// Copyright (C) 2018 Storj Labs, Inc.
+// See LICENSE for copying information.
+
+syntax = "proto3";
+option go_package = "pb";
+
+import "gogo.proto";
+
+package discovery;
+
+service Discovery {}

scripts/test-captplanet.sh

@@ -20,8 +20,8 @@ kill -9 $CAPT_PID
 captplanet run &
 CAPT_PID=$!
 
-# Wait 2 seconds for kademlia startup
-sleep 2
+# Wait 5 seconds for kademlia startup
+sleep 5
 
 #setup tmpdir for testfiles and cleanup
 TMP_DIR=$(mktemp -d -t tmp.XXXXXXXXXX)
@@ -94,8 +94,8 @@ fi
 captplanet run &
 CAPT_PID=$!
 
-# Wait 2 seconds for kademlia startup
-sleep 2
+# Wait 5 seconds for kademlia startup
+sleep 5
 
 aws s3 --endpoint=http://localhost:7777/ mb s3://bucket
 aws s3 --endpoint=http://localhost:7777/ cp $TMP_DIR/big-upload-testfile s3://bucket/big-testfile