Repository: moby/moby
File: cmd/docker-proxy/udp_proxy_linux.go (Go)

Code before:

		fromKey := newConnTrackKey(from)
		proxy.connTrackLock.Lock()
		cte, hit := proxy.connTrackTable[*fromKey]
		if !hit {
			proxyConn, err := net.DialUDP("udp", nil, proxy.backendAddr)
			if err != nil {
				log.Printf("Can't proxy a datagram to udp/%s: %s\n", proxy.backendAddr, err)
				proxy.connTrackLock.Unlock()
				continue
			}
			cte = newConnTrackEntry(proxyConn)
			proxy.connTrackTable[*fromKey] = cte

			daddr, err := readDestFromCmsg(oob, proxy.ipVer)
			if err != nil {
				log.Printf("Failed to parse control message: %v", err)
				proxy.connTrackLock.Unlock()
				continue
			}

			go proxy.replyLoop(cte, daddr, from, fromKey)
		}
		cte.mu.Lock()
		proxy.connTrackLock.Unlock()
		cte.conn.SetWriteDeadline(time.Now().Add(UDPConnTrackTimeout))
		for i := 0; i != read; {
			written, err := cte.conn.Write(readBuf[i:read])
			if err != nil {
				log.Printf("Can't proxy a datagram to udp/%s: %s\n", proxy.backendAddr, err)
				break
			}
			i += written
		}
		cte.mu.Unlock()
	}
}

func readDestFromCmsg(oob []byte, ipVer ipVersion) (_ net.IP, err error) {
	defer func() {
		// In case of partial upgrade / downgrade, docker-proxy could read
		// control messages from a socket which doesn't have the sockopt
		// IP_PKTINFO enabled. In that case, the control message will be all-0
		// and Go's ControlMessage.Parse() will report an 'invalid header
		// length' error. In that case, ignore the error and return an empty
		// daddr - the kernel will pick a source address for us anyway (but
		// maybe it'll be the wrong one).
		if err != nil && err.Error() == "invalid header length" {
			err = nil
		}
	}()

Review comment (suggestion):
Looks like the write deadline is new - is it meant to be in the first commit (fixing the write/close race)?

If it is, and there's a concern about a write blocking for such a long time, should the reply-loop's write have a deadline too? (Maybe I've missed the point?!)

Code after:
	}
}

// Run starts forwarding the traffic using UDP.
func (proxy *UDPProxy) Run() {
	readBuf := make([]byte, UDPBufSize)
	var oob []byte
	if proxy.ipVer == ip4 {
		oob = ipv4.NewControlMessage(ipv4.FlagDst)
	} else {
		oob = ipv6.NewControlMessage(ipv6.FlagDst)
	}

	for {
		read, _, _, from, err := proxy.listener.ReadMsgUDP(readBuf, oob)
		if err != nil {
			// The frontend listener socket might be closed by the signal
			// handler. In that case, don't log anything - it's not an error.
			if !errors.Is(err, net.ErrClosed) {
				log.Printf("Stopping proxy on udp/%v for udp/%v (%s)", proxy.frontendAddr, proxy.backendAddr, err)
			}
			break
		}

		fromKey := newConnTrackKey(from)
		proxy.connTrackLock.Lock()
		cte, hit := proxy.connTrackTable[*fromKey]
		if !hit {
			proxyConn, err := net.DialUDP("udp", nil, proxy.backendAddr)
			if err != nil {
				log.Printf("Can't proxy a datagram to udp/%s: %s\n", proxy.backendAddr, err)
				proxy.connTrackLock.Unlock()
				continue
			}
			cte = newConnTrackEntry(proxyConn)
			proxy.connTrackTable[*fromKey] = cte

			daddr, err := readDestFromCmsg(oob, proxy.ipVer)
			if err != nil {
				log.Printf("Failed to parse control message: %v", err)
				proxy.connTrackLock.Unlock()
				continue
			}

			go proxy.replyLoop(cte, daddr, from, fromKey)
		}
		cte.mu.Lock()
		proxy.connTrackLock.Unlock()
		cte.conn.SetWriteDeadline(time.Now().Add(proxy.connTrackTimeout))
		for i := 0; i != read; {
			written, err := cte.conn.Write(readBuf[i:read])