MM-27613: Update dependencies (#15294)
https://mattermost.atlassian.net/browse/MM-27613
Этот коммит содержится в:
Agniva De Sarker
коммит произвёл
GitHub
GitHub
родитель
54f86e7fb1
Коммит
f2a8e10216
189
vendor/github.com/uber/jaeger-client-go/utils/reconnecting_udp_conn.go
сгенерированный
поставляемый
Обычный файл
189
vendor/github.com/uber/jaeger-client-go/utils/reconnecting_udp_conn.go
сгенерированный
поставляемый
Обычный файл
@@ -0,0 +1,189 @@
|
||||
// Copyright (c) 2020 The Jaeger 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 utils
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"github.com/uber/jaeger-client-go/log"
|
||||
)
|
||||
|
||||
// reconnectingUDPConn is an implementation of udpConn that resolves hostPort every resolveTimeout, if the resolved address is
|
||||
// different than the current conn then the new address is dialed and the conn is swapped.
|
||||
type reconnectingUDPConn struct {
|
||||
hostPort string
|
||||
resolveFunc resolveFunc
|
||||
dialFunc dialFunc
|
||||
logger log.Logger
|
||||
bufferBytes int64
|
||||
|
||||
connMtx sync.RWMutex
|
||||
conn *net.UDPConn
|
||||
destAddr *net.UDPAddr
|
||||
closeChan chan struct{}
|
||||
}
|
||||
|
||||
type resolveFunc func(network string, hostPort string) (*net.UDPAddr, error)
|
||||
type dialFunc func(network string, laddr, raddr *net.UDPAddr) (*net.UDPConn, error)
|
||||
|
||||
// newReconnectingUDPConn returns a new udpConn that resolves hostPort every resolveTimeout, if the resolved address is
|
||||
// different than the current conn then the new address is dialed and the conn is swapped.
|
||||
func newReconnectingUDPConn(hostPort string, resolveTimeout time.Duration, resolveFunc resolveFunc, dialFunc dialFunc, logger log.Logger) (*reconnectingUDPConn, error) {
|
||||
conn := &reconnectingUDPConn{
|
||||
hostPort: hostPort,
|
||||
resolveFunc: resolveFunc,
|
||||
dialFunc: dialFunc,
|
||||
logger: logger,
|
||||
closeChan: make(chan struct{}),
|
||||
}
|
||||
|
||||
if err := conn.attemptResolveAndDial(); err != nil {
|
||||
logger.Error(fmt.Sprintf("failed resolving destination address on connection startup, with err: %q. retrying in %s", err.Error(), resolveTimeout))
|
||||
}
|
||||
|
||||
go conn.reconnectLoop(resolveTimeout)
|
||||
|
||||
return conn, nil
|
||||
}
|
||||
|
||||
func (c *reconnectingUDPConn) reconnectLoop(resolveTimeout time.Duration) {
|
||||
ticker := time.NewTicker(resolveTimeout)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-c.closeChan:
|
||||
return
|
||||
case <-ticker.C:
|
||||
if err := c.attemptResolveAndDial(); err != nil {
|
||||
c.logger.Error(err.Error())
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *reconnectingUDPConn) attemptResolveAndDial() error {
|
||||
newAddr, err := c.resolveFunc("udp", c.hostPort)
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to resolve new addr for host %q, with err: %w", c.hostPort, err)
|
||||
}
|
||||
|
||||
c.connMtx.RLock()
|
||||
curAddr := c.destAddr
|
||||
c.connMtx.RUnlock()
|
||||
|
||||
// dont attempt dial if an addr was successfully dialed previously and, resolved addr is the same as current conn
|
||||
if curAddr != nil && newAddr.String() == curAddr.String() {
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := c.attemptDialNewAddr(newAddr); err != nil {
|
||||
return fmt.Errorf("failed to dial newly resolved addr '%s', with err: %w", newAddr, err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *reconnectingUDPConn) attemptDialNewAddr(newAddr *net.UDPAddr) error {
|
||||
connUDP, err := c.dialFunc(newAddr.Network(), nil, newAddr)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if bufferBytes := int(atomic.LoadInt64(&c.bufferBytes)); bufferBytes != 0 {
|
||||
if err = connUDP.SetWriteBuffer(bufferBytes); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
c.connMtx.Lock()
|
||||
c.destAddr = newAddr
|
||||
// store prev to close later
|
||||
prevConn := c.conn
|
||||
c.conn = connUDP
|
||||
c.connMtx.Unlock()
|
||||
|
||||
if prevConn != nil {
|
||||
return prevConn.Close()
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// Write calls net.udpConn.Write, if it fails an attempt is made to connect to a new addr, if that succeeds the write is retried before returning
|
||||
func (c *reconnectingUDPConn) Write(b []byte) (int, error) {
|
||||
var bytesWritten int
|
||||
var err error
|
||||
|
||||
c.connMtx.RLock()
|
||||
if c.conn == nil {
|
||||
// if connection is not initialized indicate this with err in order to hook into retry logic
|
||||
err = fmt.Errorf("UDP connection not yet initialized, an address has not been resolved")
|
||||
} else {
|
||||
bytesWritten, err = c.conn.Write(b)
|
||||
}
|
||||
c.connMtx.RUnlock()
|
||||
|
||||
if err == nil {
|
||||
return bytesWritten, nil
|
||||
}
|
||||
|
||||
// attempt to resolve and dial new address in case that's the problem, if resolve and dial succeeds, try write again
|
||||
if reconnErr := c.attemptResolveAndDial(); reconnErr == nil {
|
||||
c.connMtx.RLock()
|
||||
defer c.connMtx.RUnlock()
|
||||
return c.conn.Write(b)
|
||||
}
|
||||
|
||||
// return original error if reconn fails
|
||||
return bytesWritten, err
|
||||
}
|
||||
|
||||
// Close stops the reconnectLoop, then closes the connection via net.udpConn 's implementation
|
||||
func (c *reconnectingUDPConn) Close() error {
|
||||
close(c.closeChan)
|
||||
|
||||
// acquire rw lock before closing conn to ensure calls to Write drain
|
||||
c.connMtx.Lock()
|
||||
defer c.connMtx.Unlock()
|
||||
|
||||
if c.conn != nil {
|
||||
return c.conn.Close()
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetWriteBuffer defers to the net.udpConn SetWriteBuffer implementation wrapped with a RLock. if no conn is currently held
|
||||
// and SetWriteBuffer is called store bufferBytes to be set for new conns
|
||||
func (c *reconnectingUDPConn) SetWriteBuffer(bytes int) error {
|
||||
var err error
|
||||
|
||||
c.connMtx.RLock()
|
||||
if c.conn != nil {
|
||||
err = c.conn.SetWriteBuffer(bytes)
|
||||
}
|
||||
c.connMtx.RUnlock()
|
||||
|
||||
if err == nil {
|
||||
atomic.StoreInt64(&c.bufferBytes, int64(bytes))
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
91
vendor/github.com/uber/jaeger-client-go/utils/udp_client.go
сгенерированный
поставляемый
91
vendor/github.com/uber/jaeger-client-go/utils/udp_client.go
сгенерированный
поставляемый
@@ -19,7 +19,9 @@ import (
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
"github.com/uber/jaeger-client-go/log"
|
||||
"github.com/uber/jaeger-client-go/thrift"
|
||||
|
||||
"github.com/uber/jaeger-client-go/thrift-gen/agent"
|
||||
@@ -35,41 +37,90 @@ type AgentClientUDP struct {
|
||||
agent.Agent
|
||||
io.Closer
|
||||
|
||||
connUDP *net.UDPConn
|
||||
connUDP udpConn
|
||||
client *agent.AgentClient
|
||||
maxPacketSize int // max size of datagram in bytes
|
||||
thriftBuffer *thrift.TMemoryBuffer // buffer used to calculate byte size of a span
|
||||
}
|
||||
|
||||
// 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
|
||||
type udpConn interface {
|
||||
Write([]byte) (int, error)
|
||||
SetWriteBuffer(int) error
|
||||
Close() error
|
||||
}
|
||||
|
||||
// AgentClientUDPParams allows specifying options for initializing an AgentClientUDP. An instance of this struct should
|
||||
// be passed to NewAgentClientUDPWithParams.
|
||||
type AgentClientUDPParams struct {
|
||||
HostPort string
|
||||
MaxPacketSize int
|
||||
Logger log.Logger
|
||||
DisableAttemptReconnecting bool
|
||||
AttemptReconnectInterval time.Duration
|
||||
}
|
||||
|
||||
// NewAgentClientUDPWithParams creates a client that sends spans to Jaeger Agent over UDP.
|
||||
func NewAgentClientUDPWithParams(params AgentClientUDPParams) (*AgentClientUDP, error) {
|
||||
// validate hostport
|
||||
if _, _, err := net.SplitHostPort(params.HostPort); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
thriftBuffer := thrift.NewTMemoryBufferLen(maxPacketSize)
|
||||
if params.MaxPacketSize == 0 {
|
||||
params.MaxPacketSize = UDPPacketMaxLength
|
||||
}
|
||||
|
||||
if params.Logger == nil {
|
||||
params.Logger = log.StdLogger
|
||||
}
|
||||
|
||||
if !params.DisableAttemptReconnecting && params.AttemptReconnectInterval == 0 {
|
||||
params.AttemptReconnectInterval = time.Second * 30
|
||||
}
|
||||
|
||||
thriftBuffer := thrift.NewTMemoryBufferLen(params.MaxPacketSize)
|
||||
protocolFactory := thrift.NewTCompactProtocolFactory()
|
||||
client := agent.NewAgentClientFactory(thriftBuffer, protocolFactory)
|
||||
|
||||
destAddr, err := net.ResolveUDPAddr("udp", hostPort)
|
||||
if err != nil {
|
||||
var connUDP udpConn
|
||||
var err error
|
||||
|
||||
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.AttemptReconnectInterval, net.ResolveUDPAddr, net.DialUDP, params.Logger)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
if err := connUDP.SetWriteBuffer(params.MaxPacketSize); 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 {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
clientUDP := &AgentClientUDP{
|
||||
return &AgentClientUDP{
|
||||
connUDP: connUDP,
|
||||
client: client,
|
||||
maxPacketSize: maxPacketSize,
|
||||
thriftBuffer: thriftBuffer}
|
||||
return clientUDP, nil
|
||||
maxPacketSize: params.MaxPacketSize,
|
||||
thriftBuffer: thriftBuffer,
|
||||
}, nil
|
||||
}
|
||||
|
||||
// NewAgentClientUDP creates a client that sends spans to Jaeger Agent over UDP.
|
||||
func NewAgentClientUDP(hostPort string, maxPacketSize int) (*AgentClientUDP, error) {
|
||||
return NewAgentClientUDPWithParams(AgentClientUDPParams{
|
||||
HostPort: hostPort,
|
||||
MaxPacketSize: maxPacketSize,
|
||||
})
|
||||
}
|
||||
|
||||
// EmitZipkinBatch implements EmitZipkinBatch() of Agent interface
|
||||
|
||||
Ссылка в новой задаче
Block a user