opentelemetry-go/exporters/trace/jaeger/reconnecting_udp_client_test.go

// 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 (
	"context"
	"fmt"
	"math/rand"
	"net"
	"runtime"
	"syscall"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
)

type mockResolver struct {
	mock.Mock
}

func (m *mockResolver) ResolveUDPAddr(network string, hostPort string) (*net.UDPAddr, error) {
	args := m.Called(network, hostPort)

	a0 := args.Get(0)
	if a0 == nil {
		return (*net.UDPAddr)(nil), args.Error(1)
	}
	return a0.(*net.UDPAddr), args.Error(1)
}

type mockDialer struct {
	mock.Mock
}

func (m *mockDialer) DialUDP(network string, laddr, raddr *net.UDPAddr) (*net.UDPConn, error) {
	args := m.Called(network, laddr, raddr)

	a0 := args.Get(0)
	if a0 == nil {
		return (*net.UDPConn)(nil), args.Error(1)
	}

	return a0.(*net.UDPConn), args.Error(1)
}

func newUDPListener() (net.PacketConn, error) {
	return net.ListenPacket("udp", "127.0.0.1:0")
}

func newUDPConn() (net.PacketConn, *net.UDPConn, error) {
	mockServer, err := newUDPListener()
	if err != nil {
		return nil, nil, err
	}

	addr, err := net.ResolveUDPAddr("udp", mockServer.LocalAddr().String())
	if err != nil {
		mockServer.Close()
		return nil, nil, err
	}

	conn, err := net.DialUDP("udp", nil, addr)
	if err != nil {
		mockServer.Close()
		return nil, nil, err
	}

	return mockServer, conn, nil
}

func assertSockBufferSize(t *testing.T, expectedBytes int, conn *net.UDPConn) bool {
	fd, _ := conn.File()
	bufferBytes, _ := syscall.GetsockoptInt(int(fd.Fd()), syscall.SOL_SOCKET, syscall.SO_SNDBUF)

	// The linux kernel doubles SO_SNDBUF value (to allow space for bookkeeping overhead) when it is set using setsockopt(2), and this doubled value is returned by getsockopt(2)
	// https://linux.die.net/man/7/socket
	if runtime.GOOS == "linux" {
		return assert.GreaterOrEqual(t, expectedBytes*2, bufferBytes)
	}

	return assert.Equal(t, expectedBytes, bufferBytes)
}

func assertConnWritable(t *testing.T, conn udpConn, serverConn net.PacketConn) {
	expectedString := "yo this is a test"
	_, err := conn.Write([]byte(expectedString))
	require.NoError(t, err)

	var buf = make([]byte, len(expectedString))
	err = serverConn.SetReadDeadline(time.Now().Add(time.Second))
	require.NoError(t, err)

	_, _, err = serverConn.ReadFrom(buf)
	require.NoError(t, err)
	require.Equal(t, []byte(expectedString), buf)
}

func waitForCallWithTimeout(call *mock.Call) bool {
	called := make(chan struct{})
	call.Run(func(args mock.Arguments) {
		close(called)
	})

	var wasCalled bool
	// wait at most 100 milliseconds for the second call of ResolveUDPAddr that is supposed to fail
	ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*100)
	select {
	case <-called:
		wasCalled = true
	case <-ctx.Done():
		fmt.Println("timed out")
	}
	cancel()

	return wasCalled
}

func waitForConnCondition(conn *reconnectingUDPConn, condition func(conn *reconnectingUDPConn) bool) bool {
	var conditionVal bool
	for i := 0; i < 10; i++ {
		conn.connMtx.RLock()
		conditionVal = condition(conn)
		conn.connMtx.RUnlock()
		if conditionVal || i >= 9 {
			break
		}

		time.Sleep(time.Millisecond * 10)
	}

	return conditionVal
}

func newMockUDPAddr(t *testing.T, port int) *net.UDPAddr {
	var buf = make([]byte, 4)
	// random is not seeded to ensure tests are deterministic (also doesnt matter if ip is valid)
	_, err := rand.Read(buf)
	require.NoError(t, err)

	return &net.UDPAddr{
		IP:   net.IPv4(buf[0], buf[1], buf[2], buf[3]),
		Port: port,
	}
}

func TestNewResolvedUDPConn(t *testing.T) {
	hostPort := "blahblah:34322"

	mockServer, clientConn, err := newUDPConn()
	require.NoError(t, err)
	defer mockServer.Close()

	mockUDPAddr := newMockUDPAddr(t, 34322)

	resolver := mockResolver{}
	resolver.
		On("ResolveUDPAddr", "udp", hostPort).
		Return(mockUDPAddr, nil).
		Once()

	dialer := mockDialer{}
	dialer.
		On("DialUDP", "udp", (*net.UDPAddr)(nil), mockUDPAddr).
		Return(clientConn, nil).
		Once()

	conn, err := newReconnectingUDPConn(hostPort, udpPacketMaxLength, time.Hour, resolver.ResolveUDPAddr, dialer.DialUDP, nil)
	assert.NoError(t, err)
	require.NotNil(t, conn)

	err = conn.Close()
	assert.NoError(t, err)

	// assert the actual connection was closed
	assert.Error(t, clientConn.Close())

	resolver.AssertExpectations(t)
	dialer.AssertExpectations(t)
}

func TestResolvedUDPConnWrites(t *testing.T) {
	hostPort := "blahblah:34322"

	mockServer, clientConn, err := newUDPConn()
	require.NoError(t, err)
	defer mockServer.Close()

	mockUDPAddr := newMockUDPAddr(t, 34322)

	resolver := mockResolver{}
	resolver.
		On("ResolveUDPAddr", "udp", hostPort).
		Return(mockUDPAddr, nil).
		Once()

	dialer := mockDialer{}
	dialer.
		On("DialUDP", "udp", (*net.UDPAddr)(nil), mockUDPAddr).
		Return(clientConn, nil).
		Once()

	conn, err := newReconnectingUDPConn(hostPort, udpPacketMaxLength, time.Hour, resolver.ResolveUDPAddr, dialer.DialUDP, nil)
	assert.NoError(t, err)
	require.NotNil(t, conn)

	assertConnWritable(t, conn, mockServer)

	err = conn.Close()
	assert.NoError(t, err)

	// assert the actual connection was closed
	assert.Error(t, clientConn.Close())

	resolver.AssertExpectations(t)
	dialer.AssertExpectations(t)
}

func TestResolvedUDPConnEventuallyDials(t *testing.T) {
	hostPort := "blahblah:34322"

	mockServer, clientConn, err := newUDPConn()
	require.NoError(t, err)
	defer mockServer.Close()

	mockUDPAddr := newMockUDPAddr(t, 34322)

	resolver := mockResolver{}
	resolver.
		On("ResolveUDPAddr", "udp", hostPort).
		Return(nil, fmt.Errorf("failed to resolve")).Once().
		On("ResolveUDPAddr", "udp", hostPort).
		Return(mockUDPAddr, nil)

	dialer := mockDialer{}
	dialCall := dialer.
		On("DialUDP", "udp", (*net.UDPAddr)(nil), mockUDPAddr).
		Return(clientConn, nil).Once()

	conn, err := newReconnectingUDPConn(hostPort, udpPacketMaxLength, time.Millisecond*10, resolver.ResolveUDPAddr, dialer.DialUDP, nil)
	assert.NoError(t, err)
	require.NotNil(t, conn)

	err = conn.SetWriteBuffer(udpPacketMaxLength)
	assert.NoError(t, err)

	wasCalled := waitForCallWithTimeout(dialCall)
	assert.True(t, wasCalled)

	connEstablished := waitForConnCondition(conn, func(conn *reconnectingUDPConn) bool {
		return conn.conn != nil
	})

	assert.True(t, connEstablished)

	assertConnWritable(t, conn, mockServer)
	assertSockBufferSize(t, udpPacketMaxLength, clientConn)

	err = conn.Close()
	assert.NoError(t, err)

	// assert the actual connection was closed
	assert.Error(t, clientConn.Close())

	resolver.AssertExpectations(t)
	dialer.AssertExpectations(t)
}

func TestResolvedUDPConnNoSwapIfFail(t *testing.T) {
	hostPort := "blahblah:34322"

	mockServer, clientConn, err := newUDPConn()
	require.NoError(t, err)
	defer mockServer.Close()

	mockUDPAddr := newMockUDPAddr(t, 34322)

	resolver := mockResolver{}
	resolver.
		On("ResolveUDPAddr", "udp", hostPort).
		Return(mockUDPAddr, nil).Once()

	failCall := resolver.On("ResolveUDPAddr", "udp", hostPort).
		Return(nil, fmt.Errorf("resolve failed"))

	dialer := mockDialer{}
	dialer.
		On("DialUDP", "udp", (*net.UDPAddr)(nil), mockUDPAddr).
		Return(clientConn, nil).Once()

	conn, err := newReconnectingUDPConn(hostPort, udpPacketMaxLength, time.Millisecond*10, resolver.ResolveUDPAddr, dialer.DialUDP, nil)
	assert.NoError(t, err)
	require.NotNil(t, conn)

	wasCalled := waitForCallWithTimeout(failCall)

	assert.True(t, wasCalled)

	assertConnWritable(t, conn, mockServer)

	err = conn.Close()
	assert.NoError(t, err)

	// assert the actual connection was closed
	assert.Error(t, clientConn.Close())

	resolver.AssertExpectations(t)
	dialer.AssertExpectations(t)
}

func TestResolvedUDPConnWriteRetry(t *testing.T) {
	hostPort := "blahblah:34322"

	mockServer, clientConn, err := newUDPConn()
	require.NoError(t, err)
	defer mockServer.Close()

	mockUDPAddr := newMockUDPAddr(t, 34322)

	resolver := mockResolver{}
	resolver.
		On("ResolveUDPAddr", "udp", hostPort).
		Return(nil, fmt.Errorf("failed to resolve")).Once().
		On("ResolveUDPAddr", "udp", hostPort).
		Return(mockUDPAddr, nil).Once()

	dialer := mockDialer{}
	dialer.
		On("DialUDP", "udp", (*net.UDPAddr)(nil), mockUDPAddr).
		Return(clientConn, nil).Once()

	conn, err := newReconnectingUDPConn(hostPort, udpPacketMaxLength, time.Millisecond*10, resolver.ResolveUDPAddr, dialer.DialUDP, nil)
	assert.NoError(t, err)
	require.NotNil(t, conn)

	err = conn.SetWriteBuffer(udpPacketMaxLength)
	assert.NoError(t, err)

	assertConnWritable(t, conn, mockServer)
	assertSockBufferSize(t, udpPacketMaxLength, clientConn)

	err = conn.Close()
	assert.NoError(t, err)

	// assert the actual connection was closed
	assert.Error(t, clientConn.Close())

	resolver.AssertExpectations(t)
	dialer.AssertExpectations(t)
}

func TestResolvedUDPConnWriteRetryFails(t *testing.T) {
	hostPort := "blahblah:34322"

	resolver := mockResolver{}
	resolver.
		On("ResolveUDPAddr", "udp", hostPort).
		Return(nil, fmt.Errorf("failed to resolve")).Twice()

	dialer := mockDialer{}

	conn, err := newReconnectingUDPConn(hostPort, udpPacketMaxLength, time.Millisecond*10, resolver.ResolveUDPAddr, dialer.DialUDP, nil)
	assert.NoError(t, err)
	require.NotNil(t, conn)

	err = conn.SetWriteBuffer(udpPacketMaxLength)
	assert.NoError(t, err)

	_, err = conn.Write([]byte("yo this is a test"))

	assert.Error(t, err)

	err = conn.Close()
	assert.NoError(t, err)

	resolver.AssertExpectations(t)
	dialer.AssertExpectations(t)
}

func TestResolvedUDPConnChanges(t *testing.T) {
	hostPort := "blahblah:34322"

	mockServer, clientConn, err := newUDPConn()
	require.NoError(t, err)
	defer mockServer.Close()

	mockUDPAddr1 := newMockUDPAddr(t, 34322)

	mockServer2, clientConn2, err := newUDPConn()
	require.NoError(t, err)
	defer mockServer2.Close()

	mockUDPAddr2 := newMockUDPAddr(t, 34322)

	// ensure address doesn't duplicate mockUDPAddr1
	for i := 0; i < 10 && mockUDPAddr2.IP.Equal(mockUDPAddr1.IP); i++ {
		mockUDPAddr2 = newMockUDPAddr(t, 34322)
	}

	// this is really unlikely to ever fail the test, but its here as a safeguard
	require.False(t, mockUDPAddr2.IP.Equal(mockUDPAddr1.IP))

	resolver := mockResolver{}
	resolver.
		On("ResolveUDPAddr", "udp", hostPort).
		Return(mockUDPAddr1, nil).Once().
		On("ResolveUDPAddr", "udp", hostPort).
		Return(mockUDPAddr2, nil)

	dialer := mockDialer{}
	dialer.
		On("DialUDP", "udp", (*net.UDPAddr)(nil), mockUDPAddr1).
		Return(clientConn, nil).Once()

	secondDial := dialer.
		On("DialUDP", "udp", (*net.UDPAddr)(nil), mockUDPAddr2).
		Return(clientConn2, nil).Once()

	conn, err := newReconnectingUDPConn(hostPort, udpPacketMaxLength, time.Millisecond*10, resolver.ResolveUDPAddr, dialer.DialUDP, nil)
	assert.NoError(t, err)
	require.NotNil(t, conn)

	err = conn.SetWriteBuffer(udpPacketMaxLength)
	assert.NoError(t, err)

	wasCalled := waitForCallWithTimeout(secondDial)
	assert.True(t, wasCalled)

	connSwapped := waitForConnCondition(conn, func(conn *reconnectingUDPConn) bool {
		return conn.conn == clientConn2
	})

	assert.True(t, connSwapped)

	assertConnWritable(t, conn, mockServer2)
	assertSockBufferSize(t, udpPacketMaxLength, clientConn2)

	err = conn.Close()
	assert.NoError(t, err)

	// assert the prev connection was closed
	assert.Error(t, clientConn.Close())

	// assert the actual connection was closed
	assert.Error(t, clientConn2.Close())

	resolver.AssertExpectations(t)
	dialer.AssertExpectations(t)
}

func TestResolvedUDPConnLoopWithoutChanges(t *testing.T) {
	hostPort := "blahblah:34322"

	mockServer, clientConn, err := newUDPConn()
	require.NoError(t, err)
	defer mockServer.Close()

	mockUDPAddr := newMockUDPAddr(t, 34322)

	resolver := mockResolver{}
	resolver.
		On("ResolveUDPAddr", "udp", hostPort).
		Return(mockUDPAddr, nil)

	dialer := mockDialer{}
	dialer.
		On("DialUDP", "udp", (*net.UDPAddr)(nil), mockUDPAddr).
		Return(clientConn, nil).
		Once()

	resolveTimeout := 500 * time.Millisecond
	conn, err := newReconnectingUDPConn(hostPort, udpPacketMaxLength, resolveTimeout, resolver.ResolveUDPAddr, dialer.DialUDP, nil)
	assert.NoError(t, err)
	require.NotNil(t, conn)
	assert.Equal(t, mockUDPAddr, conn.destAddr)

	// Waiting for one round of loop
	time.Sleep(3 * resolveTimeout)
	assert.Equal(t, mockUDPAddr, conn.destAddr)

	err = conn.Close()
	assert.NoError(t, err)

	// assert the actual connection was closed
	assert.Error(t, clientConn.Close())

	resolver.AssertExpectations(t)
	dialer.AssertExpectations(t)
}