Add remote gRPC option for storage plugin

examples/memstore-plugin/main.go

@@ -27,10 +27,9 @@ import (
 	googleGRPC "google.golang.org/grpc"
 
 	"github.com/jaegertracing/jaeger/plugin/storage/grpc"
+	grpcMemory "github.com/jaegertracing/jaeger/plugin/storage/grpc/memory"
 	"github.com/jaegertracing/jaeger/plugin/storage/grpc/shared"
 	"github.com/jaegertracing/jaeger/plugin/storage/memory"
-	"github.com/jaegertracing/jaeger/storage/dependencystore"
-	"github.com/jaegertracing/jaeger/storage/spanstore"
 )
 
 const (
@@ -71,10 +70,7 @@ func main() {
 	defer closer.Close()
 	opentracing.SetGlobalTracer(tracer)
 
-	memStorePlugin := &memoryStorePlugin{
-		store:        memory.NewStore(),
-		archiveStore: memory.NewStore(),
-	}
+	memStorePlugin := grpcMemory.NewStoragePlugin(memory.NewStore(), memory.NewStore())
 	grpc.ServeWithGRPCServer(&shared.PluginServices{
 		Store:        memStorePlugin,
 		ArchiveStore: memStorePlugin,
@@ -85,28 +81,3 @@ func main() {
 		})
 	})
 }
-
-type memoryStorePlugin struct {
-	store        *memory.Store
-	archiveStore *memory.Store
-}
-
-func (ns *memoryStorePlugin) DependencyReader() dependencystore.Reader {
-	return ns.store
-}
-
-func (ns *memoryStorePlugin) SpanReader() spanstore.Reader {
-	return ns.store
-}
-
-func (ns *memoryStorePlugin) SpanWriter() spanstore.Writer {
-	return ns.store
-}
-
-func (ns *memoryStorePlugin) ArchiveSpanReader() spanstore.Reader {
-	return ns.archiveStore
-}
-
-func (ns *memoryStorePlugin) ArchiveSpanWriter() spanstore.Writer {
-	return ns.archiveStore
-}

plugin/storage/grpc/config/config.go

@@ -15,16 +15,21 @@
 package config
 
 import (
+	"context"
 	"fmt"
 	"os/exec"
 	"runtime"
+	"time"
 
 	"github.com/grpc-ecosystem/grpc-opentracing/go/otgrpc"
 	"github.com/hashicorp/go-hclog"
 	"github.com/hashicorp/go-plugin"
 	"github.com/opentracing/opentracing-go"
+	"go.uber.org/zap"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 
+	"github.com/jaegertracing/jaeger/pkg/config/tlscfg"
 	"github.com/jaegertracing/jaeger/plugin/storage/grpc/shared"
 )
 
@@ -33,6 +38,9 @@ type Configuration struct {
 	PluginBinary            string `yaml:"binary" mapstructure:"binary"`
 	PluginConfigurationFile string `yaml:"configuration-file" mapstructure:"configuration_file"`
 	PluginLogLevel          string `yaml:"log-level" mapstructure:"log_level"`
+	RemoteServerAddr        string `yaml:"server" mapstructure:"server"`
+	RemoteTLS               tlscfg.Options
+	RemoteConnectTimeout    time.Duration `yaml:"connection-timeout" mapstructure:"connection-timeout"`
 }
 
 // ClientPluginServices defines services plugin can expose and its capabilities
@@ -43,11 +51,61 @@ type ClientPluginServices struct {
 
 // PluginBuilder is used to create storage plugins. Implemented by Configuration.
 type PluginBuilder interface {
-	Build() (*ClientPluginServices, error)
+	Build(logger *zap.Logger) (*ClientPluginServices, error)
+	Close() error
 }
 
 // Build instantiates a PluginServices
-func (c *Configuration) Build() (*ClientPluginServices, error) {
+func (c *Configuration) Build(logger *zap.Logger) (*ClientPluginServices, error) {
+	if c.PluginBinary != "" {
+		return c.buildPlugin()
+	} else {
+		return c.buildRemote(logger)
+	}
+}
+
+func (c *Configuration) Close() error {
+	if c.PluginBinary == "" {
+		return c.RemoteTLS.Close()
+	}
+	return nil
+}
+
+func (c *Configuration) buildRemote(logger *zap.Logger) (*ClientPluginServices, error) {
+	opts := []grpc.DialOption{
+		grpc.WithUnaryInterceptor(otgrpc.OpenTracingClientInterceptor(opentracing.GlobalTracer())),
+		grpc.WithStreamInterceptor(otgrpc.OpenTracingStreamClientInterceptor(opentracing.GlobalTracer())),
+		grpc.WithBlock(),
+	}
+	if c.RemoteTLS.Enabled {
+		tlsCfg, err := c.RemoteTLS.Config(logger)
+		if err != nil {
+			return nil, err
+		}
+		creds := credentials.NewTLS(tlsCfg)
+		opts = append(opts, grpc.WithTransportCredentials(creds))
+	} else {
+		opts = append(opts, grpc.WithInsecure())
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), c.RemoteConnectTimeout)
+	defer cancel()
+	conn, err := grpc.DialContext(ctx, c.RemoteServerAddr, opts...)
+	if err != nil {
+		return nil, fmt.Errorf("error connecting to remote storage: %w", err)
+	}
+
+	grpcClient := shared.NewGRPCClient(conn)
+	return &ClientPluginServices{
+		PluginServices: shared.PluginServices{
+			Store:        grpcClient,
+			ArchiveStore: grpcClient,
+		},
+		Capabilities: grpcClient,
+	}, nil
+}
+
+func (c *Configuration) buildPlugin() (*ClientPluginServices, error) {
 	// #nosec G204
 	cmd := exec.Command(c.PluginBinary, "--config", c.PluginConfigurationFile)
 

plugin/storage/grpc/factory.go

@@ -68,7 +68,7 @@ func (f *Factory) InitFromOptions(opts Options) {
 func (f *Factory) Initialize(metricsFactory metrics.Factory, logger *zap.Logger) error {
 	f.metricsFactory, f.logger = metricsFactory, logger
 
-	services, err := f.builder.Build()
+	services, err := f.builder.Build(logger)
 	if err != nil {
 		return fmt.Errorf("grpc-plugin builder failed to create a store: %w", err)
 	}
@@ -124,3 +124,8 @@ func (f *Factory) CreateArchiveSpanWriter() (spanstore.Writer, error) {
 	}
 	return f.archiveStore.ArchiveSpanWriter(), nil
 }
+
+// Close closes the resources held by the factory
+func (f *Factory) Close() error {
+	return f.builder.Close()
+}

plugin/storage/grpc/factory_test.go

@@ -42,7 +42,7 @@ type mockPluginBuilder struct {
 	err    error
 }
 
-func (b *mockPluginBuilder) Build() (*grpcConfig.ClientPluginServices, error) {
+func (b *mockPluginBuilder) Build(logger *zap.Logger) (*grpcConfig.ClientPluginServices, error) {
 	if b.err != nil {
 		return nil, b.err
 	}
@@ -60,6 +60,10 @@ func (b *mockPluginBuilder) Build() (*grpcConfig.ClientPluginServices, error) {
 	return services, nil
 }
 
+func (b *mockPluginBuilder) Close() error {
+	return nil
+}
+
 type mockPlugin struct {
 	spanReader       spanstore.Reader
 	spanWriter       spanstore.Writer

plugin/storage/grpc/memory/plugin.go

@@ -0,0 +1,50 @@
+// Copyright (c) 2020 The Jaeger Authors.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package memory
+
+import (
+	"github.com/jaegertracing/jaeger/plugin/storage/memory"
+	"github.com/jaegertracing/jaeger/storage/dependencystore"
+	"github.com/jaegertracing/jaeger/storage/spanstore"
+)
+
+type storagePlugin struct {
+	store        *memory.Store
+	archiveStore *memory.Store
+}
+
+func NewStoragePlugin(mainStore *memory.Store, archiveStore *memory.Store) *storagePlugin {
+	return &storagePlugin{store: mainStore, archiveStore: archiveStore}
+}
+
+func (ns *storagePlugin) DependencyReader() dependencystore.Reader {
+	return ns.store
+}
+
+func (ns *storagePlugin) SpanReader() spanstore.Reader {
+	return ns.store
+}
+
+func (ns *storagePlugin) SpanWriter() spanstore.Writer {
+	return ns.store
+}
+
+func (ns *storagePlugin) ArchiveSpanReader() spanstore.Reader {
+	return ns.archiveStore
+}
+
+func (ns *storagePlugin) ArchiveSpanWriter() spanstore.Writer {
+	return ns.archiveStore
+}

plugin/storage/grpc/options.go

@@ -16,17 +16,23 @@ package grpc
 
 import (
 	"flag"
+	"time"
 
 	"github.com/spf13/viper"
 
+	"github.com/jaegertracing/jaeger/pkg/config/tlscfg"
 	"github.com/jaegertracing/jaeger/plugin/storage/grpc/config"
 )
 
 const (
-	pluginBinary            = "grpc-storage-plugin.binary"
-	pluginConfigurationFile = "grpc-storage-plugin.configuration-file"
-	pluginLogLevel          = "grpc-storage-plugin.log-level"
-	defaultPluginLogLevel   = "warn"
+	pluginBinary             = "grpc-storage-plugin.binary"
+	pluginConfigurationFile  = "grpc-storage-plugin.configuration-file"
+	pluginLogLevel           = "grpc-storage-plugin.log-level"
+	remotePrefix             = "grpc-storage"
+	remoteServer             = remotePrefix + ".server"
+	remoteConnectionTimeout  = remotePrefix + ".connection-timeout"
+	defaultPluginLogLevel    = "warn"
+	defaultConnectionTimeout = time.Duration(5 * time.Second)
 )
 
 // Options contains GRPC plugins configs and provides the ability
@@ -35,16 +41,30 @@ type Options struct {
 	Configuration config.Configuration `mapstructure:",squash"`
 }
 
+func tlsFlagsConfig() tlscfg.ClientFlagsConfig {
+	return tlscfg.ClientFlagsConfig{
+		Prefix: remotePrefix,
+	}
+}
+
 // AddFlags adds flags for Options
 func (opt *Options) AddFlags(flagSet *flag.FlagSet) {
+	tlsFlagsConfig().AddFlags(flagSet)
+
 	flagSet.String(pluginBinary, "", "The location of the plugin binary")
 	flagSet.String(pluginConfigurationFile, "", "A path pointing to the plugin's configuration file, made available to the plugin with the --config arg")
 	flagSet.String(pluginLogLevel, defaultPluginLogLevel, "Set the log level of the plugin's logger")
+	flagSet.String(remoteServer, "", "The remote storage gRPC server address")
+	flagSet.Duration(remoteConnectionTimeout, defaultConnectionTimeout, "The remote storage gRPC server connection timeout")
+
 }
 
 // InitFromViper initializes Options with properties from viper
 func (opt *Options) InitFromViper(v *viper.Viper) {
 	opt.Configuration.PluginBinary = v.GetString(pluginBinary)
 	opt.Configuration.PluginConfigurationFile = v.GetString(pluginConfigurationFile)
 	opt.Configuration.PluginLogLevel = v.GetString(pluginLogLevel)
+	opt.Configuration.RemoteServerAddr = v.GetString(remoteServer)
+	opt.Configuration.RemoteTLS = tlsFlagsConfig().InitFromViper(v)
+	opt.Configuration.RemoteConnectTimeout = v.GetDuration(remoteConnectionTimeout)
 }

plugin/storage/grpc/options_test.go

@@ -16,6 +16,7 @@ package grpc
 
 import (
 	"testing"
+	"time"
 
 	"github.com/stretchr/testify/assert"
 
@@ -37,3 +38,37 @@ func TestOptionsWithFlags(t *testing.T) {
 	assert.Equal(t, opts.Configuration.PluginConfigurationFile, "config.json")
 	assert.Equal(t, opts.Configuration.PluginLogLevel, "debug")
 }
+
+func TestRemoteOptionsWithFlags(t *testing.T) {
+	opts := &Options{}
+	v, command := config.Viperize(opts.AddFlags)
+	err := command.ParseFlags([]string{
+		"--grpc-storage.server=localhost:2001",
+		"--grpc-storage.tls.enabled=true",
+		"--grpc-storage.connection-timeout=60s",
+	})
+	assert.NoError(t, err)
+	opts.InitFromViper(v)
+
+	assert.Equal(t, opts.Configuration.PluginBinary, "")
+	assert.Equal(t, opts.Configuration.RemoteServerAddr, "localhost:2001")
+	assert.Equal(t, opts.Configuration.RemoteTLS.Enabled, true)
+	assert.Equal(t, opts.Configuration.RemoteConnectTimeout, 60*time.Second)
+}
+
+func TestRemoteOptionsNoTLSWithFlags(t *testing.T) {
+	opts := &Options{}
+	v, command := config.Viperize(opts.AddFlags)
+	err := command.ParseFlags([]string{
+		"--grpc-storage.server=localhost:2001",
+		"--grpc-storage.tls.enabled=false",
+		"--grpc-storage.connection-timeout=60s",
+	})
+	assert.NoError(t, err)
+	opts.InitFromViper(v)
+
+	assert.Equal(t, opts.Configuration.PluginBinary, "")
+	assert.Equal(t, opts.Configuration.RemoteServerAddr, "localhost:2001")
+	assert.Equal(t, opts.Configuration.RemoteTLS.Enabled, false)
+	assert.Equal(t, opts.Configuration.RemoteConnectTimeout, 60*time.Second)
+}

plugin/storage/grpc/shared/grpc_client.go

@@ -20,6 +20,7 @@ import (
 	"io"
 	"time"
 
+	"google.golang.org/grpc"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/status"
@@ -53,6 +54,17 @@ type grpcClient struct {
 	depsReaderClient    storage_v1.DependenciesReaderPluginClient
 }
 
+func NewGRPCClient(c *grpc.ClientConn) *grpcClient {
+	return &grpcClient{
+		readerClient:        storage_v1.NewSpanReaderPluginClient(c),
+		writerClient:        storage_v1.NewSpanWriterPluginClient(c),
+		archiveReaderClient: storage_v1.NewArchiveSpanReaderPluginClient(c),
+		archiveWriterClient: storage_v1.NewArchiveSpanWriterPluginClient(c),
+		capabilitiesClient:  storage_v1.NewPluginCapabilitiesClient(c),
+		depsReaderClient:    storage_v1.NewDependenciesReaderPluginClient(c),
+	}
+}
+
 // ContextUpgradeFunc is a functional type that can be composed to upgrade context
 type ContextUpgradeFunc func(ctx context.Context) context.Context