Add reconnecting udp connection type to jaeger exporter

example/jaeger/go.sum

@@ -113,6 +113,7 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
+github.com/stretchr/objx v0.1.0 h1:4G4v2dO3VZwixGIRoQ5Lfboy6nUhCyYzaqnIAPPhYs4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=

exporters/trace/jaeger/agent.go

@@ -17,7 +17,9 @@ package jaeger
 import (
 	"fmt"
 	"io"
+	"log"
 	"net"
+	"time"
 
 	"github.com/apache/thrift/lib/go/thrift"
 
@@ -32,41 +34,76 @@ type agentClientUDP struct {
 	gen.Agent
 	io.Closer
 
-	connUDP       *net.UDPConn
+	connUDP       udpConn
 	client        *gen.AgentClient
 	maxPacketSize int                   // max size of datagram in bytes
 	thriftBuffer  *thrift.TMemoryBuffer // buffer used to calculate byte size of a span
 }
 
+type udpConn interface {
+	Write([]byte) (int, error)
+	SetWriteBuffer(int) error
+	Close() error
+}
+
+type agentClientUDPParams struct {
+	HostPort                   string
+	MaxPacketSize              int
+	Logger                     *log.Logger
+	DisableAttemptReconnecting bool
+	AttemptReconnectInterval   time.Duration
+}
+
 // newAgentClientUDP creates a client that sends spans to Jaeger Agent over UDP.
-func newAgentClientUDP(hostPort string, maxPacketSize int) (*agentClientUDP, error) {
-	if maxPacketSize == 0 {
-		maxPacketSize = udpPacketMaxLength
+func newAgentClientUDP(params agentClientUDPParams) (*agentClientUDP, error) {
+	// validate hostport
+	if _, _, err := net.SplitHostPort(params.HostPort); err != nil {
+		return nil, err
+	}
+
+	if params.MaxPacketSize == 0 {
+		params.MaxPacketSize = udpPacketMaxLength
 	}
 
-	thriftBuffer := thrift.NewTMemoryBufferLen(maxPacketSize)
+	if !params.DisableAttemptReconnecting && params.AttemptReconnectInterval == 0 {
+		params.AttemptReconnectInterval = time.Second * 30
+	}
+
+	thriftBuffer := thrift.NewTMemoryBufferLen(params.MaxPacketSize)
 	protocolFactory := thrift.NewTCompactProtocolFactory()
 	client := gen.NewAgentClientFactory(thriftBuffer, protocolFactory)
 
-	destAddr, err := net.ResolveUDPAddr("udp", hostPort)
-	if err != nil {
-		return nil, err
-	}
+	var connUDP udpConn
+	var err error
 
-	connUDP, err := net.DialUDP(destAddr.Network(), nil, destAddr)
-	if err != nil {
-		return nil, err
+	if params.DisableAttemptReconnecting {
+		destAddr, err := net.ResolveUDPAddr("udp", params.HostPort)
+		if err != nil {
+			return nil, err
+		}
+
+		connUDP, err = net.DialUDP(destAddr.Network(), nil, destAddr)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		// host is hostname, setup resolver loop in case host record changes during operation
+		connUDP, err = newReconnectingUDPConn(params.HostPort, params.MaxPacketSize, params.AttemptReconnectInterval, net.ResolveUDPAddr, net.DialUDP, params.Logger)
+		if err != nil {
+			return nil, err
+		}
 	}
-	if err := connUDP.SetWriteBuffer(maxPacketSize); err != nil {
+
+	if err := connUDP.SetWriteBuffer(params.MaxPacketSize); err != nil {
 		return nil, err
 	}
 
-	clientUDP := &agentClientUDP{
+	return &agentClientUDP{
 		connUDP:       connUDP,
 		client:        client,
-		maxPacketSize: maxPacketSize,
-		thriftBuffer:  thriftBuffer}
-	return clientUDP, nil
+		maxPacketSize: params.MaxPacketSize,
+		thriftBuffer:  thriftBuffer,
+	}, nil
 }
 
 // EmitBatch implements EmitBatch() of Agent interface

exporters/trace/jaeger/agent_test.go

@@ -0,0 +1,111 @@
+// Copyright The OpenTelemetry 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 jaeger
+
+import (
+	"log"
+	"net"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewAgentClientUDPWithParamsBadHostport(t *testing.T) {
+	hostPort := "blahblah"
+
+	agentClient, err := newAgentClientUDP(agentClientUDPParams{
+		HostPort: hostPort,
+	})
+
+	assert.Error(t, err)
+	assert.Nil(t, agentClient)
+}
+
+func TestNewAgentClientUDPWithParams(t *testing.T) {
+	mockServer, err := newUDPListener()
+	require.NoError(t, err)
+	defer mockServer.Close()
+
+	agentClient, err := newAgentClientUDP(agentClientUDPParams{
+		HostPort:      mockServer.LocalAddr().String(),
+		MaxPacketSize: 25000,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, agentClient)
+	assert.Equal(t, 25000, agentClient.maxPacketSize)
+
+	if assert.IsType(t, &reconnectingUDPConn{}, agentClient.connUDP) {
+		assert.Equal(t, (*log.Logger)(nil), agentClient.connUDP.(*reconnectingUDPConn).logger)
+	}
+
+	assert.NoError(t, agentClient.Close())
+}
+
+func TestNewAgentClientUDPWithParamsDefaults(t *testing.T) {
+	mockServer, err := newUDPListenerOnPort(6831)
+	require.NoError(t, err)
+	defer mockServer.Close()
+
+	agentClient, err := newAgentClientUDP(agentClientUDPParams{
+		HostPort: "localhost:6831",
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, agentClient)
+	assert.Equal(t, udpPacketMaxLength, agentClient.maxPacketSize)
+
+	if assert.IsType(t, &reconnectingUDPConn{}, agentClient.connUDP) {
+		assert.Equal(t, (*log.Logger)(nil), agentClient.connUDP.(*reconnectingUDPConn).logger)
+	}
+
+	assert.NoError(t, agentClient.Close())
+}
+
+func TestNewAgentClientUDPDefaults(t *testing.T) {
+	mockServer, err := newUDPListenerOnPort(6831)
+	require.NoError(t, err)
+	defer mockServer.Close()
+
+	agentClient, err := newAgentClientUDP(agentClientUDPParams{
+		HostPort: "localhost:6831",
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, agentClient)
+	assert.Equal(t, udpPacketMaxLength, agentClient.maxPacketSize)
+
+	if assert.IsType(t, &reconnectingUDPConn{}, agentClient.connUDP) {
+		assert.Equal(t, (*log.Logger)(nil), agentClient.connUDP.(*reconnectingUDPConn).logger)
+	}
+
+	assert.NoError(t, agentClient.Close())
+}
+
+func TestNewAgentClientUDPWithParamsReconnectingDisabled(t *testing.T) {
+	mockServer, err := newUDPListener()
+	require.NoError(t, err)
+	defer mockServer.Close()
+
+	agentClient, err := newAgentClientUDP(agentClientUDPParams{
+		HostPort:                   mockServer.LocalAddr().String(),
+		Logger:                     nil,
+		DisableAttemptReconnecting: true,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, agentClient)
+	assert.Equal(t, udpPacketMaxLength, agentClient.maxPacketSize)
+
+	assert.IsType(t, &net.UDPConn{}, agentClient.connUDP)
+
+	assert.NoError(t, agentClient.Close())
+}

exporters/trace/jaeger/go.sum

@@ -113,6 +113,7 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
 github.com/rogpeppe/go-internal v1.3.0/go.mod h1:M8bDsm7K2OlrFYOpmOWEs/qY81heoFRclV5y23lUDJ4=
+github.com/stretchr/objx v0.1.0 h1:4G4v2dO3VZwixGIRoQ5Lfboy6nUhCyYzaqnIAPPhYs4=
 github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
 github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
 github.com/stretchr/testify v1.6.1 h1:hDPOHmpOpP40lSULcqw7IrRb/u7w6RpDC9399XyoNd0=