Merge pull request #27 from ultravioletrs/NOISSUE-grpcclient

NOISSUE - Add GRPC Client

cmd/cli/main.go

@@ -2,30 +2,25 @@ package main
 
 import (
 	"fmt"
-	"io"
-	"io/ioutil"
 	"log"
 	"os"
-	"time"
 
 	"github.com/mainflux/mainflux/logger"
-	"github.com/opentracing/opentracing-go"
 	"github.com/spf13/cobra"
-	jconfig "github.com/uber/jaeger-client-go/config"
-	agentgrpc "github.com/ultravioletrs/agent/agent/api/grpc"
 	"github.com/ultravioletrs/agent/cli"
 	"github.com/ultravioletrs/agent/internal/env"
-	agentsdk "github.com/ultravioletrs/agent/pkg/sdk"
-	ggrpc "google.golang.org/grpc"
+	"github.com/ultravioletrs/agent/pkg/clients/grpc"
+	"github.com/ultravioletrs/agent/pkg/sdk"
 )
 
-const svcName = "cli"
+const (
+	svcName            = "cli"
+	envPrefixAgentGRPC = "AGENT_GRPC_"
+)
 
 type config struct {
-	LogLevel         string `env:"AGENT_LOG_LEVEL"      envDefault:"info"`
-	AgentGRPCURL     string `env:"AGENT_GRPC_URL"       envDefault:"localhost:7002"`
-	AgentGRPCTimeout string `env:"AGENT_GRPC_TIMEOUT"   envDefault:"1s"`
-	JaegerURL        string `env:"AGENT_JAEGER_URL"     envDefault:""`
+	LogLevel  string `env:"AGENT_LOG_LEVEL"      envDefault:"info"`
+	JaegerURL string `env:"AGENT_JAEGER_URL"     envDefault:""`
 }
 
 func main() {
@@ -39,18 +34,20 @@ func main() {
 		log.Fatalf("Error creating logger: %s", err)
 	}
 
-	conn := connectToGrpc("agent", cfg.AgentGRPCURL, logger)
-
-	agentTracer, agentCloser := initJaeger("agent", cfg.JaegerURL, logger)
-	defer agentCloser.Close()
+	agentGRPCConfig := grpc.Config{}
+	if err := env.Parse(&agentGRPCConfig, env.Options{Prefix: envPrefixAgentGRPC}); err != nil {
+		logger.Fatal(fmt.Sprintf("failed to load %s gRPC client configuration : %s", svcName, err))
+	}
 
-	timeout, err := time.ParseDuration(cfg.AgentGRPCTimeout)
+	agentGRPCClient, agentClient, err := grpc.NewClient(agentGRPCConfig)
 	if err != nil {
-		log.Fatalf("Error parsing timeout: %s", err)
+		logger.Fatal(err.Error())
 	}
-	agentClient := agentgrpc.NewClient(agentTracer, conn, timeout)
+	defer agentGRPCClient.Close()
 
-	sdk := agentsdk.NewAgentSDK(logger, agentClient)
+	logger.Info("Successfully connected to agent grpc server " + agentGRPCClient.Secure())
+
+	sdk := sdk.NewAgentSDK(logger, agentClient)
 
 	cli.SetSDK(sdk)
 
@@ -71,39 +68,3 @@ func main() {
 		os.Exit(1)
 	}
 }
-
-func connectToGrpc(name string, url string, logger logger.Logger) *ggrpc.ClientConn {
-	opts := []ggrpc.DialOption{ggrpc.WithInsecure()}
-	conn, err := ggrpc.Dial(url, opts...)
-	if err != nil {
-		logger.Error(fmt.Sprintf("Failed to connect to %s service: %s", name, err))
-		os.Exit(1)
-	}
-	logger.Info(fmt.Sprintf("Connected to %s gRPC server on %s", name, url))
-
-	return conn
-}
-
-func initJaeger(svcName, url string, logger logger.Logger) (opentracing.Tracer, io.Closer) {
-	if url == "" {
-		return opentracing.NoopTracer{}, ioutil.NopCloser(nil)
-	}
-
-	tracer, closer, err := jconfig.Configuration{
-		ServiceName: svcName,
-		Sampler: &jconfig.SamplerConfig{
-			Type:  "const",
-			Param: 1,
-		},
-		Reporter: &jconfig.ReporterConfig{
-			LocalAgentHostPort: url,
-			LogSpans:           true,
-		},
-	}.NewTracer()
-	if err != nil {
-		logger.Error(fmt.Sprintf("Failed to init Jaeger client: %s", err))
-		os.Exit(1)
-	}
-
-	return tracer, closer
-}

pkg/clients/doc.go

@@ -0,0 +1,3 @@
+// Package clients contains the domain concept definitions needed to support
+// Agent Client functionality.
+package clients

pkg/clients/grpc/agent.go

@@ -0,0 +1,17 @@
+package grpc
+
+import (
+	"github.com/opentracing/opentracing-go"
+	"github.com/ultravioletrs/agent/agent"
+	agentapi "github.com/ultravioletrs/agent/agent/api/grpc"
+)
+
+// NewClient creates new agent gRPC client instance.
+func NewClient(cfg Config) (Client, agent.AgentServiceClient, error) {
+	client, err := newClient(cfg)
+	if err != nil {
+		return nil, nil, err
+	}
+
+	return client, agentapi.NewClient(opentracing.NoopTracer{}, client.Connection(), cfg.Timeout), nil
+}

pkg/clients/grpc/connect.go

@@ -0,0 +1,99 @@
+package grpc
+
+import (
+	"time"
+
+	"github.com/mainflux/mainflux/pkg/errors"
+	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
+	gogrpc "google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+	"google.golang.org/grpc/credentials/insecure"
+)
+
+var (
+	errGrpcConnect = errors.New("failed to connect to grpc server")
+	errGrpcClose   = errors.New("failed to close grpc connection")
+)
+
+type Config struct {
+	ClientTLS bool          `env:"CLIENT_TLS"    envDefault:"false"`
+	CACerts   string        `env:"CA_CERTS"      envDefault:""`
+	URL       string        `env:"URL"           envDefault:"localhost:7002"`
+	Timeout   time.Duration `env:"TIMEOUT"       envDefault:"1s"`
+}
+
+type Client interface {
+	// Close closes gRPC connection.
+	Close() error
+
+	// Secure is used for pretty printing TLS info.
+	Secure() string
+
+	// Connection returns the gRPC connection.
+	Connection() *gogrpc.ClientConn
+}
+
+type client struct {
+	*gogrpc.ClientConn
+	cfg    Config
+	secure bool
+}
+
+var _ Client = (*client)(nil)
+
+func newClient(cfg Config) (Client, error) {
+	conn, secure, err := connect(cfg)
+	if err != nil {
+		return nil, err
+	}
+
+	return &client{
+		ClientConn: conn,
+		cfg:        cfg,
+		secure:     secure,
+	}, nil
+}
+
+func (c *client) Close() error {
+	if err := c.ClientConn.Close(); err != nil {
+		return errors.Wrap(errGrpcClose, err)
+	}
+
+	return nil
+}
+
+func (c *client) Secure() string {
+	if c.secure {
+		return "with TLS"
+	}
+	return "without TLS"
+}
+
+func (c *client) Connection() *gogrpc.ClientConn {
+	return c.ClientConn
+}
+
+// connect creates new gRPC client and connect to gRPC server.
+func connect(cfg Config) (*gogrpc.ClientConn, bool, error) {
+	var opts []gogrpc.DialOption
+	secure := false
+	tc := insecure.NewCredentials()
+
+	if cfg.ClientTLS && cfg.CACerts != "" {
+		var err error
+		tc, err = credentials.NewClientTLSFromFile(cfg.CACerts, "")
+		if err != nil {
+			return nil, secure, err
+		}
+		secure = true
+	}
+
+	opts = append(opts, gogrpc.WithTransportCredentials(tc), gogrpc.WithUnaryInterceptor(otelgrpc.UnaryClientInterceptor()))
+
+	conn, err := gogrpc.Dial(cfg.URL, opts...)
+	if err != nil {
+		return nil, secure, errors.Wrap(errGrpcConnect, err)
+	}
+
+	return conn, secure, nil
+}

pkg/clients/grpc/doc.go

@@ -0,0 +1,3 @@
+// Package grpc contains the domain concept definitions needed to support
+// Agent Client grpc functionality.
+package grpc