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.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=

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://github.com/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()

receiver/jaegerreceiver/jaeger_agent_test.go

@@ -17,18 +17,23 @@ package jaegerreceiver
 import (
 	"context"
 	"fmt"
+	"net"
+	"net/http"
 	"testing"
 	"time"
 
 	"contrib.go.opencensus.io/exporter/jaeger"
 	commonpb "github.com/census-instrumentation/opencensus-proto/gen-go/agent/common/v1"
 	tracepb "github.com/census-instrumentation/opencensus-proto/gen-go/trace/v1"
 	"github.com/google/go-cmp/cmp"
+	"github.com/stretchr/testify/assert"
 	"go.opencensus.io/trace"
+	"go.uber.org/zap"
 
 	"github.com/open-telemetry/opentelemetry-collector/consumer/consumerdata"
 	"github.com/open-telemetry/opentelemetry-collector/exporter/exportertest"
 	"github.com/open-telemetry/opentelemetry-collector/internal"
+	"github.com/open-telemetry/opentelemetry-collector/internal/testutils"
 	"github.com/open-telemetry/opentelemetry-collector/receiver/receivertest"
 )
 
@@ -40,6 +45,24 @@ func TestJaegerAgentUDP_ThriftCompact_6831(t *testing.T) {
 	})
 }
 
+func TestJaegerAgentUDP_ThriftCompact_InvalidPort(t *testing.T) {
+	port := 999999
+
+	config := &Configuration{
+		AgentCompactThriftPort: int(port),
+	}
+	jr, err := New(context.Background(), config, nil, zap.NewNop())
+	if err != nil {
+		t.Fatalf("Failed to create new Jaeger Receiver: %v", err)
+	}
+
+	mh := receivertest.NewMockHost()
+	err = jr.StartTraceReception(mh)
+	assert.Error(t, err, "should not have been able to startTraceReception")
+
+	jr.StopTraceReception()
+}
+
 func TestJaegerAgentUDP_ThriftBinary_6832(t *testing.T) {
 	t.Skipf("Unfortunately due to Jaeger internal versioning, OpenCensus-Go's Thrift seems to conflict with ours")
 
@@ -50,10 +73,98 @@ func TestJaegerAgentUDP_ThriftBinary_6832(t *testing.T) {
 	})
 }
 
+func TestJaegerAgentUDP_ThriftBinary_PortInUse(t *testing.T) {
+	// This test confirms that the thrift binary port is opened correctly.  This is all we can test at the moment.  See above.
+	port := testutils.GetAvailablePort(t)
+
+	config := &Configuration{
+		AgentBinaryThriftPort: int(port),
+	}
+	jr, err := New(context.Background(), config, nil, zap.NewNop())
+	if err != nil {
+		t.Fatalf("Failed to create new Jaeger Receiver: %v", err)
+	}
+
+	mh := receivertest.NewMockHost()
+	if err := jr.(*jReceiver).startAgent(mh); err != nil {
+		t.Fatalf("StartTraceReception failed: %v", err)
+	}
+	defer jr.StopTraceReception()
+
+	l, err := net.Listen("udp", fmt.Sprintf("localhost:%d", port))
+	assert.Error(t, err, "should not have been able to listen to the port")
+
+	if l != nil {
+		l.Close()
+	}
+}
+
+func TestJaegerAgentUDP_ThriftBinary_InvalidPort(t *testing.T) {
+	port := 999999
+
+	config := &Configuration{
+		AgentBinaryThriftPort: int(port),
+	}
+	jr, err := New(context.Background(), config, nil, zap.NewNop())
+	if err != nil {
+		t.Fatalf("Failed to create new Jaeger Receiver: %v", err)
+	}
+
+	mh := receivertest.NewMockHost()
+	err = jr.StartTraceReception(mh)
+	assert.Error(t, err, "should not have been able to startTraceReception")
+
+	jr.StopTraceReception()
+}
+
+func TestJaegerHTTP(t *testing.T) {
+	port := testutils.GetAvailablePort(t)
+	config := &Configuration{
+		AgentHTTPPort: int(port),
+	}
+	jr, err := New(context.Background(), config, nil, zap.NewNop())
+	if err != nil {
+		t.Fatalf("Failed to create new Jaeger Receiver: %v", err)
+	}
+	defer jr.StopTraceReception()
+
+	mh := receivertest.NewMockHost()
+	if err := jr.StartTraceReception(mh); err != nil {
+		t.Fatalf("StartTraceReception failed: %v", err)
+	}
+
+	// allow http server to start
+	if err := testutils.WaitForPort(t, port); err != nil {
+		t.Fatalf("WaitForPort failed: %v", err)
+	}
+
+	// this functionality is just stubbed out at the moment.  just confirm they 200.
+	testURL := fmt.Sprintf("http://localhost:%d/sampling?service=test", port)
+	resp, err := http.Get(testURL)
+	assert.NoError(t, err, "should not have failed to make request")
+	if resp != nil {
+		assert.Equal(t, 200, resp.StatusCode, "should have returned 200")
+	}
+
+	testURL = fmt.Sprintf("http://localhost:%d/sampling?service=test", port)
+	resp, err = http.Get(testURL)
+	assert.NoError(t, err, "should not have failed to make request")
+	if resp != nil {
+		assert.Equal(t, 200, resp.StatusCode, "should have returned 200")
+	}
+
+	testURL = fmt.Sprintf("http://localhost:%d/baggageRestrictions?service=test", port)
+	resp, err = http.Get(testURL)
+	assert.NoError(t, err, "should not have failed to make request")
+	if resp != nil {
+		assert.Equal(t, 200, resp.StatusCode, "should have returned 200")
+	}
+}
+
 func testJaegerAgent(t *testing.T, agentEndpoint string, receiverConfig *Configuration) {
 	// 1. Create the Jaeger receiver aka "server"
 	sink := new(exportertest.SinkTraceExporter)
-	jr, err := New(context.Background(), receiverConfig, sink)
+	jr, err := New(context.Background(), receiverConfig, sink, zap.NewNop())
 	if err != nil {
 		t.Fatalf("Failed to create new Jaeger Receiver: %v", err)
 	}
@@ -220,6 +331,7 @@ func testJaegerAgent(t *testing.T, agentEndpoint string, receiverConfig *Configu
 					},
 				},
 			},
+			SourceFormat: "jaeger",
 		},
 	}