mirror of
https://github.com/moby/moby.git
synced 2026-08-05 07:30:57 +00:00
Make the docker proxy a standalone binary not a re-exec
Rather than re-execing docker as the proxy, create a new command docker-proxy that is much smaller to save memory in the case where there are a lot of procies being created. Also allows the proxy to be replaced, for example in Docker for Mac we have a proxy that proxies to osx instead of locally. This is the vendoring pull for https://github.com/docker/docker/pull/23312 Signed-off-by: Justin Cormack <justin.cormack@docker.com>
This commit is contained in:
@@ -7,6 +7,7 @@ docker = docker run --rm -it ${dockerargs} $$EXTRA_ARGS ${container_env} ${build
|
||||
ciargs = -e CIRCLECI -e "COVERALLS_TOKEN=$$COVERALLS_TOKEN" -e "INSIDECONTAINER=-incontainer=true"
|
||||
cidocker = docker run ${dockerargs} ${ciargs} $$EXTRA_ARGS ${container_env} ${build_image}
|
||||
CROSS_PLATFORMS = linux/amd64 linux/386 linux/arm windows/amd64
|
||||
export PATH := $(CURDIR)/bin:$(PATH)
|
||||
|
||||
all: ${build_image}.created build check integration-tests clean
|
||||
|
||||
@@ -24,10 +25,11 @@ build: ${build_image}.created
|
||||
build-local:
|
||||
@mkdir -p "bin"
|
||||
$(shell which godep) go build -tags experimental -o "bin/dnet" ./cmd/dnet
|
||||
$(shell which godep) go build -o "bin/docker-proxy" ./cmd/proxy
|
||||
|
||||
clean:
|
||||
@if [ -d bin ]; then \
|
||||
echo "Removing dnet binaries"; \
|
||||
echo "Removing dnet and proxy binaries"; \
|
||||
rm -rf bin; \
|
||||
fi
|
||||
|
||||
@@ -41,6 +43,7 @@ cross: ${build_image}.created
|
||||
|
||||
cross-local:
|
||||
$(shell which godep) go build -o "bin/dnet-$$GOOS-$$GOARCH" ./cmd/dnet
|
||||
$(shell which godep) go build -o "bin/docker-proxy-$$GOOS-$$GOARCH" ./cmd/proxy
|
||||
|
||||
check: ${build_image}.created
|
||||
@${docker} ./wrapmake.sh check-local
|
||||
@@ -102,4 +105,4 @@ circle-ci-check: ${build_image}.created
|
||||
circle-ci-build: ${build_image}.created
|
||||
@${cidocker} make build-local
|
||||
|
||||
circle-ci: circle-ci-check circle-ci-cross circle-ci-build integration-tests
|
||||
circle-ci: circle-ci-build circle-ci-check circle-ci-cross integration-tests
|
||||
|
||||
67
libnetwork/cmd/proxy/main.go
Normal file
67
libnetwork/cmd/proxy/main.go
Normal file
@@ -0,0 +1,67 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
func main() {
|
||||
f := os.NewFile(3, "signal-parent")
|
||||
host, container := parseHostContainerAddrs()
|
||||
|
||||
p, err := NewProxy(host, container)
|
||||
if err != nil {
|
||||
fmt.Fprintf(f, "1\n%s", err)
|
||||
f.Close()
|
||||
os.Exit(1)
|
||||
}
|
||||
go handleStopSignals(p)
|
||||
fmt.Fprint(f, "0\n")
|
||||
f.Close()
|
||||
|
||||
// Run will block until the proxy stops
|
||||
p.Run()
|
||||
}
|
||||
|
||||
// parseHostContainerAddrs parses the flags passed on reexec to create the TCP or UDP
|
||||
// net.Addrs to map the host and container ports
|
||||
func parseHostContainerAddrs() (host net.Addr, container net.Addr) {
|
||||
var (
|
||||
proto = flag.String("proto", "tcp", "proxy protocol")
|
||||
hostIP = flag.String("host-ip", "", "host ip")
|
||||
hostPort = flag.Int("host-port", -1, "host port")
|
||||
containerIP = flag.String("container-ip", "", "container ip")
|
||||
containerPort = flag.Int("container-port", -1, "container port")
|
||||
)
|
||||
|
||||
flag.Parse()
|
||||
|
||||
switch *proto {
|
||||
case "tcp":
|
||||
host = &net.TCPAddr{IP: net.ParseIP(*hostIP), Port: *hostPort}
|
||||
container = &net.TCPAddr{IP: net.ParseIP(*containerIP), Port: *containerPort}
|
||||
case "udp":
|
||||
host = &net.UDPAddr{IP: net.ParseIP(*hostIP), Port: *hostPort}
|
||||
container = &net.UDPAddr{IP: net.ParseIP(*containerIP), Port: *containerPort}
|
||||
default:
|
||||
log.Fatalf("unsupported protocol %s", *proto)
|
||||
}
|
||||
|
||||
return host, container
|
||||
}
|
||||
|
||||
func handleStopSignals(p Proxy) {
|
||||
s := make(chan os.Signal, 10)
|
||||
signal.Notify(s, os.Interrupt, syscall.SIGTERM)
|
||||
|
||||
for range s {
|
||||
p.Close()
|
||||
|
||||
os.Exit(0)
|
||||
}
|
||||
}
|
||||
219
libnetwork/cmd/proxy/network_proxy_test.go
Normal file
219
libnetwork/cmd/proxy/network_proxy_test.go
Normal file
@@ -0,0 +1,219 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
var _ = flag.Bool("incontainer", false, "Indicates if the test is running in a container")
|
||||
|
||||
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))
|
||||
}
|
||||
}
|
||||
37
libnetwork/cmd/proxy/proxy.go
Normal file
37
libnetwork/cmd/proxy/proxy.go
Normal file
@@ -0,0 +1,37 @@
|
||||
// docker-proxy provides a network Proxy interface and implementations for TCP
|
||||
// and UDP.
|
||||
package main
|
||||
|
||||
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"))
|
||||
}
|
||||
}
|
||||
31
libnetwork/cmd/proxy/stub_proxy.go
Normal file
31
libnetwork/cmd/proxy/stub_proxy.go
Normal file
@@ -0,0 +1,31 @@
|
||||
package main
|
||||
|
||||
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
|
||||
}
|
||||
96
libnetwork/cmd/proxy/tcp_proxy.go
Normal file
96
libnetwork/cmd/proxy/tcp_proxy.go
Normal file
@@ -0,0 +1,96 @@
|
||||
package main
|
||||
|
||||
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 }
|
||||
169
libnetwork/cmd/proxy/udp_proxy.go
Normal file
169
libnetwork/cmd/proxy/udp_proxy.go
Normal file
@@ -0,0 +1,169 @@
|
||||
package main
|
||||
|
||||
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")
|
||||
}
|
||||
@@ -90,7 +90,10 @@ func (pm *PortMapper) MapRange(container net.Addr, hostIP net.IP, hostPortStart,
|
||||
}
|
||||
|
||||
if useProxy {
|
||||
m.userlandProxy = newProxy(proto, hostIP, allocatedHostPort, container.(*net.TCPAddr).IP, container.(*net.TCPAddr).Port)
|
||||
m.userlandProxy, err = newProxy(proto, hostIP, allocatedHostPort, container.(*net.TCPAddr).IP, container.(*net.TCPAddr).Port)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
} else {
|
||||
m.userlandProxy = newDummyProxy(proto, hostIP, allocatedHostPort)
|
||||
}
|
||||
@@ -107,7 +110,10 @@ func (pm *PortMapper) MapRange(container net.Addr, hostIP net.IP, hostPortStart,
|
||||
}
|
||||
|
||||
if useProxy {
|
||||
m.userlandProxy = newProxy(proto, hostIP, allocatedHostPort, container.(*net.UDPAddr).IP, container.(*net.UDPAddr).Port)
|
||||
m.userlandProxy, err = newProxy(proto, hostIP, allocatedHostPort, container.(*net.UDPAddr).IP, container.(*net.UDPAddr).Port)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
} else {
|
||||
m.userlandProxy = newDummyProxy(proto, hostIP, allocatedHostPort)
|
||||
}
|
||||
|
||||
@@ -2,8 +2,8 @@ package portmapper
|
||||
|
||||
import "net"
|
||||
|
||||
func newMockProxyCommand(proto string, hostIP net.IP, hostPort int, containerIP net.IP, containerPort int) userlandProxy {
|
||||
return &mockProxyCommand{}
|
||||
func newMockProxyCommand(proto string, hostIP net.IP, hostPort int, containerIP net.IP, containerPort int) (userlandProxy, error) {
|
||||
return &mockProxyCommand{}, nil
|
||||
}
|
||||
|
||||
type mockProxyCommand struct {
|
||||
|
||||
@@ -1,29 +1,19 @@
|
||||
package portmapper
|
||||
|
||||
import (
|
||||
"flag"
|
||||
"fmt"
|
||||
"io"
|
||||
"io/ioutil"
|
||||
"log"
|
||||
"net"
|
||||
"os"
|
||||
"os/exec"
|
||||
"os/signal"
|
||||
"strconv"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"github.com/docker/docker/pkg/proxy"
|
||||
"github.com/docker/docker/pkg/reexec"
|
||||
)
|
||||
|
||||
const userlandProxyCommandName = "docker-proxy"
|
||||
|
||||
func init() {
|
||||
reexec.Register(userlandProxyCommandName, execProxy)
|
||||
}
|
||||
|
||||
type userlandProxy interface {
|
||||
Start() error
|
||||
Stop() error
|
||||
@@ -35,66 +25,15 @@ type proxyCommand struct {
|
||||
cmd *exec.Cmd
|
||||
}
|
||||
|
||||
// execProxy is the reexec function that is registered to start the userland proxies
|
||||
func execProxy() {
|
||||
f := os.NewFile(3, "signal-parent")
|
||||
host, container := parseHostContainerAddrs()
|
||||
func newProxyCommand(proto string, hostIP net.IP, hostPort int, containerIP net.IP, containerPort int) (userlandProxy, error) {
|
||||
cmd, err := exec.LookPath(userlandProxyCommandName)
|
||||
|
||||
p, err := proxy.NewProxy(host, container)
|
||||
if err != nil {
|
||||
fmt.Fprintf(f, "1\n%s", err)
|
||||
f.Close()
|
||||
os.Exit(1)
|
||||
}
|
||||
go handleStopSignals(p)
|
||||
fmt.Fprint(f, "0\n")
|
||||
f.Close()
|
||||
|
||||
// Run will block until the proxy stops
|
||||
p.Run()
|
||||
}
|
||||
|
||||
// parseHostContainerAddrs parses the flags passed on reexec to create the TCP or UDP
|
||||
// net.Addrs to map the host and container ports
|
||||
func parseHostContainerAddrs() (host net.Addr, container net.Addr) {
|
||||
var (
|
||||
proto = flag.String("proto", "tcp", "proxy protocol")
|
||||
hostIP = flag.String("host-ip", "", "host ip")
|
||||
hostPort = flag.Int("host-port", -1, "host port")
|
||||
containerIP = flag.String("container-ip", "", "container ip")
|
||||
containerPort = flag.Int("container-port", -1, "container port")
|
||||
)
|
||||
|
||||
flag.Parse()
|
||||
|
||||
switch *proto {
|
||||
case "tcp":
|
||||
host = &net.TCPAddr{IP: net.ParseIP(*hostIP), Port: *hostPort}
|
||||
container = &net.TCPAddr{IP: net.ParseIP(*containerIP), Port: *containerPort}
|
||||
case "udp":
|
||||
host = &net.UDPAddr{IP: net.ParseIP(*hostIP), Port: *hostPort}
|
||||
container = &net.UDPAddr{IP: net.ParseIP(*containerIP), Port: *containerPort}
|
||||
default:
|
||||
log.Fatalf("unsupported protocol %s", *proto)
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return host, container
|
||||
}
|
||||
|
||||
func handleStopSignals(p proxy.Proxy) {
|
||||
s := make(chan os.Signal, 10)
|
||||
signal.Notify(s, os.Interrupt, syscall.SIGTERM, syscall.SIGSTOP)
|
||||
|
||||
for range s {
|
||||
p.Close()
|
||||
|
||||
os.Exit(0)
|
||||
}
|
||||
}
|
||||
|
||||
func newProxyCommand(proto string, hostIP net.IP, hostPort int, containerIP net.IP, containerPort int) userlandProxy {
|
||||
args := []string{
|
||||
userlandProxyCommandName,
|
||||
cmd,
|
||||
"-proto", proto,
|
||||
"-host-ip", hostIP.String(),
|
||||
"-host-port", strconv.Itoa(hostPort),
|
||||
@@ -104,13 +43,13 @@ func newProxyCommand(proto string, hostIP net.IP, hostPort int, containerIP net.
|
||||
|
||||
return &proxyCommand{
|
||||
cmd: &exec.Cmd{
|
||||
Path: reexec.Self(),
|
||||
Path: cmd,
|
||||
Args: args,
|
||||
SysProcAttr: &syscall.SysProcAttr{
|
||||
Pdeathsig: syscall.SIGTERM, // send a sigterm to the proxy if the daemon process dies
|
||||
},
|
||||
},
|
||||
}
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (p *proxyCommand) Start() error {
|
||||
|
||||
Reference in New Issue
Block a user