80727ae90b
* adds netstate rpc server pagination, mocks pagination in test/util.go * updates ns client example, combines ns client and server test to netstate_test, adds pagination to bolt client * better organizes netstate test calls * wip breaking netstate test into smaller tests * wip modularizing netstate tests * adds some test panics * wip netstate test attempts * testing bug in netstate TestDeleteAuth * wip fixes global variable problem, still issues with list * wip fixes get request params and args * fixes bug in path when using MakePointers helper fn * updates mockdb list func, adds test, changes Limit to int * fixes merge conflicts * fixes broken tests from merge * remove unnecessary PointerEntry struct * removes error when Get returns nil value from boltdb * breaks boltdb client tests into smaller tests * renames AssertNoErr test helper to HandleErr * adds StartingKey and Limit parameters to redis list func, adds beginning of redis tests * adds helper func for mockdb List function * if no starting key provided for netstate List, the first value in storage will be used * adds basic pagination for redis List function, adds tests * adds list limit to call in overlay/server.go * streamlines/fixes some nits from review * removes use of obsolete EncryptedUnencryptedSize * uses MockKeyValueStore instead of redis instance in redis client test * changes test to expect nil returned for getting missing key * remove error from `KeyValueStore#Get` * fix bolt test * Merge pull request #1 from bryanchriswhite/nat-pagination remove error from `KeyValueStore#Get` * adds Get returning error back to KeyValueStore interface and affected clients * trying to appease travis: returns errors in Get calls in overlay/cache and cache_test * handles redis get error when no key found
149 lines
4.1 KiB
Go
149 lines
4.1 KiB
Go
// Copyright (C) 2018 Storj Labs, Inc.
|
|
// See LICENSE for copying information.
|
|
|
|
package netstate
|
|
|
|
import (
|
|
"context"
|
|
|
|
"github.com/golang/protobuf/proto"
|
|
"go.uber.org/zap"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/status"
|
|
|
|
"storj.io/storj/netstate/auth"
|
|
pb "storj.io/storj/protos/netstate"
|
|
"storj.io/storj/storage"
|
|
)
|
|
|
|
// Server implements the network state RPC service
|
|
type Server struct {
|
|
DB storage.KeyValueStore
|
|
logger *zap.Logger
|
|
}
|
|
|
|
// NewServer creates instance of Server
|
|
func NewServer(db storage.KeyValueStore, logger *zap.Logger) *Server {
|
|
return &Server{
|
|
DB: db,
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
func (s *Server) validateAuth(APIKeyBytes []byte) error {
|
|
if !auth.ValidateAPIKey(string(APIKeyBytes)) {
|
|
s.logger.Error("unauthorized request: ", zap.Error(grpc.Errorf(codes.Unauthenticated, "Invalid API credential")))
|
|
return grpc.Errorf(codes.Unauthenticated, "Invalid API credential")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// Put formats and hands off a key/value (path/pointer) to be saved to boltdb
|
|
func (s *Server) Put(ctx context.Context, putReq *pb.PutRequest) (*pb.PutResponse, error) {
|
|
s.logger.Debug("entering netstate put")
|
|
|
|
APIKeyBytes := []byte(putReq.APIKey)
|
|
if err := s.validateAuth(APIKeyBytes); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
pointerBytes, err := proto.Marshal(putReq.Pointer)
|
|
if err != nil {
|
|
s.logger.Error("err marshaling pointer", zap.Error(err))
|
|
return nil, status.Errorf(codes.Internal, err.Error())
|
|
}
|
|
|
|
if err := s.DB.Put(putReq.Path, pointerBytes); err != nil {
|
|
s.logger.Error("err putting pointer", zap.Error(err))
|
|
return nil, status.Errorf(codes.Internal, err.Error())
|
|
}
|
|
s.logger.Debug("put to the db: " + string(putReq.Path))
|
|
|
|
return &pb.PutResponse{}, nil
|
|
}
|
|
|
|
// Get formats and hands off a file path to get from boltdb
|
|
func (s *Server) Get(ctx context.Context, req *pb.GetRequest) (*pb.GetResponse, error) {
|
|
s.logger.Debug("entering netstate get")
|
|
|
|
APIKeyBytes := []byte(req.APIKey)
|
|
if err := s.validateAuth(APIKeyBytes); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
pointerBytes, err := s.DB.Get(req.Path)
|
|
if err != nil {
|
|
s.logger.Error("err getting file", zap.Error(err))
|
|
return nil, status.Errorf(codes.Internal, err.Error())
|
|
}
|
|
|
|
return &pb.GetResponse{
|
|
Pointer: pointerBytes,
|
|
}, nil
|
|
}
|
|
|
|
// List calls the bolt client's List function and returns all Path keys in the Pointers bucket
|
|
func (s *Server) List(ctx context.Context, req *pb.ListRequest) (*pb.ListResponse, error) {
|
|
s.logger.Debug("entering netstate list")
|
|
|
|
if req.Limit <= 0 {
|
|
return nil, Error.New("err Limit is less than or equal to 0")
|
|
}
|
|
|
|
APIKeyBytes := []byte(req.APIKey)
|
|
if err := s.validateAuth(APIKeyBytes); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
var keyList storage.Keys
|
|
if req.StartingPathKey == nil {
|
|
pathKeys, err := s.DB.List(nil, storage.Limit(req.Limit))
|
|
if err != nil {
|
|
s.logger.Error("err listing path keys with no starting key", zap.Error(err))
|
|
return nil, status.Errorf(codes.Internal, err.Error())
|
|
}
|
|
keyList = pathKeys
|
|
} else if req.StartingPathKey != nil {
|
|
pathKeys, err := s.DB.List(storage.Key(req.StartingPathKey), storage.Limit(req.Limit))
|
|
if err != nil {
|
|
s.logger.Error("err listing path keys", zap.Error(err))
|
|
return nil, status.Errorf(codes.Internal, err.Error())
|
|
}
|
|
keyList = pathKeys
|
|
}
|
|
|
|
truncated := isItTruncated(keyList, int(req.Limit))
|
|
|
|
s.logger.Debug("path keys retrieved")
|
|
return &pb.ListResponse{
|
|
Paths: keyList.ByteSlices(),
|
|
Truncated: truncated,
|
|
}, nil
|
|
}
|
|
|
|
func isItTruncated(keyList storage.Keys, limit int) bool {
|
|
if len(keyList) == limit {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
// Delete formats and hands off a file path to delete from boltdb
|
|
func (s *Server) Delete(ctx context.Context, req *pb.DeleteRequest) (*pb.DeleteResponse, error) {
|
|
s.logger.Debug("entering netstate delete")
|
|
|
|
APIKeyBytes := []byte(req.APIKey)
|
|
if err := s.validateAuth(APIKeyBytes); err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
err := s.DB.Delete(req.Path)
|
|
if err != nil {
|
|
s.logger.Error("err deleting path and pointer", zap.Error(err))
|
|
return nil, status.Errorf(codes.Internal, err.Error())
|
|
}
|
|
s.logger.Debug("deleted pointer at path: " + string(req.Path))
|
|
return &pb.DeleteResponse{}, nil
|
|
}
|