fix: gRPC client

Author: Andrew Hare

cmd/provider-services/cmd/manifest.go

@@ -2,9 +2,10 @@ package cmd
 
 import (
 	"bytes"
-	"crypto/tls"
+	"context"
 	"encoding/json"
 	"fmt"
+	"time"
 
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
@@ -14,16 +15,16 @@ import (
 	sdk "github.com/cosmos/cosmos-sdk/types"
 
 	dtypes "github.com/akash-network/akash-api/go/node/deployment/v1beta3"
+	ptypes "github.com/akash-network/akash-api/go/node/provider/v1beta3"
+	leasev1 "github.com/akash-network/akash-api/go/provider/lease/v1"
 	"github.com/akash-network/node/sdl"
 	cutils "github.com/akash-network/node/x/cert/utils"
 
 	aclient "github.com/akash-network/provider/client"
-	gwrest "github.com/akash-network/provider/gateway/rest"
+	gwgrpc "github.com/akash-network/provider/gateway/grpc"
 )
 
-var (
-	errSubmitManifestFailed = errors.New("submit manifest to some providers has been failed")
-)
+var errSubmitManifestFailed = errors.New("submit manifest to some providers has been failed")
 
 // SendManifestCmd looks up the Providers blockchain information,
 // and POSTs the SDL file to the Gateway address.
@@ -94,32 +95,50 @@ func doSendManifest(cmd *cobra.Command, sdlpath string) error {
 		ErrorMessage string      `json:"errorMessage,omitempty" yaml:"errorMessage,omitempty"`
 	}
 
-	results := make([]result, len(leases))
-
-	submitFailed := false
+	var (
+		results      = make([]result, len(leases))
+		submitFailed = false
+	)
 
 	for i, lid := range leases {
-		prov, _ := sdk.AccAddressFromBech32(lid.Provider)
-		gclient, err := gwrest.NewClient(cl, prov, []tls.Certificate{cert})
-		if err != nil {
-			return err
-		}
+		err := func() error {
+			ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
+			defer cancel()
+
+			provAddr, _ := sdk.AccAddressFromBech32(lid.Provider)
+			prov, err := cl.Provider(context.Background(), &ptypes.QueryProviderRequest{Owner: provAddr.String()})
+			if err != nil {
+				return fmt.Errorf("query client provider: %w", err)
+			}
 
-		err = gclient.SubmitManifest(cmd.Context(), dseq, mani)
-		res := result{
-			Provider: prov,
-			Status:   "PASS",
-		}
-		if err != nil {
-			res.Error = err.Error()
-			if e, valid := err.(gwrest.ClientResponseError); valid {
-				res.ErrorMessage = e.Message
+			c, err := gwgrpc.NewClient(ctx, prov.GetProvider().HostURI, cert, cl)
+			if err != nil {
+				return fmt.Errorf("new grpc conn: %w", err)
 			}
-			res.Status = "FAIL"
-			submitFailed = true
-		}
 
-		results[i] = res
+			defer c.Close()
+
+			res := result{
+				Provider: provAddr,
+				Status:   "PASS",
+			}
+
+			if _, err = c.SendManifest(ctx, &leasev1.SendManifestRequest{
+				LeaseId:  lid,
+				Manifest: mani,
+			}); err != nil {
+				res.Error = err.Error()
+				res.Status = "FAIL"
+				submitFailed = true
+			}
+
+			results[i] = res
+
+			return nil
+		}()
+		if err != nil {
+			return err
+		}
 	}
 
 	buf := &bytes.Buffer{}
@@ -146,7 +165,6 @@ func doSendManifest(cmd *cobra.Command, sdlpath string) error {
 	}
 
 	_, err = fmt.Fprint(cmd.OutOrStdout(), buf.String())
-
 	if err != nil {
 		return err
 	}

cmd/provider-services/cmd/run.go

@@ -109,9 +109,7 @@ const (
 	serviceHostnameOperator = "hostname-operator"
 )
 
-var (
-	errInvalidConfig = errors.New("Invalid configuration")
-)
+var errInvalidConfig = errors.New("Invalid configuration")
 
 // RunCmd launches the Akash Provider service
 func RunCmd() *cobra.Command {
@@ -206,7 +204,7 @@ func RunCmd() *cobra.Command {
 		panic(err)
 	}
 
-	cmd.Flags().String(FlagGatewayGRPCListenAddress, "0.0.0.0:8444", "Gateway listen address")
+	cmd.Flags().String(FlagGatewayGRPCListenAddress, "0.0.0.0:8442", "Gateway gRPC listen address")
 	if err := viper.BindPFlag(FlagGatewayGRPCListenAddress, cmd.Flags().Lookup(FlagGatewayGRPCListenAddress)); err != nil {
 		panic(err)
 	}
@@ -420,9 +418,11 @@ var allowedBidPricingStrategies = [...]string{
 	bidPricingStrategyShellScript,
 }
 
-var errNoSuchBidPricingStrategy = fmt.Errorf("No such bid pricing strategy. Allowed: %v", allowedBidPricingStrategies)
-var errInvalidValueForBidPrice = errors.New("not a valid bid price")
-var errBidPriceNegative = errors.New("Bid price cannot be a negative number")
+var (
+	errNoSuchBidPricingStrategy = fmt.Errorf("No such bid pricing strategy. Allowed: %v", allowedBidPricingStrategies)
+	errInvalidValueForBidPrice  = errors.New("not a valid bid price")
+	errBidPriceNegative         = errors.New("Bid price cannot be a negative number")
+)
 
 func strToBidPriceScale(val string) (decimal.Decimal, error) {
 	v, err := decimal.NewFromString(val)

gateway/grpc/client.go

@@ -0,0 +1,56 @@
+package grpc
+
+import (
+	"context"
+	"crypto/tls"
+	"crypto/x509"
+	"fmt"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+
+	ctypes "github.com/akash-network/akash-api/go/node/cert/v1beta3"
+	leasev1 "github.com/akash-network/akash-api/go/provider/lease/v1"
+	providerv1 "github.com/akash-network/akash-api/go/provider/v1"
+
+	"github.com/akash-network/provider/gateway/utils"
+)
+
+type Client struct {
+	providerv1.ProviderRPCClient
+	leasev1.LeaseRPCClient
+
+	conn *grpc.ClientConn
+}
+
+func (c *Client) Close() error {
+	return c.conn.Close()
+}
+
+func NewClient(ctx context.Context, addr string, cert tls.Certificate, cquery ctypes.QueryClient) (*Client, error) {
+	tlsConfig := tls.Config{
+		InsecureSkipVerify: true,
+		Certificates:       []tls.Certificate{cert},
+		VerifyPeerCertificate: func(certificates [][]byte, _ [][]*x509.Certificate) error {
+			if _, err := utils.VerifyOwnerCertBytes(ctx, certificates, "", x509.ExtKeyUsageClientAuth, cquery); err != nil {
+				return err
+			}
+			return nil
+		},
+	}
+
+	conn, err := grpc.DialContext(ctx, addr,
+		grpc.WithBlock(),
+		grpc.WithTransportCredentials(credentials.NewTLS(&tlsConfig)),
+	)
+	if err != nil {
+		return nil, fmt.Errorf("grpc dial context %s: %w", addr, err)
+	}
+
+	return &Client{
+		ProviderRPCClient: providerv1.NewProviderRPCClient(conn),
+		LeaseRPCClient:    leasev1.NewLeaseRPCClient(conn),
+
+		conn: conn,
+	}, nil
+}