Skip to content

Commit

Permalink
Add reconnecting udp connection type to jaeger exporter (open-telemet…
Browse files Browse the repository at this point in the history
…ry#1063)

* port reconnecting udp client from jaeger-client-go

* Fix precommit issues

* Fix license check

* Add initial value for max packet size

* Fix for atomic usage on 386 arch

* Modify reconnecting option to an affirmative

* Add changelog entry

* Dont hold rlock for writes

Co-authored-by: Tyler Yahn <MrAlias@users.noreply.github.com>
  • Loading branch information
2 people authored and evantorrie committed Sep 10, 2020
1 parent e6a144c commit 08875a3
Show file tree
Hide file tree
Showing 8 changed files with 862 additions and 19 deletions.
6 changes: 6 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,12 @@ This project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.htm

## [Unreleased]

### Added

- Add reconnecting udp connection type to Jaeger exporter.
This change adds a new optional implementation of the udp conn interface used to detect changes to an agent's host dns record.
It then adopts the new destination address to ensure the exporter doesn't get stuck. This change was ported from jaegertracing/jaeger-client-go#520. (#1063)

## [0.11.0] - 2020-08-24

### Added
Expand Down
1 change: 1 addition & 0 deletions example/jaeger/go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,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=
Expand Down
71 changes: 54 additions & 17 deletions exporters/trace/jaeger/agent.go
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,9 @@ package jaeger
import (
"fmt"
"io"
"log"
"net"
"time"

"github.com/apache/thrift/lib/go/thrift"

Expand All @@ -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
AttemptReconnecting 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.AttemptReconnecting && 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.AttemptReconnecting {
// 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
}
} else {
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
}
}
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
Expand Down
94 changes: 94 additions & 0 deletions exporters/trace/jaeger/agent_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
// 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,
AttemptReconnecting: true,
})
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 := newUDPListener()
require.NoError(t, err)
defer mockServer.Close()

agentClient, err := newAgentClientUDP(agentClientUDPParams{
HostPort: mockServer.LocalAddr().String(),
AttemptReconnecting: true,
})
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,
AttemptReconnecting: false,
})
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())
}
1 change: 1 addition & 0 deletions exporters/trace/jaeger/go.sum
Original file line number Diff line number Diff line change
Expand Up @@ -115,6 +115,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=
Expand Down
Loading

0 comments on commit 08875a3

Please sign in to comment.