Move all tchannel packages to a single top level package

cmd/agent/app/builder.go

@@ -29,8 +29,6 @@ import (
 	"github.com/jaegertracing/jaeger/cmd/agent/app/httpserver"
 	"github.com/jaegertracing/jaeger/cmd/agent/app/processors"
 	"github.com/jaegertracing/jaeger/cmd/agent/app/reporter"
-	"github.com/jaegertracing/jaeger/cmd/agent/app/reporter/grpc"
-	"github.com/jaegertracing/jaeger/cmd/agent/app/reporter/tchannel"
 	"github.com/jaegertracing/jaeger/cmd/agent/app/servers"
 	"github.com/jaegertracing/jaeger/cmd/agent/app/servers/thriftudp"
 	"github.com/jaegertracing/jaeger/ports"
@@ -214,28 +212,24 @@ func defaultInt(value int, defaultVal int) int {
 	return value
 }
 
+// ProxyBuilderOptions holds config for CollectorProxyBuilder
+type ProxyBuilderOptions struct {
+	reporter.Options
+	Logger  *zap.Logger
+	Metrics metrics.Factory
+}
+
+// CollectorProxyBuilder is a func which builds CollectorProxy.
+type CollectorProxyBuilder func(ProxyBuilderOptions) (CollectorProxy, error)
+
 // CreateCollectorProxy creates collector proxy
 func CreateCollectorProxy(
-	opts *reporter.Options,
-	tchanBuilder *tchannel.Builder,
-	grpcBuilder *grpc.ConnBuilder,
-	logger *zap.Logger,
-	mFactory metrics.Factory,
+	opts ProxyBuilderOptions,
+	builders map[reporter.Type]CollectorProxyBuilder,
 ) (CollectorProxy, error) {
-	// GRPC type is set as default in viper, but we check for legacy flags
-	// to keep backward compatibility
-	if opts.ReporterType == reporter.GRPC &&
-		len(tchanBuilder.CollectorHostPorts) > 0 &&
-		len(grpcBuilder.CollectorHostPorts) == 0 {
-		logger.Warn("Using deprecated configuration", zap.String("option", "--collector-host.port"))
-		return tchannel.NewCollectorProxy(tchanBuilder, mFactory, logger)
-	}
-	switch opts.ReporterType {
-	case reporter.GRPC:
-		return grpc.NewCollectorProxy(grpcBuilder, opts.AgentTags, mFactory, logger)
-	case reporter.TCHANNEL:
-		return tchannel.NewCollectorProxy(tchanBuilder, mFactory, logger)
-	default:
+	builder, ok := builders[opts.ReporterType]
+	if !ok {
 		return nil, fmt.Errorf("unknown reporter type %s", string(opts.ReporterType))
 	}
+	return builder(opts)
 }

cmd/agent/app/builder_test.go

@@ -35,7 +35,8 @@ import (
 	"github.com/jaegertracing/jaeger/cmd/agent/app/configmanager"
 	"github.com/jaegertracing/jaeger/cmd/agent/app/reporter"
 	"github.com/jaegertracing/jaeger/cmd/agent/app/reporter/grpc"
-	"github.com/jaegertracing/jaeger/cmd/agent/app/reporter/tchannel"
+	"github.com/jaegertracing/jaeger/tchannel/agent/app/reporter/tchannel"
+	"github.com/jaegertracing/jaeger/tchannel/collector/app"
 	"github.com/jaegertracing/jaeger/thrift-gen/baggage"
 	"github.com/jaegertracing/jaeger/thrift-gen/jaeger"
 	"github.com/jaegertracing/jaeger/thrift-gen/sampling"
@@ -201,8 +202,8 @@ func TestCreateCollectorProxy(t *testing.T) {
 			err: "at least one collector hostPort address is required when resolver is not available",
 		},
 		{
-			flags:  []string{"--collector.host-port=foo"},
-			metric: metricstest.ExpectedMetric{Name: "reporter.batches.failures", Tags: map[string]string{"protocol": "tchannel", "format": "jaeger"}, Value: 1},
+			flags: []string{"--collector.host-port=foo"},
+			err:   "at least one collector hostPort address is required when resolver is not available",
 		},
 		{
 			flags:  []string{"--reporter.type=tchannel"},
@@ -213,12 +214,8 @@ func TestCreateCollectorProxy(t *testing.T) {
 			metric: metricstest.ExpectedMetric{Name: "reporter.batches.failures", Tags: map[string]string{"protocol": "tchannel", "format": "jaeger"}, Value: 1},
 		},
 		{
-			flags:  []string{"--reporter.type=grpc", "--collector.host-port=foo"},
-			metric: metricstest.ExpectedMetric{Name: "reporter.batches.failures", Tags: map[string]string{"protocol": "tchannel", "format": "jaeger"}, Value: 1},
-		},
-		{
-			flags:  []string{"--reporter.type=grpc", "--collector.host-port=foo"},
-			metric: metricstest.ExpectedMetric{Name: "reporter.batches.failures", Tags: map[string]string{"protocol": "tchannel", "format": "jaeger"}, Value: 1},
+			flags: []string{"--reporter.type=grpc", "--collector.host-port=foo"},
+			err:   "at least one collector hostPort address is required when resolver is not available",
 		},
 		{
 			flags:  []string{"--reporter.type=grpc", "--reporter.grpc.host-port=foo", "--collector.host-port=foo"},
@@ -235,6 +232,7 @@ func TestCreateCollectorProxy(t *testing.T) {
 		tchannel.AddFlags(flags)
 		grpc.AddFlags(flags)
 		reporter.AddFlags(flags)
+		app.AddFlags(flags)
 
 		command := cobra.Command{}
 		command.PersistentFlags().AddGoFlagSet(flags)
@@ -249,7 +247,16 @@ func TestCreateCollectorProxy(t *testing.T) {
 		grpcBuilder := grpc.NewConnBuilder().InitFromViper(v)
 
 		metricsFactory := metricstest.NewFactory(time.Microsecond)
-		proxy, err := CreateCollectorProxy(rOpts, tchan, grpcBuilder, zap.NewNop(), metricsFactory)
+
+		builders := map[reporter.Type]CollectorProxyBuilder{
+			reporter.GRPC:         GRPCCollectorProxyBuilder(grpcBuilder),
+			tchannel.ReporterType: TCollectorProxyBuilder(tchan),
+		}
+		proxy, err := CreateCollectorProxy(ProxyBuilderOptions{
+			Options: *rOpts,
+			Metrics: metricsFactory,
+			Logger:  zap.NewNop(),
+		}, builders)
 		if test.err != "" {
 			assert.EqualError(t, err, test.err)
 			assert.Nil(t, proxy)
@@ -262,11 +269,14 @@ func TestCreateCollectorProxy(t *testing.T) {
 }
 
 func TestCreateCollectorProxy_UnknownReporter(t *testing.T) {
-	rOpts := new(reporter.Options)
 	tchan := tchannel.NewBuilder()
 	grpcBuilder := grpc.NewConnBuilder()
 
-	proxy, err := CreateCollectorProxy(rOpts, tchan, grpcBuilder, zap.NewNop(), metrics.NullFactory)
+	builders := map[reporter.Type]CollectorProxyBuilder{
+		reporter.GRPC:         GRPCCollectorProxyBuilder(grpcBuilder),
+		tchannel.ReporterType: TCollectorProxyBuilder(tchan),
+	}
+	proxy, err := CreateCollectorProxy(ProxyBuilderOptions{}, builders)
 	assert.Nil(t, proxy)
 	assert.EqualError(t, err, "unknown reporter type ")
 }

cmd/agent/app/processors/thrift_processor_test.go

@@ -28,10 +28,10 @@ import (
 	"go.uber.org/zap"
 
 	"github.com/jaegertracing/jaeger/cmd/agent/app/reporter"
-	tchreporter "github.com/jaegertracing/jaeger/cmd/agent/app/reporter/tchannel"
 	"github.com/jaegertracing/jaeger/cmd/agent/app/servers"
 	"github.com/jaegertracing/jaeger/cmd/agent/app/servers/thriftudp"
 	"github.com/jaegertracing/jaeger/cmd/agent/app/testutils"
+	tchreporter "github.com/jaegertracing/jaeger/tchannel/agent/app/reporter/tchannel"
 	"github.com/jaegertracing/jaeger/thrift-gen/agent"
 	"github.com/jaegertracing/jaeger/thrift-gen/jaeger"
 	"github.com/jaegertracing/jaeger/thrift-gen/zipkincore"

cmd/agent/app/proxy_builders.go

@@ -0,0 +1,34 @@
+// 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 app
+
+import (
+	"github.com/jaegertracing/jaeger/cmd/agent/app/reporter/grpc"
+	"github.com/jaegertracing/jaeger/tchannel/agent/app/reporter/tchannel"
+)
+
+// GRPCCollectorProxyBuilder creates CollectorProxyBuilder for GRPC reporter
+func GRPCCollectorProxyBuilder(builder *grpc.ConnBuilder) CollectorProxyBuilder {
+	return func(opts ProxyBuilderOptions) (proxy CollectorProxy, err error) {
+		return grpc.NewCollectorProxy(builder, opts.AgentTags, opts.Metrics, opts.Logger)
+	}
+}
+
+// TCollectorProxyBuilder creates CollectorProxyBuilder for Tchannel reporter
+func TCollectorProxyBuilder(builder *tchannel.Builder) CollectorProxyBuilder {
+	return func(opts ProxyBuilderOptions) (proxy CollectorProxy, err error) {
+		return tchannel.NewCollectorProxy(builder, opts.Metrics, opts.Logger)
+	}
+}

cmd/agent/app/reporter/flags.go

@@ -31,8 +31,6 @@ const (
 	// Agent tags
 	agentTagsDeprecated = "jaeger.tags"
 	agentTags           = "agent.tags"
-	// TCHANNEL is name of tchannel reporter.
-	TCHANNEL Type = "tchannel"
 	// GRPC is name of gRPC reporter.
 	GRPC Type = "grpc"
 )
@@ -48,7 +46,7 @@ type Options struct {
 
 // AddFlags adds flags for Options.
 func AddFlags(flags *flag.FlagSet) {
-	flags.String(reporterType, string(GRPC), fmt.Sprintf("Reporter type to use e.g. %s, %s[%s]", string(GRPC), string(TCHANNEL), "NOTE: Deprecated since 1.16"))
+	flags.String(reporterType, string(GRPC), fmt.Sprintf("Reporter type to use e.g. %s", string(GRPC)))
 	if !setupcontext.IsAllInOne() {
 		flags.String(agentTagsDeprecated, "", "(deprecated) see --"+agentTags)
 		flags.String(agentTags, "", "One or more tags to be added to the Process tags of all spans passing through this agent. Ex: key1=value1,key2=${envVar:defaultValue}")
@@ -58,9 +56,6 @@ func AddFlags(flags *flag.FlagSet) {
 // InitFromViper initializes Options with properties retrieved from Viper.
 func (b *Options) InitFromViper(v *viper.Viper, logger *zap.Logger) *Options {
 	b.ReporterType = Type(v.GetString(reporterType))
-	if b.ReporterType == TCHANNEL {
-		logger.Warn("Using deprecated reporter type", zap.Any(reporterType, TCHANNEL))
-	}
 	if !setupcontext.IsAllInOne() {
 		if len(v.GetString(agentTagsDeprecated)) > 0 {
 			logger.Warn("Using deprecated configuration", zap.String("option", agentTagsDeprecated))

cmd/agent/main.go

@@ -28,12 +28,12 @@ import (
 	"github.com/jaegertracing/jaeger/cmd/agent/app"
 	"github.com/jaegertracing/jaeger/cmd/agent/app/reporter"
 	"github.com/jaegertracing/jaeger/cmd/agent/app/reporter/grpc"
-	"github.com/jaegertracing/jaeger/cmd/agent/app/reporter/tchannel"
 	"github.com/jaegertracing/jaeger/cmd/docs"
 	"github.com/jaegertracing/jaeger/cmd/flags"
 	"github.com/jaegertracing/jaeger/pkg/config"
 	"github.com/jaegertracing/jaeger/pkg/version"
 	"github.com/jaegertracing/jaeger/ports"
+	"github.com/jaegertracing/jaeger/tchannel/agent/app/reporter/tchannel"
 )
 
 func main() {
@@ -57,7 +57,15 @@ func main() {
 			rOpts := new(reporter.Options).InitFromViper(v, logger)
 			tchanBuilder := tchannel.NewBuilder().InitFromViper(v, logger)
 			grpcBuilder := grpc.NewConnBuilder().InitFromViper(v)
-			cp, err := app.CreateCollectorProxy(rOpts, tchanBuilder, grpcBuilder, logger, mFactory)
+			builders := map[reporter.Type]app.CollectorProxyBuilder{
+				reporter.GRPC:         app.GRPCCollectorProxyBuilder(grpcBuilder),
+				tchannel.ReporterType: app.TCollectorProxyBuilder(tchanBuilder),
+			}
+			cp, err := app.CreateCollectorProxy(app.ProxyBuilderOptions{
+				Options: *rOpts,
+				Logger:  logger,
+				Metrics: mFactory,
+			}, builders)
 			if err != nil {
 				logger.Fatal("Could not create collector proxy", zap.Error(err))
 			}

cmd/all-in-one/main.go

@@ -33,7 +33,6 @@ import (
 	agentApp "github.com/jaegertracing/jaeger/cmd/agent/app"
 	agentRep "github.com/jaegertracing/jaeger/cmd/agent/app/reporter"
 	agentGrpcRep "github.com/jaegertracing/jaeger/cmd/agent/app/reporter/grpc"
-	agentTchanRep "github.com/jaegertracing/jaeger/cmd/agent/app/reporter/tchannel"
 	"github.com/jaegertracing/jaeger/cmd/all-in-one/setupcontext"
 	collectorApp "github.com/jaegertracing/jaeger/cmd/collector/app"
 	"github.com/jaegertracing/jaeger/cmd/docs"
@@ -50,6 +49,8 @@ import (
 	"github.com/jaegertracing/jaeger/storage/dependencystore"
 	"github.com/jaegertracing/jaeger/storage/spanstore"
 	storageMetrics "github.com/jaegertracing/jaeger/storage/spanstore/metrics"
+	agentTchanRep "github.com/jaegertracing/jaeger/tchannel/agent/app/reporter/tchannel"
+	tCollector "github.com/jaegertracing/jaeger/tchannel/collector/app"
 )
 
 // all-in-one/main is a standalone full-stack jaeger backend, backed by a memory store
@@ -130,11 +131,24 @@ by default uses only in-memory database.`,
 				HealthCheck:    svc.HC(),
 			})
 			c.Start(cOpts)
+			tCollectorOpts := new(tCollector.Options).InitFromViper(v)
+			tc, err := tCollector.Start("jaeger-collector", tCollectorOpts, logger, c.SpanHandlers(), strategyStore)
+			if err != nil {
+				logger.Fatal("Could not start Tchannel thrift collector", zap.Error(err))
+			}
 
 			// agent
 			grpcBuilder.CollectorHostPorts = append(grpcBuilder.CollectorHostPorts, fmt.Sprintf("127.0.0.1:%d", cOpts.CollectorGRPCPort))
 			agentMetricsFactory := metricsFactory.Namespace(metrics.NSOptions{Name: "agent", Tags: nil})
-			cp, err := agentApp.CreateCollectorProxy(repOpts, tchanBuilder, grpcBuilder, logger, agentMetricsFactory)
+			builders := map[agentRep.Type]agentApp.CollectorProxyBuilder{
+				agentRep.GRPC:              agentApp.GRPCCollectorProxyBuilder(grpcBuilder),
+				agentTchanRep.ReporterType: agentApp.TCollectorProxyBuilder(tchanBuilder),
+			}
+			cp, err := agentApp.CreateCollectorProxy(agentApp.ProxyBuilderOptions{
+				Options: *repOpts,
+				Logger:  logger,
+				Metrics: agentMetricsFactory,
+			}, builders)
 			if err != nil {
 				logger.Fatal("Could not create collector proxy", zap.Error(err))
 			}
@@ -151,6 +165,7 @@ by default uses only in-memory database.`,
 				agent.Stop()
 				cp.Close()
 				c.Close()
+				tc.Close()
 				querySrv.Close()
 				if closer, ok := spanWriter.(io.Closer); ok {
 					err := closer.Close()
@@ -178,6 +193,7 @@ by default uses only in-memory database.`,
 		agentTchanRep.AddFlags,
 		agentGrpcRep.AddFlags,
 		collectorApp.AddFlags,
+		tCollector.AddFlags,
 		queryApp.AddFlags,
 		strategyStoreFactory.AddFlags,
 	)

cmd/collector/app/builder_flags.go

@@ -29,7 +29,6 @@ const (
 	collectorDynQueueSizeMemory   = "collector.queue-size-memory"
 	collectorQueueSize            = "collector.queue-size"
 	collectorNumWorkers           = "collector.num-workers"
-	collectorPort                 = "collector.port"
 	collectorHTTPPort             = "collector.http-port"
 	collectorGRPCPort             = "collector.grpc-port"
 	collectorTags                 = "collector.tags"
@@ -52,8 +51,6 @@ type CollectorOptions struct {
 	QueueSize int
 	// NumWorkers is the number of internal workers in a collector
 	NumWorkers int
-	// CollectorPort is the port that the collector service listens in on for tchannel requests
-	CollectorPort int
 	// CollectorHTTPPort is the port that the collector service listens in on for http requests
 	CollectorHTTPPort int
 	// CollectorGRPCPort is the port that the collector service listens in on for gRPC requests
@@ -75,7 +72,6 @@ func AddFlags(flags *flag.FlagSet) {
 	flags.Uint(collectorDynQueueSizeMemory, 0, "(experimental) The max memory size in MiB to use for the dynamic queue.")
 	flags.Int(collectorQueueSize, DefaultQueueSize, "The queue size of the collector")
 	flags.Int(collectorNumWorkers, DefaultNumWorkers, "The number of workers pulling items from the queue")
-	flags.Int(collectorPort, ports.CollectorTChannel, "The TChannel port for the collector service")
 	flags.Int(collectorHTTPPort, ports.CollectorHTTP, "The HTTP port for the collector service")
 	flags.Int(collectorGRPCPort, ports.CollectorGRPC, "The gRPC port for the collector service")
 	flags.String(collectorTags, "", "One or more tags to be added to the Process tags of all spans passing through this collector. Ex: key1=value1,key2=${envVar:defaultValue}")
@@ -90,7 +86,6 @@ func (cOpts *CollectorOptions) InitFromViper(v *viper.Viper) *CollectorOptions {
 	cOpts.DynQueueSizeMemory = v.GetUint(collectorDynQueueSizeMemory) * 1024 * 1024 // we receive in MiB and store in bytes
 	cOpts.QueueSize = v.GetInt(collectorQueueSize)
 	cOpts.NumWorkers = v.GetInt(collectorNumWorkers)
-	cOpts.CollectorPort = v.GetInt(collectorPort)
 	cOpts.CollectorHTTPPort = v.GetInt(collectorHTTPPort)
 	cOpts.CollectorGRPCPort = v.GetInt(collectorGRPCPort)
 	cOpts.CollectorTags = flags.ParseJaegerTags(v.GetString(collectorTags))

cmd/collector/app/collector.go

@@ -20,7 +20,6 @@ import (
 	"time"
 
 	"github.com/uber/jaeger-lib/metrics"
-	"github.com/uber/tchannel-go"
 	"go.uber.org/zap"
 	"google.golang.org/grpc"
 
@@ -47,7 +46,6 @@ type Collector struct {
 	hServer    *http.Server
 	zkServer   *http.Server
 	grpcServer *grpc.Server
-	tchServer  *tchannel.Channel
 }
 
 // CollectorParams to construct a new Jaeger Collector.
@@ -84,19 +82,6 @@ func (c *Collector) Start(builderOpts *CollectorOptions) error {
 	c.spanProcessor = handlerBuilder.BuildSpanProcessor()
 	c.spanHandlers = handlerBuilder.BuildHandlers(c.spanProcessor)
 
-	if tchServer, err := server.StartThriftServer(&server.ThriftServerParams{
-		ServiceName:          c.serviceName,
-		Port:                 builderOpts.CollectorPort,
-		JaegerBatchesHandler: c.spanHandlers.JaegerBatchesHandler,
-		ZipkinSpansHandler:   c.spanHandlers.ZipkinSpansHandler,
-		StrategyStore:        c.strategyStore,
-		Logger:               c.logger,
-	}); err != nil {
-		c.logger.Fatal("could not start Thrift collector", zap.Error(err))
-	} else {
-		c.tchServer = tchServer
-	}
-
 	if grpcServer, err := server.StartGRPCServer(&server.GRPCServerParams{
 		Port:          builderOpts.CollectorGRPCPort,
 		Handler:       c.spanHandlers.GRPCHandler,
@@ -144,11 +129,6 @@ func (c *Collector) Close() error {
 		c.grpcServer.GracefulStop()
 	}
 
-	// TChannel server
-	if c.tchServer != nil {
-		c.tchServer.Close()
-	}
-
 	// HTTP server
 	if c.hServer != nil {
 		timeout, cancel := context.WithTimeout(context.Background(), 5*time.Second)
@@ -175,3 +155,8 @@ func (c *Collector) Close() error {
 
 	return nil
 }
+
+// SpanHandlers returns span handlers used by the Collector.
+func (c *Collector) SpanHandlers() *SpanHandlers {
+	return c.spanHandlers
+}