akash provider: multipl fixes

 * fix required flags
 * show error if api host is not expected provider
 * actually show all providers

Author: boz

cmd/akash/provider.go

@@ -33,8 +33,6 @@ func providerCommand() *cobra.Command {
 	}
 
 	session.AddFlagNode(cmd, cmd.PersistentFlags())
-	session.AddFlagKey(cmd, cmd.PersistentFlags())
-	session.AddFlagNonce(cmd, cmd.PersistentFlags())
 
 	cmd.AddCommand(createProviderCommand())
 	cmd.AddCommand(runCommand())
@@ -54,6 +52,8 @@ func createProviderCommand() *cobra.Command {
 		RunE:  session.WithSession(session.RequireNode(doCreateProviderCommand)),
 	}
 
+	session.AddFlagKey(cmd, cmd.Flags())
+	session.AddFlagNonce(cmd, cmd.Flags())
 	session.AddFlagKeyType(cmd, cmd.Flags())
 
 	return cmd
@@ -132,6 +132,8 @@ func runCommand() *cobra.Command {
 		RunE:  session.WithSession(session.RequireNode(session.RequireHost(doProviderRunCommand))),
 	}
 
+	session.AddFlagKey(cmd, cmd.Flags())
+	session.AddFlagNonce(cmd, cmd.Flags())
 	session.AddFlagHost(cmd, cmd.PersistentFlags())
 
 	cmd.Flags().Bool("kube", false, "use kubernetes cluster")
@@ -209,7 +211,7 @@ func doProviderRunCommand(session session.Session, cmd *cobra.Command, args []st
 
 		go func() {
 			defer cancel()
-			errch <- grpc.RunServer(ctx, session.Log(), "tcp", "9090", service.ManifestHandler(), cclient, service)
+			errch <- grpc.Run(ctx, ":9090", psession, cclient, service, service.ManifestHandler())
 		}()
 
 		go func() {
@@ -245,7 +247,7 @@ func doProviderStatusCommand(session session.Session, cmd *cobra.Command, args [
 		return err
 	}
 
-	var providers []types.Provider
+	var providers []*types.Provider
 
 	if len(args) == 0 {
 		providers = plist.Providers
@@ -265,20 +267,34 @@ func doProviderStatusCommand(session session.Session, cmd *cobra.Command, args [
 	}
 
 	type outputItem struct {
-		Provider *types.Provider
-		Status   *types.ServerStatusParseable
-		Error    string `json:",omitempty"`
+		Provider *types.Provider              `json:"provider,omitempty"`
+		Status   *types.ServerStatusParseable `json:"status,omitempty"`
+		Error    string                       `json:"error,omitempty"`
 	}
 
 	output := []outputItem{}
 
 	for _, provider := range providers {
-		status, err := http.Status(session.Ctx(), &provider)
+
+		status, err := http.Status(session.Ctx(), provider)
 		if err != nil {
-			output = append(output, outputItem{Provider: &provider, Error: err.Error()})
+			output = append(output, outputItem{Provider: provider, Error: err.Error()})
+			continue
+		}
+
+		if !bytes.Equal(status.Provider, provider.Address) {
+			output = append(output, outputItem{
+				Provider: provider,
+				Status:   status,
+				Error:    "Status received from incorrect provider",
+			})
 			continue
 		}
-		output = append(output, outputItem{Provider: &provider, Status: status})
+
+		output = append(output, outputItem{
+			Provider: provider,
+			Status:   status,
+		})
 	}
 
 	buf, err := json.MarshalIndent(output, "", " ")
@@ -298,7 +314,8 @@ func closeFulfillmentCommand() *cobra.Command {
 		RunE:  session.WithSession(session.RequireNode(doCloseFulfillmentCommand)),
 	}
 
-	session.AddFlagKeyType(cmd, cmd.Flags())
+	session.AddFlagKey(cmd, cmd.Flags())
+	session.AddFlagNonce(cmd, cmd.Flags())
 
 	return cmd
 }
@@ -329,8 +346,6 @@ func closeLeaseCommand() *cobra.Command {
 		RunE:  session.WithSession(session.RequireNode(doCloseLeaseCommand)),
 	}
 
-	session.AddFlagKeyType(cmd, cmd.Flags())
-
 	return cmd
 }
 

provider/grpc/client_test.go

@@ -2,20 +2,17 @@ package grpc
 
 import (
 	"context"
-	"os"
 	"testing"
-	"time"
 
 	"github.com/ovrclk/akash/manifest"
-	kmocks "github.com/ovrclk/akash/provider/cluster/kube/mocks"
+	cmocks "github.com/ovrclk/akash/provider/cluster/mocks"
 	"github.com/ovrclk/akash/provider/manifest/mocks"
-	pmocks "github.com/ovrclk/akash/provider/mocks"
+	"github.com/ovrclk/akash/provider/session"
 	"github.com/ovrclk/akash/sdl"
 	"github.com/ovrclk/akash/testutil"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 	"github.com/stretchr/testify/require"
-	"github.com/tendermint/tmlibs/log"
 )
 
 func TestSendManifest(t *testing.T) {
@@ -28,29 +25,37 @@ func TestSendManifest(t *testing.T) {
 	mani, err := sdl.Manifest()
 	require.NoError(t, err)
 
-	_, kmgr := testutil.NewNamedKey(t)
+	key, kmgr := testutil.NewNamedKey(t)
 	signer := testutil.Signer(t, kmgr)
 
+	provider := testutil.Provider(key.Address(), 1)
+	session := session.New(testutil.Logger(), provider, nil, nil)
+
 	deployment := testutil.DeploymentAddress(t)
 
 	req, _, err := manifest.SignManifest(mani, signer, deployment)
 	assert.NoError(t, err)
 
-	sclient := &pmocks.StatusClient{}
-
 	handler := &mocks.Handler{}
 	handler.On("HandleManifest", mock.Anything, mock.Anything).Return(nil)
 
-	client := &kmocks.Client{}
+	client := &cmocks.Client{}
+
+	donech := make(chan struct{})
+	ctx, cancel := context.WithCancel(context.Background())
 
-	server := newServer(log.NewTMLogger(os.Stdout), "tcp", ":3001", handler, client, sclient)
 	go func() {
-		err := server.listenAndServe()
-		require.NoError(t, err)
+		defer close(donech)
+		assert.NoError(t, Run(ctx, ":3001", session, client, nil, handler))
 	}()
 
-	time.Sleep(1 * time.Second)
+	testutil.SleepForThreadStart(t)
 
 	_, err = c.Deploy(context.TODO(), req)
 	assert.NoError(t, err)
+
+	testutil.SleepForThreadStart(t)
+
+	cancel()
+	<-donech
 }

provider/grpc/server.go

@@ -1,7 +1,6 @@
 package grpc
 
 import (
-	"fmt"
 	"net"
 	"net/http"
 	"strings"
@@ -10,67 +9,46 @@ import (
 	"github.com/ovrclk/akash/keys"
 	"github.com/ovrclk/akash/provider"
 	"github.com/ovrclk/akash/provider/cluster"
-	"github.com/ovrclk/akash/provider/cluster/kube"
 	"github.com/ovrclk/akash/provider/manifest"
+	"github.com/ovrclk/akash/provider/session"
 	"github.com/ovrclk/akash/types"
 	"github.com/ovrclk/akash/version"
 	"github.com/tendermint/tmlibs/log"
 	"golang.org/x/net/context"
-	"golang.org/x/net/netutil"
 	"google.golang.org/grpc"
 )
 
 type server struct {
-	cluster.Client
-	*grpc.Server
+	session session.Session
+	client  cluster.Client
 	status  provider.StatusClient
 	handler manifest.Handler
-	network string
-	port    string
 	log     log.Logger
 }
 
-func RunServer(ctx context.Context, log log.Logger, network,
-	port string, handler manifest.Handler, client kube.Client,
-	status provider.StatusClient) error {
-
-	address := fmt.Sprintf(":%v", port)
-
-	server := newServer(log, network, address, handler, client, status)
-
-	ctx, cancel := context.WithCancel(ctx)
-
-	donech := make(chan struct{})
-
-	go func() {
-		defer close(donech)
-		<-ctx.Done()
-		log.Info("Shutting down server")
-		server.GracefulStop()
-	}()
-
-	log.Info("Starting GRPC server", "address", address)
-	err := server.listenAndServe()
-	cancel()
-
-	<-donech
-
-	log.Info("GRPC server shutdown")
-
-	return err
+func Run(
+	ctx context.Context,
+	address string,
+	session session.Session,
+	client cluster.Client,
+	status provider.StatusClient,
+	handler manifest.Handler) error {
+	server := create(session, client, status, handler)
+	return run(ctx, server, address)
 }
 
-func (s server) Status(ctx context.Context, req *types.Empty) (*types.ServerStatus, error) {
+func (s *server) Status(ctx context.Context, req *types.Empty) (*types.ServerStatus, error) {
 	status, err := s.status.Status(ctx)
 	if err != nil {
 		return nil, err
 	}
 
 	vsn := version.Get()
 	return &types.ServerStatus{
-		Code:     http.StatusOK,
+		Provider: s.session.Provider().Address,
 		Version:  &vsn,
-		Provider: status,
+		Status:   status,
+		Code:     http.StatusOK,
 		Message:  "OK",
 	}, nil
 }
@@ -85,7 +63,7 @@ func (s server) Deploy(ctx context.Context, req *types.ManifestRequest) (*types.
 }
 
 // Lease status will retry for one minute
-func (s server) LeaseStatus(ctx context.Context, req *types.LeaseStatusRequest) (*types.LeaseStatusResponse, error) {
+func (s *server) LeaseStatus(ctx context.Context, req *types.LeaseStatusRequest) (*types.LeaseStatusResponse, error) {
 	attempts := 12
 
 	lease, err := keys.ParseLeasePath(strings.Join([]string{req.Deployment, req.Group, req.Order, req.Provider}, "/"))
@@ -94,14 +72,14 @@ func (s server) LeaseStatus(ctx context.Context, req *types.LeaseStatusRequest)
 		return nil, types.ErrInternalError{Message: "internal error"}
 	}
 
-	response, err := s.Client.LeaseStatus(lease.LeaseID)
+	response, err := s.client.LeaseStatus(lease.LeaseID)
 	if err == nil {
 		return response, err
 	}
 
 	for i := 0; i < attempts; i++ {
 		time.Sleep(time.Second * 5)
-		response, err = s.Client.LeaseStatus(lease.LeaseID)
+		response, err = s.client.LeaseStatus(lease.LeaseID)
 		if err != cluster.ErrNoDeployments {
 			break
 		}
@@ -110,23 +88,23 @@ func (s server) LeaseStatus(ctx context.Context, req *types.LeaseStatusRequest)
 	return response, err
 }
 
-func (s server) ServiceStatus(ctx context.Context,
+func (s *server) ServiceStatus(ctx context.Context,
 	req *types.ServiceStatusRequest) (*types.ServiceStatusResponse, error) {
 	lease, err := keys.ParseLeasePath(strings.Join([]string{req.Deployment, req.Group, req.Order, req.Provider}, "/"))
 	if err != nil {
 		s.log.Error(err.Error())
 		return nil, types.ErrInternalError{Message: "internal error"}
 	}
-	return s.Client.ServiceStatus(lease.LeaseID, req.Name)
+	return s.client.ServiceStatus(lease.LeaseID, req.Name)
 }
 
-func (s server) ServiceLogs(req *types.LogRequest, server types.Cluster_ServiceLogsServer) error {
+func (s *server) ServiceLogs(req *types.LogRequest, server types.Cluster_ServiceLogsServer) error {
 	lease, err := keys.ParseLeasePath(strings.Join([]string{req.Deployment, req.Group, req.Order, req.Provider}, "/"))
 	if err != nil {
 		s.log.Error(err.Error())
 		return types.ErrInternalError{Message: "internal error"}
 	}
-	logs, err := s.Client.ServiceLogs(server.Context(), lease.LeaseID, req.Options.TailLines, req.Options.Follow)
+	logs, err := s.client.ServiceLogs(server.Context(), lease.LeaseID, req.Options.TailLines, req.Options.Follow)
 	if err != nil {
 		s.log.Error(err.Error())
 		return types.ErrInternalError{Message: "internal error"}
@@ -165,28 +143,54 @@ func (s server) ServiceLogs(req *types.LogRequest, server types.Cluster_ServiceL
 	return nil
 }
 
-// NewServer network can be "tcp", "tcp4", "tcp6", "unix" or "unixpacket". phandler is the provider cluster handler
-func newServer(log log.Logger, network, port string, handler manifest.Handler,
-	client kube.Client, status provider.StatusClient) *server {
-	s := &server{
+func create(
+	session session.Session,
+	client cluster.Client,
+	status provider.StatusClient,
+	handler manifest.Handler) *server {
+
+	log := session.Log().With("cmp", "grpc-server")
+
+	return &server{
+		session: session,
+		client:  client,
 		status:  status,
 		handler: handler,
-		network: network,
-		port:    port,
-		Server:  grpc.NewServer(grpc.MaxConcurrentStreams(2), grpc.MaxRecvMsgSize(500000)),
 		log:     log,
-		Client:  client,
 	}
-	types.RegisterClusterServer(s.Server, s)
-	return s
 }
 
-func (s *server) listenAndServe() error {
-	l, err := net.Listen(s.network, s.port)
+func run(ctx context.Context, server *server, address string) error {
+
+	fd, err := net.Listen("tcp4", address)
 	if err != nil {
 		return err
 	}
-	l = netutil.LimitListener(l, 10)
-	s.log.Info("Running manifest server", "port", s.port, "network", s.network)
-	return s.Server.Serve(l)
+
+	gserver := grpc.NewServer(grpc.MaxConcurrentStreams(2), grpc.MaxRecvMsgSize(500000))
+	types.RegisterClusterServer(gserver, server)
+
+	ctx, cancel := context.WithCancel(ctx)
+	donech := make(chan struct{})
+
+	go func() {
+		defer close(donech)
+		<-ctx.Done()
+		server.log.Info("Shutting down server")
+		gserver.GracefulStop()
+	}()
+
+	server.log.Info("Starting GRPC server", "address", address)
+	err = gserver.Serve(fd)
+	server.log.Info("GRPC server shutdown.")
+	if ctx.Err() == context.Canceled {
+		err = nil
+	}
+	cancel()
+
+	<-donech
+
+	server.log.Info("GRPC done.")
+
+	return err
 }