Make the docker proxy a standalone binary not a re-exec
This reduces memory usage with a lot of docker proxy processes. On Docker for Mac we are currently carrying a patch to replace the binary as we modify it to forward ports to the Mac rather than the Linux VM, this allows us to simply replace this binary in our packaging with one that has a compatible interface. This patch does not provide an easy way to substitute a binary as the interface is complex and there are few use cases, but where needed this can be done. Signed-off-by: Justin Cormack <justin.cormack@docker.com>
This commit is contained in:
Justin Cormack
parent
656c66dd50
commit
0682468431
14 changed files with 36 additions and 549 deletions
|
|
@ -131,6 +131,9 @@ clean() {
|
|||
findArgs+=( -path "vendor/src/$import" )
|
||||
done
|
||||
|
||||
# The docker proxy command is built from libnetwork
|
||||
findArgs+=( -or -path vendor/src/github.com/docker/libnetwork/cmd/proxy )
|
||||
|
||||
local IFS=$'\n'
|
||||
local prune=( $($find vendor -depth -type d -not '(' "${findArgs[@]}" ')') )
|
||||
unset IFS
|
||||
|
|
|
|||
|
|
@ -2,3 +2,4 @@
|
|||
|
||||
DOCKER_CLIENT_BINARY_NAME='docker'
|
||||
DOCKER_DAEMON_BINARY_NAME='dockerd'
|
||||
DOCKER_PROXY_BINARY_NAME='docker-proxy'
|
||||
|
|
|
|||
|
|
@ -22,6 +22,7 @@ override_dh_auto_install:
|
|||
mkdir -p debian/docker-engine/usr/bin
|
||||
cp -aT "$$(readlink -f bundles/$(VERSION)/dynbinary-client/docker)" debian/docker-engine/usr/bin/docker
|
||||
cp -aT "$$(readlink -f bundles/$(VERSION)/dynbinary-daemon/dockerd)" debian/docker-engine/usr/bin/dockerd
|
||||
cp -aT "$$(readlink -f bundles/$(VERSION)/dynbinary-daemon/docker-proxy)" debian/docker-engine/usr/bin/docker-proxy
|
||||
cp -aT /usr/local/bin/containerd debian/docker-engine/usr/bin/docker-containerd
|
||||
cp -aT /usr/local/bin/containerd-shim debian/docker-engine/usr/bin/docker-containerd-shim
|
||||
cp -aT /usr/local/bin/ctr debian/docker-engine/usr/bin/docker-containerd-ctr
|
||||
|
|
|
|||
|
|
@ -126,6 +126,7 @@ export DOCKER_GITCOMMIT=%{_gitcommit}
|
|||
install -d $RPM_BUILD_ROOT/%{_bindir}
|
||||
install -p -m 755 bundles/%{_origversion}/dynbinary-client/docker-%{_origversion} $RPM_BUILD_ROOT/%{_bindir}/docker
|
||||
install -p -m 755 bundles/%{_origversion}/dynbinary-daemon/dockerd-%{_origversion} $RPM_BUILD_ROOT/%{_bindir}/dockerd
|
||||
install -p -m 755 bundles/%{_origversion}/dynbinary-daemon/docker-proxy-%{_origversion} $RPM_BUILD_ROOT/%{_bindir}/docker-proxy
|
||||
|
||||
# install containerd
|
||||
install -p -m 755 /usr/local/bin/containerd $RPM_BUILD_ROOT/%{_bindir}/docker-containerd
|
||||
|
|
|
|||
|
|
@ -9,5 +9,8 @@ set -e
|
|||
export BINARY_SHORT_NAME="$DOCKER_DAEMON_BINARY_NAME"
|
||||
export SOURCE_PATH='./cmd/dockerd'
|
||||
source "${MAKEDIR}/.binary"
|
||||
export BINARY_SHORT_NAME="$DOCKER_PROXY_BINARY_NAME"
|
||||
export SOURCE_PATH='./vendor/src/github.com/docker/libnetwork/cmd/proxy'
|
||||
source "${MAKEDIR}/.binary"
|
||||
copy_containerd "$DEST" 'hash'
|
||||
)
|
||||
|
|
|
|||
|
|
@ -9,4 +9,7 @@ set -e
|
|||
export BUILDFLAGS=( "${BUILDFLAGS[@]/netgo /}" ) # disable netgo, since we don't need it for a dynamic binary
|
||||
export BUILDFLAGS=( "${BUILDFLAGS[@]/static_build /}" ) # we're not building a "static" binary here
|
||||
source "${MAKEDIR}/.binary"
|
||||
export BINARY_SHORT_NAME='docker-proxy'
|
||||
export SOURCE_PATH='./vendor/src/github.com/docker/libnetwork/cmd/proxy'
|
||||
source "${MAKEDIR}/.binary"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -5,6 +5,9 @@ BINARY_NAME="dockerd-$VERSION"
|
|||
BINARY_EXTENSION="$(binary_extension)"
|
||||
BINARY_FULLNAME="$BINARY_NAME$BINARY_EXTENSION"
|
||||
|
||||
PROXY_NAME="docker-proxy-$VERSION"
|
||||
PROXY_FULLNAME="$PROXY_NAME$BINARY_EXTENSION"
|
||||
|
||||
CLIENTBIN_NAME="docker-$VERSION"
|
||||
CLIENTBIN_FULLNAME="$CLIENTBIN_NAME$BINARY_EXTENSION"
|
||||
|
||||
|
|
@ -29,6 +32,21 @@ go build -compiler=gccgo \
|
|||
echo "Created binary: $DEST/$BINARY_FULLNAME"
|
||||
ln -sf "$BINARY_FULLNAME" "$DEST/dockerd$BINARY_EXTENSION"
|
||||
|
||||
go build -compiler=gccgo \
|
||||
-o "$DEST/$PROXY_FULLNAME" \
|
||||
"${BUILDFLAGS[@]}" \
|
||||
-gccgoflags "
|
||||
-g
|
||||
$EXTLDFLAGS_STATIC
|
||||
-Wl,--no-export-dynamic
|
||||
-ldl
|
||||
-pthread
|
||||
" \
|
||||
./vendor/src/github.com/docker/libnetwork/cmd/proxy
|
||||
|
||||
echo "Created binary: $DEST/$PROXY_FULLNAME"
|
||||
ln -sf "$PROXY_FULLNAME" "$DEST/docker-proxy$BINARY_EXTENSION"
|
||||
|
||||
copy_containerd "$DEST" "hash"
|
||||
hash_files "$DEST/$BINARY_FULLNAME"
|
||||
|
||||
|
|
|
|||
|
|
@ -7,4 +7,5 @@ rm -rf "$DEST"
|
|||
DEST="$(dirname $DEST)/binary-daemon"
|
||||
source "${MAKEDIR}/.binary-setup"
|
||||
install_binary "${DEST}/${DOCKER_DAEMON_BINARY_NAME}"
|
||||
install_binary "${DEST}/${DOCKER_PROXY_BINARY_NAME}"
|
||||
)
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ for d in "$CROSS/"*/*; do
|
|||
|
||||
BINARY_NAME="${DOCKER_CLIENT_BINARY_NAME}-$VERSION"
|
||||
DAEMON_BINARY_NAME="${DOCKER_DAEMON_BINARY_NAME}-$VERSION"
|
||||
PROXY_BINARY_NAME="${DOCKER_PROXY_BINARY_NAME}-$VERSION"
|
||||
BINARY_EXTENSION="$(export GOOS && binary_extension)"
|
||||
if [ "$GOOS" = 'windows' ]; then
|
||||
# if windows use a zip, not tgz
|
||||
|
|
@ -29,6 +30,7 @@ for d in "$CROSS/"*/*; do
|
|||
fi
|
||||
BINARY_FULLNAME="$BINARY_NAME$BINARY_EXTENSION"
|
||||
DAEMON_BINARY_FULLNAME="$DAEMON_BINARY_NAME$BINARY_EXTENSION"
|
||||
PROXY_BINARY_FULLNAME="$PROXY_BINARY_NAME$BINARY_EXTENSION"
|
||||
mkdir -p "$DEST/$GOOS/$GOARCH"
|
||||
TGZ="$DEST/$GOOS/$GOARCH/$BINARY_NAME$BUNDLE_EXTENSION"
|
||||
|
||||
|
|
@ -47,6 +49,9 @@ for d in "$CROSS/"*/*; do
|
|||
if [ -f "$d/$DAEMON_BINARY_FULLNAME" ]; then
|
||||
cp -L "$d/$DAEMON_BINARY_FULLNAME" "$TAR_PATH/${DOCKER_DAEMON_BINARY_NAME}${BINARY_EXTENSION}"
|
||||
fi
|
||||
if [ -f "$d/$PROXY_BINARY_FULLNAME" ]; then
|
||||
cp -L "$d/$PROXY_BINARY_FULLNAME" "$TAR_PATH/${DOCKER_PROXY_BINARY_NAME}${BINARY_EXTENSION}"
|
||||
fi
|
||||
|
||||
# copy over all the containerd binaries
|
||||
copy_containerd $TAR_PATH
|
||||
|
|
|
|||
|
|
@ -1,216 +0,0 @@
|
|||
package proxy
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
var testBuf = []byte("Buffalo buffalo Buffalo buffalo buffalo buffalo Buffalo buffalo")
|
||||
var testBufSize = len(testBuf)
|
||||
|
||||
type EchoServer interface {
|
||||
Run()
|
||||
Close()
|
||||
LocalAddr() net.Addr
|
||||
}
|
||||
|
||||
type TCPEchoServer struct {
|
||||
listener net.Listener
|
||||
testCtx *testing.T
|
||||
}
|
||||
|
||||
type UDPEchoServer struct {
|
||||
conn net.PacketConn
|
||||
testCtx *testing.T
|
||||
}
|
||||
|
||||
func NewEchoServer(t *testing.T, proto, address string) EchoServer {
|
||||
var server EchoServer
|
||||
if strings.HasPrefix(proto, "tcp") {
|
||||
listener, err := net.Listen(proto, address)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
server = &TCPEchoServer{listener: listener, testCtx: t}
|
||||
} else {
|
||||
socket, err := net.ListenPacket(proto, address)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
server = &UDPEchoServer{conn: socket, testCtx: t}
|
||||
}
|
||||
return server
|
||||
}
|
||||
|
||||
func (server *TCPEchoServer) Run() {
|
||||
go func() {
|
||||
for {
|
||||
client, err := server.listener.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
go func(client net.Conn) {
|
||||
if _, err := io.Copy(client, client); err != nil {
|
||||
server.testCtx.Logf("can't echo to the client: %v\n", err.Error())
|
||||
}
|
||||
client.Close()
|
||||
}(client)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (server *TCPEchoServer) LocalAddr() net.Addr { return server.listener.Addr() }
|
||||
func (server *TCPEchoServer) Close() { server.listener.Close() }
|
||||
|
||||
func (server *UDPEchoServer) Run() {
|
||||
go func() {
|
||||
readBuf := make([]byte, 1024)
|
||||
for {
|
||||
read, from, err := server.conn.ReadFrom(readBuf)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
for i := 0; i != read; {
|
||||
written, err := server.conn.WriteTo(readBuf[i:read], from)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
i += written
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (server *UDPEchoServer) LocalAddr() net.Addr { return server.conn.LocalAddr() }
|
||||
func (server *UDPEchoServer) Close() { server.conn.Close() }
|
||||
|
||||
func testProxyAt(t *testing.T, proto string, proxy Proxy, addr string) {
|
||||
defer proxy.Close()
|
||||
go proxy.Run()
|
||||
client, err := net.Dial(proto, addr)
|
||||
if err != nil {
|
||||
t.Fatalf("Can't connect to the proxy: %v", err)
|
||||
}
|
||||
defer client.Close()
|
||||
client.SetDeadline(time.Now().Add(10 * time.Second))
|
||||
if _, err = client.Write(testBuf); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
recvBuf := make([]byte, testBufSize)
|
||||
if _, err = client.Read(recvBuf); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(testBuf, recvBuf) {
|
||||
t.Fatal(fmt.Errorf("Expected [%v] but got [%v]", testBuf, recvBuf))
|
||||
}
|
||||
}
|
||||
|
||||
func testProxy(t *testing.T, proto string, proxy Proxy) {
|
||||
testProxyAt(t, proto, proxy, proxy.FrontendAddr().String())
|
||||
}
|
||||
|
||||
func TestTCP4Proxy(t *testing.T) {
|
||||
backend := NewEchoServer(t, "tcp", "127.0.0.1:0")
|
||||
defer backend.Close()
|
||||
backend.Run()
|
||||
frontendAddr := &net.TCPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0}
|
||||
proxy, err := NewProxy(frontendAddr, backend.LocalAddr())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
testProxy(t, "tcp", proxy)
|
||||
}
|
||||
|
||||
func TestTCP6Proxy(t *testing.T) {
|
||||
backend := NewEchoServer(t, "tcp", "[::1]:0")
|
||||
defer backend.Close()
|
||||
backend.Run()
|
||||
frontendAddr := &net.TCPAddr{IP: net.IPv6loopback, Port: 0}
|
||||
proxy, err := NewProxy(frontendAddr, backend.LocalAddr())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
testProxy(t, "tcp", proxy)
|
||||
}
|
||||
|
||||
func TestTCPDualStackProxy(t *testing.T) {
|
||||
// If I understand `godoc -src net favoriteAddrFamily` (used by the
|
||||
// net.Listen* functions) correctly this should work, but it doesn't.
|
||||
t.Skip("No support for dual stack yet")
|
||||
backend := NewEchoServer(t, "tcp", "[::1]:0")
|
||||
defer backend.Close()
|
||||
backend.Run()
|
||||
frontendAddr := &net.TCPAddr{IP: net.IPv6loopback, Port: 0}
|
||||
proxy, err := NewProxy(frontendAddr, backend.LocalAddr())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ipv4ProxyAddr := &net.TCPAddr{
|
||||
IP: net.IPv4(127, 0, 0, 1),
|
||||
Port: proxy.FrontendAddr().(*net.TCPAddr).Port,
|
||||
}
|
||||
testProxyAt(t, "tcp", proxy, ipv4ProxyAddr.String())
|
||||
}
|
||||
|
||||
func TestUDP4Proxy(t *testing.T) {
|
||||
backend := NewEchoServer(t, "udp", "127.0.0.1:0")
|
||||
defer backend.Close()
|
||||
backend.Run()
|
||||
frontendAddr := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0}
|
||||
proxy, err := NewProxy(frontendAddr, backend.LocalAddr())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
testProxy(t, "udp", proxy)
|
||||
}
|
||||
|
||||
func TestUDP6Proxy(t *testing.T) {
|
||||
backend := NewEchoServer(t, "udp", "[::1]:0")
|
||||
defer backend.Close()
|
||||
backend.Run()
|
||||
frontendAddr := &net.UDPAddr{IP: net.IPv6loopback, Port: 0}
|
||||
proxy, err := NewProxy(frontendAddr, backend.LocalAddr())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
testProxy(t, "udp", proxy)
|
||||
}
|
||||
|
||||
func TestUDPWriteError(t *testing.T) {
|
||||
frontendAddr := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 0}
|
||||
// Hopefully, this port will be free: */
|
||||
backendAddr := &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1), Port: 25587}
|
||||
proxy, err := NewProxy(frontendAddr, backendAddr)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer proxy.Close()
|
||||
go proxy.Run()
|
||||
client, err := net.Dial("udp", "127.0.0.1:25587")
|
||||
if err != nil {
|
||||
t.Fatalf("Can't connect to the proxy: %v", err)
|
||||
}
|
||||
defer client.Close()
|
||||
// Make sure the proxy doesn't stop when there is no actual backend:
|
||||
client.Write(testBuf)
|
||||
client.Write(testBuf)
|
||||
backend := NewEchoServer(t, "udp", "127.0.0.1:25587")
|
||||
defer backend.Close()
|
||||
backend.Run()
|
||||
client.SetDeadline(time.Now().Add(10 * time.Second))
|
||||
if _, err = client.Write(testBuf); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
recvBuf := make([]byte, testBufSize)
|
||||
if _, err = client.Read(recvBuf); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !bytes.Equal(testBuf, recvBuf) {
|
||||
t.Fatal(fmt.Errorf("Expected [%v] but got [%v]", testBuf, recvBuf))
|
||||
}
|
||||
}
|
||||
|
|
@ -1,37 +0,0 @@
|
|||
// Package proxy provides a network Proxy interface and implementations for TCP
|
||||
// and UDP.
|
||||
package proxy
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"net"
|
||||
)
|
||||
|
||||
// Proxy defines the behavior of a proxy. It forwards traffic back and forth
|
||||
// between two endpoints : the frontend and the backend.
|
||||
// It can be used to do software port-mapping between two addresses.
|
||||
// e.g. forward all traffic between the frontend (host) 127.0.0.1:3000
|
||||
// to the backend (container) at 172.17.42.108:4000.
|
||||
type Proxy interface {
|
||||
// Run starts forwarding traffic back and forth between the front
|
||||
// and back-end addresses.
|
||||
Run()
|
||||
// Close stops forwarding traffic and close both ends of the Proxy.
|
||||
Close()
|
||||
// FrontendAddr returns the address on which the proxy is listening.
|
||||
FrontendAddr() net.Addr
|
||||
// BackendAddr returns the proxied address.
|
||||
BackendAddr() net.Addr
|
||||
}
|
||||
|
||||
// NewProxy creates a Proxy according to the specified frontendAddr and backendAddr.
|
||||
func NewProxy(frontendAddr, backendAddr net.Addr) (Proxy, error) {
|
||||
switch frontendAddr.(type) {
|
||||
case *net.UDPAddr:
|
||||
return NewUDPProxy(frontendAddr.(*net.UDPAddr), backendAddr.(*net.UDPAddr))
|
||||
case *net.TCPAddr:
|
||||
return NewTCPProxy(frontendAddr.(*net.TCPAddr), backendAddr.(*net.TCPAddr))
|
||||
default:
|
||||
panic(fmt.Errorf("Unsupported protocol"))
|
||||
}
|
||||
}
|
||||
|
|
@ -1,31 +0,0 @@
|
|||
package proxy
|
||||
|
||||
import (
|
||||
"net"
|
||||
)
|
||||
|
||||
// StubProxy is a proxy that is a stub (does nothing).
|
||||
type StubProxy struct {
|
||||
frontendAddr net.Addr
|
||||
backendAddr net.Addr
|
||||
}
|
||||
|
||||
// Run does nothing.
|
||||
func (p *StubProxy) Run() {}
|
||||
|
||||
// Close does nothing.
|
||||
func (p *StubProxy) Close() {}
|
||||
|
||||
// FrontendAddr returns the frontend address.
|
||||
func (p *StubProxy) FrontendAddr() net.Addr { return p.frontendAddr }
|
||||
|
||||
// BackendAddr returns the backend address.
|
||||
func (p *StubProxy) BackendAddr() net.Addr { return p.backendAddr }
|
||||
|
||||
// NewStubProxy creates a new StubProxy
|
||||
func NewStubProxy(frontendAddr, backendAddr net.Addr) (Proxy, error) {
|
||||
return &StubProxy{
|
||||
frontendAddr: frontendAddr,
|
||||
backendAddr: backendAddr,
|
||||
}, nil
|
||||
}
|
||||
|
|
@ -1,96 +0,0 @@
|
|||
package proxy
|
||||
|
||||
import (
|
||||
"io"
|
||||
"net"
|
||||
"sync"
|
||||
"syscall"
|
||||
|
||||
"github.com/Sirupsen/logrus"
|
||||
)
|
||||
|
||||
// TCPProxy is a proxy for TCP connections. It implements the Proxy interface to
|
||||
// handle TCP traffic forwarding between the frontend and backend addresses.
|
||||
type TCPProxy struct {
|
||||
listener *net.TCPListener
|
||||
frontendAddr *net.TCPAddr
|
||||
backendAddr *net.TCPAddr
|
||||
}
|
||||
|
||||
// NewTCPProxy creates a new TCPProxy.
|
||||
func NewTCPProxy(frontendAddr, backendAddr *net.TCPAddr) (*TCPProxy, error) {
|
||||
listener, err := net.ListenTCP("tcp", frontendAddr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
// If the port in frontendAddr was 0 then ListenTCP will have a picked
|
||||
// a port to listen on, hence the call to Addr to get that actual port:
|
||||
return &TCPProxy{
|
||||
listener: listener,
|
||||
frontendAddr: listener.Addr().(*net.TCPAddr),
|
||||
backendAddr: backendAddr,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (proxy *TCPProxy) clientLoop(client *net.TCPConn, quit chan bool) {
|
||||
backend, err := net.DialTCP("tcp", nil, proxy.backendAddr)
|
||||
if err != nil {
|
||||
logrus.Printf("Can't forward traffic to backend tcp/%v: %s\n", proxy.backendAddr, err)
|
||||
client.Close()
|
||||
return
|
||||
}
|
||||
|
||||
var wg sync.WaitGroup
|
||||
var broker = func(to, from *net.TCPConn) {
|
||||
if _, err := io.Copy(to, from); err != nil {
|
||||
// If the socket we are writing to is shutdown with
|
||||
// SHUT_WR, forward it to the other end of the pipe:
|
||||
if err, ok := err.(*net.OpError); ok && err.Err == syscall.EPIPE {
|
||||
from.CloseWrite()
|
||||
}
|
||||
}
|
||||
to.CloseRead()
|
||||
wg.Done()
|
||||
}
|
||||
|
||||
wg.Add(2)
|
||||
go broker(client, backend)
|
||||
go broker(backend, client)
|
||||
|
||||
finish := make(chan struct{})
|
||||
go func() {
|
||||
wg.Wait()
|
||||
close(finish)
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-quit:
|
||||
case <-finish:
|
||||
}
|
||||
client.Close()
|
||||
backend.Close()
|
||||
<-finish
|
||||
}
|
||||
|
||||
// Run starts forwarding the traffic using TCP.
|
||||
func (proxy *TCPProxy) Run() {
|
||||
quit := make(chan bool)
|
||||
defer close(quit)
|
||||
for {
|
||||
client, err := proxy.listener.Accept()
|
||||
if err != nil {
|
||||
logrus.Printf("Stopping proxy on tcp/%v for tcp/%v (%s)", proxy.frontendAddr, proxy.backendAddr, err)
|
||||
return
|
||||
}
|
||||
go proxy.clientLoop(client.(*net.TCPConn), quit)
|
||||
}
|
||||
}
|
||||
|
||||
// Close stops forwarding the traffic.
|
||||
func (proxy *TCPProxy) Close() { proxy.listener.Close() }
|
||||
|
||||
// FrontendAddr returns the TCP address on which the proxy is listening.
|
||||
func (proxy *TCPProxy) FrontendAddr() net.Addr { return proxy.frontendAddr }
|
||||
|
||||
// BackendAddr returns the TCP proxied address.
|
||||
func (proxy *TCPProxy) BackendAddr() net.Addr { return proxy.backendAddr }
|
||||
|
|
@ -1,169 +0,0 @@
|
|||
package proxy
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"net"
|
||||
"strings"
|
||||
"sync"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/Sirupsen/logrus"
|
||||
)
|
||||
|
||||
const (
|
||||
// UDPConnTrackTimeout is the timeout used for UDP connection tracking
|
||||
UDPConnTrackTimeout = 90 * time.Second
|
||||
// UDPBufSize is the buffer size for the UDP proxy
|
||||
UDPBufSize = 65507
|
||||
)
|
||||
|
||||
// A net.Addr where the IP is split into two fields so you can use it as a key
|
||||
// in a map:
|
||||
type connTrackKey struct {
|
||||
IPHigh uint64
|
||||
IPLow uint64
|
||||
Port int
|
||||
}
|
||||
|
||||
func newConnTrackKey(addr *net.UDPAddr) *connTrackKey {
|
||||
if len(addr.IP) == net.IPv4len {
|
||||
return &connTrackKey{
|
||||
IPHigh: 0,
|
||||
IPLow: uint64(binary.BigEndian.Uint32(addr.IP)),
|
||||
Port: addr.Port,
|
||||
}
|
||||
}
|
||||
return &connTrackKey{
|
||||
IPHigh: binary.BigEndian.Uint64(addr.IP[:8]),
|
||||
IPLow: binary.BigEndian.Uint64(addr.IP[8:]),
|
||||
Port: addr.Port,
|
||||
}
|
||||
}
|
||||
|
||||
type connTrackMap map[connTrackKey]*net.UDPConn
|
||||
|
||||
// UDPProxy is proxy for which handles UDP datagrams. It implements the Proxy
|
||||
// interface to handle UDP traffic forwarding between the frontend and backend
|
||||
// addresses.
|
||||
type UDPProxy struct {
|
||||
listener *net.UDPConn
|
||||
frontendAddr *net.UDPAddr
|
||||
backendAddr *net.UDPAddr
|
||||
connTrackTable connTrackMap
|
||||
connTrackLock sync.Mutex
|
||||
}
|
||||
|
||||
// NewUDPProxy creates a new UDPProxy.
|
||||
func NewUDPProxy(frontendAddr, backendAddr *net.UDPAddr) (*UDPProxy, error) {
|
||||
listener, err := net.ListenUDP("udp", frontendAddr)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &UDPProxy{
|
||||
listener: listener,
|
||||
frontendAddr: listener.LocalAddr().(*net.UDPAddr),
|
||||
backendAddr: backendAddr,
|
||||
connTrackTable: make(connTrackMap),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (proxy *UDPProxy) replyLoop(proxyConn *net.UDPConn, clientAddr *net.UDPAddr, clientKey *connTrackKey) {
|
||||
defer func() {
|
||||
proxy.connTrackLock.Lock()
|
||||
delete(proxy.connTrackTable, *clientKey)
|
||||
proxy.connTrackLock.Unlock()
|
||||
proxyConn.Close()
|
||||
}()
|
||||
|
||||
readBuf := make([]byte, UDPBufSize)
|
||||
for {
|
||||
proxyConn.SetReadDeadline(time.Now().Add(UDPConnTrackTimeout))
|
||||
again:
|
||||
read, err := proxyConn.Read(readBuf)
|
||||
if err != nil {
|
||||
if err, ok := err.(*net.OpError); ok && err.Err == syscall.ECONNREFUSED {
|
||||
// This will happen if the last write failed
|
||||
// (e.g: nothing is actually listening on the
|
||||
// proxied port on the container), ignore it
|
||||
// and continue until UDPConnTrackTimeout
|
||||
// expires:
|
||||
goto again
|
||||
}
|
||||
return
|
||||
}
|
||||
for i := 0; i != read; {
|
||||
written, err := proxy.listener.WriteToUDP(readBuf[i:read], clientAddr)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
i += written
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Run starts forwarding the traffic using UDP.
|
||||
func (proxy *UDPProxy) Run() {
|
||||
readBuf := make([]byte, UDPBufSize)
|
||||
for {
|
||||
read, from, err := proxy.listener.ReadFromUDP(readBuf)
|
||||
if err != nil {
|
||||
// NOTE: Apparently ReadFrom doesn't return
|
||||
// ECONNREFUSED like Read do (see comment in
|
||||
// UDPProxy.replyLoop)
|
||||
if !isClosedError(err) {
|
||||
logrus.Printf("Stopping proxy on udp/%v for udp/%v (%s)", proxy.frontendAddr, proxy.backendAddr, err)
|
||||
}
|
||||
break
|
||||
}
|
||||
|
||||
fromKey := newConnTrackKey(from)
|
||||
proxy.connTrackLock.Lock()
|
||||
proxyConn, hit := proxy.connTrackTable[*fromKey]
|
||||
if !hit {
|
||||
proxyConn, err = net.DialUDP("udp", nil, proxy.backendAddr)
|
||||
if err != nil {
|
||||
logrus.Printf("Can't proxy a datagram to udp/%s: %s\n", proxy.backendAddr, err)
|
||||
proxy.connTrackLock.Unlock()
|
||||
continue
|
||||
}
|
||||
proxy.connTrackTable[*fromKey] = proxyConn
|
||||
go proxy.replyLoop(proxyConn, from, fromKey)
|
||||
}
|
||||
proxy.connTrackLock.Unlock()
|
||||
for i := 0; i != read; {
|
||||
written, err := proxyConn.Write(readBuf[i:read])
|
||||
if err != nil {
|
||||
logrus.Printf("Can't proxy a datagram to udp/%s: %s\n", proxy.backendAddr, err)
|
||||
break
|
||||
}
|
||||
i += written
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Close stops forwarding the traffic.
|
||||
func (proxy *UDPProxy) Close() {
|
||||
proxy.listener.Close()
|
||||
proxy.connTrackLock.Lock()
|
||||
defer proxy.connTrackLock.Unlock()
|
||||
for _, conn := range proxy.connTrackTable {
|
||||
conn.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// FrontendAddr returns the UDP address on which the proxy is listening.
|
||||
func (proxy *UDPProxy) FrontendAddr() net.Addr { return proxy.frontendAddr }
|
||||
|
||||
// BackendAddr returns the proxied UDP address.
|
||||
func (proxy *UDPProxy) BackendAddr() net.Addr { return proxy.backendAddr }
|
||||
|
||||
func isClosedError(err error) bool {
|
||||
/* This comparison is ugly, but unfortunately, net.go doesn't export errClosing.
|
||||
* See:
|
||||
* http://golang.org/src/pkg/net/net.go
|
||||
* https://code.google.com/p/go/issues/detail?id=4337
|
||||
* https://groups.google.com/forum/#!msg/golang-nuts/0_aaCvBmOcM/SptmDyX1XJMJ
|
||||
*/
|
||||
return strings.HasSuffix(err.Error(), "use of closed network connection")
|
||||
}
|
||||
Loading…
Reference in a new issue