Add Jaeger Agent Configuration

README.md

@@ -167,6 +167,10 @@ receivers:
  endpoint: "localhost:14268"
  thrift-tchannel:
  endpoint: "localhost:14267"
+ thrift-compact:
+ endpoint: "localhost:6831"
+ thrift-binary:
+ endpoint: "localhost:6832"
 
  prometheus:
  config:

go.mod

@@ -42,7 +42,7 @@ require (
  github.com/stretchr/testify v1.4.0
  github.com/uber-go/atomic v1.4.0 // indirect
  github.com/uber/jaeger-client-go v2.16.0+incompatible // indirect
- github.com/uber/jaeger-lib v2.0.0+incompatible
+ github.com/uber/jaeger-lib v2.2.0+incompatible
  github.com/uber/tchannel-go v1.10.0
  go.opencensus.io v0.22.1
  go.uber.org/zap v1.10.0

go.sum

@@ -48,6 +48,7 @@ github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRF
 github.com/apache/thrift v0.0.0-20161221203622-b2a4d4ae21c7 h1:Fv9bK1Q+ly/ROk4aJsVMeuIwPel4bEnD8EPiI91nZMg=
 github.com/apache/thrift v0.0.0-20161221203622-b2a4d4ae21c7/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
 github.com/apache/thrift v0.0.0-20161221203622-b2a4d4ae21c7/go.mod h1:cp2SuWMxlEZw2r+iP2GNCdIi4C1qmUzdZFSVb+bacwQ=
+github.com/apache/thrift v0.13.0 h1:5hryIiq9gtn+MiLVn0wP37kb/uTeRZgN08WoCsAhIhI=
 github.com/armon/circbuf v0.0.0-20150827004946-bbbad097214e/go.mod h1:3U/XgcO3hCbHZ8TKRvWD2dDTCfh9M9ya+I9JpbB7O8o=
 github.com/armon/consul-api v0.0.0-20180202201655-eb2c6b5be1b6/go.mod h1:grANhF5doyWs3UAsr3K4I6qtAmlQcZDesFNEHPZAzj8=
 github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da h1:8GUt8eRujhVEGZFFEjBj46YV4rDjvGrNxb0KMWYkL2I=
@@ -553,6 +554,8 @@ github.com/uber/jaeger-client-go v2.16.0+incompatible h1:Q2Pp6v3QYiocMxomCaJuwQG
 github.com/uber/jaeger-client-go v2.16.0+incompatible/go.mod h1:WVhlPFC8FDjOFMMWRy2pZqQJSXxYSwNYOkTr/Z6d3Kk=
 github.com/uber/jaeger-lib v2.0.0+incompatible h1:iMSCV0rmXEogjNWPh2D0xk9YVKvrtGoHJNe9ebLu/pw=
 github.com/uber/jaeger-lib v2.0.0+incompatible/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U=
+github.com/uber/jaeger-lib v2.2.0+incompatible h1:MxZXOiR2JuoANZ3J6DE/U0kSFv/eJ/GfSYVCjK7dyaw=
+github.com/uber/jaeger-lib v2.2.0+incompatible/go.mod h1:ComeNDZlWwrWnDv8aPp0Ba6+uUTzImX/AauajbLI56U=
 github.com/uber/tchannel-go v1.10.0 h1:YOihLHuvkwT3nzvpgqFtexFW+pb5vD1Tz7h/bIWApgE=
 github.com/uber/tchannel-go v1.10.0/go.mod h1:Rrgz1eL8kMjW/nEzZos0t+Heq0O4LhnUJVA32OvWKHo=
 github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc=

internal/testutils/testutils.go

@@ -16,9 +16,11 @@ package testutils
 
 import (
  "encoding/json"
+ "fmt"
  "net"
  "strconv"
  "testing"
+ "time"
 
  "github.com/stretchr/testify/require"
 )
@@ -67,3 +69,23 @@ func GetAvailablePort(t *testing.T) uint16 {
 
  return uint16(portInt)
 }
+
+// WaitForPort repeatedly attempts to open a local port until it either succeeds or 5 seconds pass
+// It is useful if you need to asynchronously start a service and wait for it to start
+func WaitForPort(t *testing.T, port uint16) error {
+ t.Helper()
+
+ totalDuration := 5 * time.Second
+ wait := 100 * time.Millisecond
+ address := fmt.Sprintf("localhost:%d", port)
+ for i := totalDuration; i > 0; i -= wait {
+ conn, err := net.Dial("tcp", address)
+
+ if err == nil && conn != nil {
+ conn.Close()
+ return nil
+ }
+ time.Sleep(wait)
+ }
+ return fmt.Errorf("failed to wait for port %d", port)
+}

internal/testutils/testutils_test.go

@@ -15,6 +15,7 @@
 package testutils
 
 import (
+ "fmt"
  "net"
  "strconv"
  "testing"
@@ -33,6 +34,22 @@ func TestGetAvailablePort(t *testing.T) {
  testEndpointAvailable(t, "localhost:"+portStr)
 }
 
+func TestWaitForPort(t *testing.T) {
+ port := GetAvailablePort(t)
+ err := WaitForPort(t, port)
+ require.Error(t, err)
+
+ port = GetAvailablePort(t)
+ l, err := net.Listen("tcp", fmt.Sprintf("localhost:%d", port))
+ require.NoError(t, err)
+
+ err = WaitForPort(t, port)
+ require.NoError(t, err)
+
+ err = l.Close()
+ require.NoError(t, err)
+}
+
 func testEndpointAvailable(t *testing.T, endpoint string) {
  // Endpoint should be free.
  ln0, err := net.Listen("tcp", endpoint)

receiver/README.md

@@ -107,7 +107,7 @@ This receiver receives traces in the [Jaeger](https://www.jaegertracing.io)
 format. It translates them into the internal format and sends
 it to processors and exporters.
 
-It supports multiple protocols:
+It supports the Jaeger Collector protocols:
 - Thrift HTTP
 - Thrift TChannel
 - gRPC
@@ -120,6 +120,12 @@ receivers:
  jaeger:
 ```
 
+It also supports the Jaeger Agent protocols:
+- Thrift Compact
+- Thrift Binary
+
+By default, these services are not started unless an endpoint is explicitly defined.
+
 It is possible to configure the protocols on different ports, refer to
 [config.yaml](jaegerreceiver/testdata/config.yaml) for detailed config
 examples.

receiver/jaegerreceiver/config_test.go

@@ -65,6 +65,16 @@ func TestLoadConfig(t *testing.T) {
  Endpoint: "0.0.0.0:123",
  },
  },
+ "thrift-compact": {
+ ReceiverSettings: configmodels.ReceiverSettings{
+ Endpoint: "0.0.0.0:456",
+ },
+ },
+ "thrift-binary": {
+ ReceiverSettings: configmodels.ReceiverSettings{
+ Endpoint: "0.0.0.0:789",
+ },
+ },
  },
  })
 

receiver/jaegerreceiver/factory.go

@@ -41,6 +41,8 @@ const (
  // TODO https:/open-telemetry/opentelemetry-collector/issues/267
  // Remove ThriftTChannel support.
  protoThriftTChannel = "thrift-tchannel"
+ protoThriftBinary = "thrift-binary"
+ protoThriftCompact = "thrift-compact"
 
  // Default endpoints to bind to.
  defaultGRPCBindEndpoint = "localhost:14250"
@@ -103,6 +105,8 @@ func (f *Factory) CreateTraceReceiver(
  protoGRPC := rCfg.Protocols[protoGRPC]
  protoHTTP := rCfg.Protocols[protoThriftHTTP]
  protoTChannel := rCfg.Protocols[protoThriftTChannel]
+ protoThriftCompact := rCfg.Protocols[protoThriftCompact]
+ protoThriftBinary := rCfg.Protocols[protoThriftBinary]
 
  config := Configuration{}
  var grpcServerOptions []grpc.ServerOption
@@ -141,19 +145,37 @@ func (f *Factory) CreateTraceReceiver(
  }
  }
 
- if (protoGRPC == nil && protoHTTP == nil && protoTChannel == nil) ||
- (config.CollectorGRPCPort == 0 && config.CollectorHTTPPort == 0 && config.CollectorThriftPort == 0) {
- err := fmt.Errorf("either %v, %v, or %v protocol endpoint with non-zero port must be enabled for %s receiver",
+ if protoThriftBinary != nil && protoThriftBinary.IsEnabled() {
+ var err error
+ config.AgentBinaryThriftPort, err = extractPortFromEndpoint(protoThriftBinary.Endpoint)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ if protoThriftCompact != nil && protoThriftCompact.IsEnabled() {
+ var err error
+ config.AgentCompactThriftPort, err = extractPortFromEndpoint(protoThriftCompact.Endpoint)
+ if err != nil {
+ return nil, err
+ }
+ }
+
+ if (protoGRPC == nil && protoHTTP == nil && protoTChannel == nil && protoThriftBinary == nil && protoThriftCompact == nil) ||
+ (config.CollectorGRPCPort == 0 && config.CollectorHTTPPort == 0 && config.CollectorThriftPort == 0 && config.AgentBinaryThriftPort == 0 && config.AgentCompactThriftPort == 0) {
+ err := fmt.Errorf("either %v, %v, %v, %v, or %v protocol endpoint with non-zero port must be enabled for %s receiver",
  protoGRPC,
  protoThriftHTTP,
  protoThriftTChannel,
+ protoThriftCompact,
+ protoThriftBinary,
  typeStr,
  )
  return nil, err
  }
 
  // Create the receiver.
- return New(ctx, &config, nextConsumer)
+ return New(ctx, &config, nextConsumer, logger)
 }
 
 // CreateMetricsReceiver creates a metrics receiver based on provided config.

receiver/jaegerreceiver/factory_test.go

@@ -23,6 +23,8 @@ import (
 
  "github.com/open-telemetry/opentelemetry-collector/config/configcheck"
  "github.com/open-telemetry/opentelemetry-collector/config/configerror"
+ "github.com/open-telemetry/opentelemetry-collector/config/configmodels"
+ "github.com/open-telemetry/opentelemetry-collector/receiver"
 )
 
 func TestCreateDefaultConfig(t *testing.T) {
@@ -75,6 +77,34 @@ func TestCreateInvalidTChannelEndpoint(t *testing.T) {
  assert.Error(t, err, "receiver creation with invalid tchannel endpoint must fail")
 }
 
+func TestCreateInvalidThriftBinaryEndpoint(t *testing.T) {
+ factory := Factory{}
+ cfg := factory.CreateDefaultConfig()
+ rCfg := cfg.(*Config)
+
+ rCfg.Protocols[protoThriftBinary] = &receiver.SecureReceiverSettings{
+ ReceiverSettings: configmodels.ReceiverSettings{
+ Endpoint: "",
+ },
+ }
+ _, err := factory.CreateTraceReceiver(context.Background(), zap.NewNop(), cfg, nil)
+ assert.Error(t, err, "receiver creation with no endpoints must fail")
+}
+
+func TestCreateInvalidThriftCompactEndpoint(t *testing.T) {
+ factory := Factory{}
+ cfg := factory.CreateDefaultConfig()
+ rCfg := cfg.(*Config)
+
+ rCfg.Protocols[protoThriftCompact] = &receiver.SecureReceiverSettings{
+ ReceiverSettings: configmodels.ReceiverSettings{
+ Endpoint: "",
+ },
+ }
+ _, err := factory.CreateTraceReceiver(context.Background(), zap.NewNop(), cfg, nil)
+ assert.Error(t, err, "receiver creation with no endpoints must fail")
+}
+
 func TestCreateNoPort(t *testing.T) {
  factory := Factory{}
  cfg := factory.CreateDefaultConfig()