Normalize CLI flags to use host:port addresses

cmd/all-in-one/main.go

@@ -22,7 +22,6 @@ import (
 	"net"
 	"net/http"
 	"os"
-	"strconv"
 
 	"github.com/gorilla/mux"
 	"github.com/opentracing/opentracing-go"
@@ -185,7 +184,8 @@ func startAgent(
 ) {
 	metricsFactory := baseFactory.Namespace(metrics.NSOptions{Name: "agent", Tags: nil})
 
-	grpcBuilder.CollectorHostPorts = append(grpcBuilder.CollectorHostPorts, fmt.Sprintf("127.0.0.1:%d", cOpts.CollectorGRPCPort))
+	addr := getAddressFromCLIOptions(cOpts.CollectorGRPCPort, cOpts.CollectorGRPCHostPort, "collector.grpc-port", logger)
+	grpcBuilder.CollectorHostPorts = append(grpcBuilder.CollectorHostPorts, addr)
 	cp, err := agentApp.CreateCollectorProxy(repOpts, tchanBuilder, grpcBuilder, logger, metricsFactory)
 	if err != nil {
 		logger.Fatal("Could not create collector proxy", zap.Error(err))
@@ -234,16 +234,18 @@ func startCollector(
 		server.Register(jc.NewTChanCollectorServer(batchHandler))
 		server.Register(zc.NewTChanZipkinCollectorServer(batchHandler))
 		server.Register(sc.NewTChanSamplingManagerServer(sampling.NewHandler(strategyStore)))
-		portStr := ":" + strconv.Itoa(cOpts.CollectorPort)
-		listener, err := net.Listen("tcp", portStr)
+
+		tchanAddr := getAddressFromCLIOptions(cOpts.CollectorPort, cOpts.CollectorTChanHostPort, "collector.http-port", logger)
+		listener, err := net.Listen("tcp", tchanAddr)
 		if err != nil {
 			logger.Fatal("Unable to start listening on channel", zap.Error(err))
 		}
-		logger.Info("Starting jaeger-collector TChannel server", zap.Int("port", cOpts.CollectorPort))
+		logger.Info("Starting jaeger-collector TChannel server", zap.String("addr", tchanAddr))
 		ch.Serve(listener)
 	}
 
-	server, err := startGRPCServer(cOpts.CollectorGRPCPort, grpcHandler, strategyStore, logger)
+	grpcAddr := getAddressFromCLIOptions(cOpts.CollectorGRPCPort, cOpts.CollectorGRPCHostPort, "collector.grpc-port", logger)
+	server, err := startGRPCServer(grpcAddr, grpcHandler, strategyStore, logger)
 	if err != nil {
 		logger.Fatal("Could not start gRPC collector", zap.Error(err))
 	}
@@ -252,14 +254,16 @@ func startCollector(
 		r := mux.NewRouter()
 		apiHandler := collectorApp.NewAPIHandler(jaegerBatchesHandler)
 		apiHandler.RegisterRoutes(r)
-		httpPortStr := ":" + strconv.Itoa(cOpts.CollectorHTTPPort)
 		recoveryHandler := recoveryhandler.NewRecoveryHandler(logger, true)
 
-		go startZipkinHTTPAPI(logger, cOpts.CollectorZipkinHTTPPort, zipkinSpansHandler, recoveryHandler)
+		zipkinAddr := getAddressFromCLIOptions(cOpts.CollectorZipkinHTTPPort, cOpts.CollectorZipkinHTTPHostPort, "collector.zipkin.http-port", logger)
+
+		go startZipkinHTTPAPI(logger, zipkinAddr, zipkinSpansHandler, recoveryHandler)
 
-		logger.Info("Starting jaeger-collector HTTP server", zap.Int("http-port", cOpts.CollectorHTTPPort))
+		httpAddr := getAddressFromCLIOptions(cOpts.CollectorHTTPPort, cOpts.CollectorHTTPHostPort, "collector.http-port", logger)
+		logger.Info("Starting jaeger-collector HTTP server", zap.String("http-addr", httpAddr))
 		go func() {
-			if err := http.ListenAndServe(httpPortStr, recoveryHandler(r)); err != nil {
+			if err := http.ListenAndServe(httpAddr, recoveryHandler(r)); err != nil {
 				logger.Fatal("Could not launch jaeger-collector HTTP server", zap.Error(err))
 			}
 			hc.Set(healthcheck.Unavailable)
@@ -269,13 +273,13 @@ func startCollector(
 }
 
 func startGRPCServer(
-	port int,
+	addr string,
 	handler *collectorApp.GRPCHandler,
 	samplingStore strategystore.StrategyStore,
 	logger *zap.Logger,
 ) (*grpc.Server, error) {
 	server := grpc.NewServer()
-	_, err := grpcserver.StartGRPCCollector(port, server, handler, samplingStore, logger, func(err error) {
+	_, err := grpcserver.StartGRPCCollector(addr, server, handler, samplingStore, logger, func(err error) {
 		logger.Fatal("gRPC collector failed", zap.Error(err))
 	})
 	if err != nil {
@@ -286,18 +290,17 @@ func startGRPCServer(
 
 func startZipkinHTTPAPI(
 	logger *zap.Logger,
-	zipkinPort int,
+	zipkinAddr string,
 	zipkinSpansHandler collectorApp.ZipkinSpansHandler,
 	recoveryHandler func(http.Handler) http.Handler,
 ) {
-	if zipkinPort != 0 {
+	if zipkinAddr != "" {
 		r := mux.NewRouter()
 		zHandler := zipkin.NewAPIHandler(zipkinSpansHandler)
 		zHandler.RegisterRoutes(r)
-		httpPortStr := ":" + strconv.Itoa(zipkinPort)
-		logger.Info("Listening for Zipkin HTTP traffic", zap.Int("zipkin.http-port", zipkinPort))
+		logger.Info("Listening for Zipkin HTTP traffic", zap.String("zipkin.http-addr", zipkinAddr))
 
-		if err := http.ListenAndServe(httpPortStr, recoveryHandler(r)); err != nil {
+		if err := http.ListenAndServe(zipkinAddr, recoveryHandler(r)); err != nil {
 			logger.Fatal("Could not launch service", zap.Error(err))
 		}
 	}
@@ -362,3 +365,13 @@ func initTracer(metricsFactory metrics.Factory, logger *zap.Logger) io.Closer {
 	opentracing.SetGlobalTracer(tracer)
 	return closer
 }
+
+// Utility function to decide listening address based on port (deprecated flags) or host:port (new flags)
+func getAddressFromCLIOptions(port int, addr string, deprecatedFlag string, logger *zap.Logger) string {
+	if port != 0 {
+		logger.Warn("Using deprecated configuration", zap.String("option", deprecatedFlag))
+		return ports.PortToHostPort(port)
+	}
+
+	return addr
+}

cmd/collector/app/builder/builder_flags.go

@@ -30,11 +30,15 @@ const (
 	collectorPort                 = "collector.port"
 	collectorHTTPPort             = "collector.http-port"
 	collectorGRPCPort             = "collector.grpc-port"
+	collectorTChanHostPort        = "collector.tchan-server.host-port"
+	collectorHTTPHostPort         = "collector.http-server.host-port"
+	collectorGRPCHostPort         = "collector.grpc-server.host-port"
 	collectorGRPCTLS              = "collector.grpc.tls"
 	collectorGRPCCert             = "collector.grpc.tls.cert"
 	collectorGRPCKey              = "collector.grpc.tls.key"
 	collectorGRPCClientCA         = "collector.grpc.tls.client.ca"
-	collectorZipkinHTTPort        = "collector.zipkin.http-port"
+	collectorZipkinHTTPPort       = "collector.zipkin.http-port"
+	collectorZipkinHTTPHostPort   = "collector.zipkin.host-port"
 	collectorZipkinAllowedOrigins = "collector.zipkin.allowed-origins"
 	collectorZipkinAllowedHeaders = "collector.zipkin.allowed-headers"
 )
@@ -51,6 +55,12 @@ type CollectorOptions struct {
 	CollectorHTTPPort int
 	// CollectorGRPCPort is the port that the collector service listens in on for gRPC requests
 	CollectorGRPCPort int
+	// CollectorTChanHostPort is the host:port address that the collector service listens in on for tchannel requests
+	CollectorTChanHostPort string
+	// CollectorHTTPHostPort is the host:port address that the collector service listens in on for http requests
+	CollectorHTTPHostPort string
+	// CollectorGRPCHostPort is the host:port address that the collector service listens in on for gRPC requests
+	CollectorGRPCHostPort string
 	// CollectorGRPCTLS defines if the server is setup with TLS
 	CollectorGRPCTLS bool
 	// CollectorGRPCCert is the path to a TLS certificate file for the server
@@ -61,6 +71,8 @@ type CollectorOptions struct {
 	CollectorGRPCKey string
 	// CollectorZipkinHTTPPort is the port that the Zipkin collector service listens in on for http requests
 	CollectorZipkinHTTPPort int
+	// CollectorZipkinHTTPHostPort is the host:port address that the Zipkin collector service listens in on for http requests
+	CollectorZipkinHTTPHostPort string
 	// CollectorZipkinAllowedOrigins is a list of origins a cross-domain request to the Zipkin collector service can be executed from
 	CollectorZipkinAllowedOrigins string
 	// CollectorZipkinAllowedHeaders is a list of headers that the Zipkin collector service allowes the client to use with cross-domain requests
@@ -71,11 +83,15 @@ type CollectorOptions struct {
 func AddFlags(flags *flag.FlagSet) {
 	flags.Int(collectorQueueSize, app.DefaultQueueSize, "The queue size of the collector")
 	flags.Int(collectorNumWorkers, app.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.Int(collectorZipkinHTTPort, 0, "The HTTP port for the Zipkin collector service e.g. 9411")
-	flags.Bool(collectorGRPCTLS, false, "Enable TLS for the gRPC collector port")
+	flags.Int(collectorPort, ports.CollectorTChannel, "The TChannel port for the collector service - (deprecated) see - "+collectorTChanHostPort)
+	flags.Int(collectorHTTPPort, ports.CollectorHTTP, "The HTTP port for the collector service - (deprecated) see -"+collectorHTTPHostPort)
+	flags.Int(collectorGRPCPort, ports.CollectorGRPC, "The gRPC port for the collector service - (deprecated) see -"+collectorGRPCHostPort)
+	flags.Int(collectorZipkinHTTPPort, 0, "The HTTP port for the Zipkin collector service e.g. 9411 - (deprecated) see -"+collectorZipkinHTTPHostPort)
+	flags.String(collectorTChanHostPort, ports.PortToHostPort(ports.CollectorTChannel), "The host:port (e.g. 127.0.0.1:5555 or :5555) of the collector's TChannel server")
+	flags.String(collectorHTTPHostPort, ports.PortToHostPort(ports.CollectorHTTP), "The host:port (e.g. 127.0.0.1:5555 or :5555) of the collector's HTTP server")
+	flags.String(collectorGRPCHostPort, ports.PortToHostPort(ports.CollectorGRPC), "The host:port (e.g. 127.0.0.1:5555 or :5555) of the collector's GRPC server")
+	flags.String(collectorZipkinHTTPHostPort, ports.PortToHostPort(0), "The host:port (e.g. 127.0.0.1:5555 or :5555) of the collector's Zipkin server")
+	flags.Bool(collectorGRPCTLS, false, "Enable TLS for the gRPC collector endpoint")
 	flags.String(collectorGRPCCert, "", "Path to TLS certificate for the gRPC collector TLS service")
 	flags.String(collectorGRPCKey, "", "Path to TLS key for the gRPC collector TLS cert")
 	flags.String(collectorGRPCClientCA, "", "Path to a TLS CA to verify certificates presented by clients (if unset, all clients are permitted)")
@@ -90,11 +106,15 @@ func (cOpts *CollectorOptions) InitFromViper(v *viper.Viper) *CollectorOptions {
 	cOpts.CollectorPort = v.GetInt(collectorPort)
 	cOpts.CollectorHTTPPort = v.GetInt(collectorHTTPPort)
 	cOpts.CollectorGRPCPort = v.GetInt(collectorGRPCPort)
+	cOpts.CollectorTChanHostPort = v.GetString(collectorTChanHostPort)
+	cOpts.CollectorHTTPHostPort = v.GetString(collectorHTTPHostPort)
+	cOpts.CollectorGRPCHostPort = v.GetString(collectorGRPCHostPort)
 	cOpts.CollectorGRPCTLS = v.GetBool(collectorGRPCTLS)
 	cOpts.CollectorGRPCCert = v.GetString(collectorGRPCCert)
 	cOpts.CollectorGRPCClientCA = v.GetString(collectorGRPCClientCA)
 	cOpts.CollectorGRPCKey = v.GetString(collectorGRPCKey)
-	cOpts.CollectorZipkinHTTPPort = v.GetInt(collectorZipkinHTTPort)
+	cOpts.CollectorZipkinHTTPPort = v.GetInt(collectorZipkinHTTPPort)
+	cOpts.CollectorZipkinHTTPHostPort = v.GetString(collectorZipkinHTTPHostPort)
 	cOpts.CollectorZipkinAllowedOrigins = v.GetString(collectorZipkinAllowedOrigins)
 	cOpts.CollectorZipkinAllowedHeaders = v.GetString(collectorZipkinAllowedHeaders)
 	return cOpts

cmd/collector/app/grpcserver/grpc_server.go

@@ -71,15 +71,14 @@ func TLSConfig(cert, key, clientCA string) (*tls.Config, error) {
 
 // StartGRPCCollector configures and starts gRPC endpoints exposed by collector.
 func StartGRPCCollector(
-	port int,
+	addr string,
 	server *grpc.Server,
 	handler *app.GRPCHandler,
 	samplingStrategy strategystore.StrategyStore,
 	logger *zap.Logger,
 	serveErr func(error),
 ) (net.Addr, error) {
-	grpcPortStr := ":" + strconv.Itoa(port)
-	lis, err := net.Listen("tcp", grpcPortStr)
+	lis, err := net.Listen("tcp", addr)
 	if err != nil {
 		return nil, errors.Wrap(err, "failed to listen on gRPC port")
 	}

cmd/collector/app/grpcserver/grpc_server_test.go

@@ -29,6 +29,7 @@ import (
 
 	"github.com/jaegertracing/jaeger/cmd/collector/app"
 	"github.com/jaegertracing/jaeger/model"
+	"github.com/jaegertracing/jaeger/ports"
 	"github.com/jaegertracing/jaeger/proto-gen/api_v2"
 	"github.com/jaegertracing/jaeger/thrift-gen/sampling"
 )
@@ -38,8 +39,8 @@ func TestFailToListen(t *testing.T) {
 	l, _ := zap.NewDevelopment()
 	handler := app.NewGRPCHandler(l, &mockSpanProcessor{})
 	server := grpc.NewServer()
-	const invalidPort = -1
-	addr, err := StartGRPCCollector(invalidPort, server, handler, &mockSamplingStore{}, l, func(e error) {
+	const invalidAddr = ":-1"
+	addr, err := StartGRPCCollector(invalidAddr, server, handler, &mockSamplingStore{}, l, func(e error) {
 	})
 	assert.Nil(t, addr)
 	assert.EqualError(t, err, "failed to listen on gRPC port: listen tcp: address -1: invalid port")
@@ -63,7 +64,7 @@ func TestSpanCollector(t *testing.T) {
 	l, _ := zap.NewDevelopment()
 	handler := app.NewGRPCHandler(l, &mockSpanProcessor{})
 	server := grpc.NewServer()
-	addr, err := StartGRPCCollector(0, server, handler, &mockSamplingStore{}, l, func(e error) {
+	addr, err := StartGRPCCollector(ports.PortToHostPort(0), server, handler, &mockSamplingStore{}, l, func(e error) {
 	})
 	require.NoError(t, err)
 

cmd/collector/main.go

@@ -22,7 +22,6 @@ import (
 	"net"
 	"net/http"
 	"os"
-	"strconv"
 	"strings"
 
 	"github.com/gorilla/mux"
@@ -120,12 +119,13 @@ func main() {
 				server.Register(jc.NewTChanCollectorServer(batchHandler))
 				server.Register(zc.NewTChanZipkinCollectorServer(batchHandler))
 				server.Register(sc.NewTChanSamplingManagerServer(sampling.NewHandler(strategyStore)))
-				portStr := ":" + strconv.Itoa(builderOpts.CollectorPort)
-				listener, err := net.Listen("tcp", portStr)
+
+				addr := getAddressFromCLIOptions(builderOpts.CollectorPort, builderOpts.CollectorTChanHostPort, "collector.http-port", logger)
+				listener, err := net.Listen("tcp", addr)
 				if err != nil {
 					logger.Fatal("Unable to start listening on channel", zap.Error(err))
 				}
-				logger.Info("Starting jaeger-collector TChannel server", zap.Int("port", builderOpts.CollectorPort))
+				logger.Info("Starting jaeger-collector TChannel server", zap.String("tchan-addr", addr))
 				ch.Serve(listener)
 			}
 
@@ -138,15 +138,17 @@ func main() {
 				r := mux.NewRouter()
 				apiHandler := app.NewAPIHandler(jaegerBatchesHandler)
 				apiHandler.RegisterRoutes(r)
-				httpPortStr := ":" + strconv.Itoa(builderOpts.CollectorHTTPPort)
 				recoveryHandler := recoveryhandler.NewRecoveryHandler(logger, true)
 				httpHandler := recoveryHandler(r)
 
-				go startZipkinHTTPAPI(logger, builderOpts.CollectorZipkinHTTPPort, builderOpts.CollectorZipkinAllowedOrigins, builderOpts.CollectorZipkinAllowedHeaders, zipkinSpansHandler, recoveryHandler)
+				zipkinAddr := getAddressFromCLIOptions(builderOpts.CollectorZipkinHTTPPort, builderOpts.CollectorZipkinHTTPHostPort, "collector.zipkin.http-port", logger)
+
+				go startZipkinHTTPAPI(logger, zipkinAddr, builderOpts.CollectorZipkinAllowedOrigins, builderOpts.CollectorZipkinAllowedHeaders, zipkinSpansHandler, recoveryHandler)
 
-				logger.Info("Starting jaeger-collector HTTP server", zap.Int("http-port", builderOpts.CollectorHTTPPort))
+				httpAddr := getAddressFromCLIOptions(builderOpts.CollectorHTTPPort, builderOpts.CollectorHTTPHostPort, "collector.http-port", logger)
+				logger.Info("Starting jaeger-collector HTTP server", zap.String("http-addr", httpAddr))
 				go func() {
-					if err := http.ListenAndServe(httpPortStr, httpHandler); err != nil {
+					if err := http.ListenAndServe(httpAddr, httpHandler); err != nil {
 						logger.Fatal("Could not launch service", zap.Error(err))
 					}
 					svc.HC().Set(healthcheck.Unavailable)
@@ -207,7 +209,9 @@ func startGRPCServer(
 	} else { // server without TLS
 		server = grpc.NewServer()
 	}
-	_, err := grpcserver.StartGRPCCollector(opts.CollectorGRPCPort, server, handler, samplingStore, logger, func(err error) {
+
+	addr := getAddressFromCLIOptions(opts.CollectorGRPCPort, opts.CollectorGRPCHostPort, "collector.grpc-port", logger)
+	_, err := grpcserver.StartGRPCCollector(addr, server, handler, samplingStore, logger, func(err error) {
 		logger.Fatal("gRPC collector failed", zap.Error(err))
 	})
 	if err != nil {
@@ -218,13 +222,13 @@ func startGRPCServer(
 
 func startZipkinHTTPAPI(
 	logger *zap.Logger,
-	zipkinPort int,
+	zipkinAddr string,
 	allowedOrigins string,
 	allowedHeaders string,
 	zipkinSpansHandler app.ZipkinSpansHandler,
 	recoveryHandler func(http.Handler) http.Handler,
 ) {
-	if zipkinPort != 0 {
+	if zipkinAddr != "" {
 		zHandler := zipkin.NewAPIHandler(zipkinSpansHandler)
 		r := mux.NewRouter()
 		zHandler.RegisterRoutes(r)
@@ -238,10 +242,9 @@ func startZipkinHTTPAPI(
 			AllowedHeaders: headers,
 		})
 
-		httpPortStr := ":" + strconv.Itoa(zipkinPort)
-		logger.Info("Listening for Zipkin HTTP traffic", zap.Int("zipkin.http-port", zipkinPort))
+		logger.Info("Listening for Zipkin HTTP traffic", zap.String("zipkin.http-addr", zipkinAddr))
 
-		if err := http.ListenAndServe(httpPortStr, c.Handler(recoveryHandler(r))); err != nil {
+		if err := http.ListenAndServe(zipkinAddr, c.Handler(recoveryHandler(r))); err != nil {
 			logger.Fatal("Could not launch service", zap.Error(err))
 		}
 	}
@@ -261,3 +264,13 @@ func initSamplingStrategyStore(
 	}
 	return strategyStore
 }
+
+// Utility function to decide listening address based on port (deprecated flags) or host:port (new flags)
+func getAddressFromCLIOptions(port int, addr string, deprecatedFlag string, logger *zap.Logger) string {
+	if port != 0 {
+		logger.Warn("Using deprecated configuration", zap.String("option", deprecatedFlag))
+		return ports.PortToHostPort(port)
+	}
+
+	return addr
+}