Add simple threading to UDP packet handling, dump message when UDP buffers likely overflowed. #196
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -25,6 +25,7 @@ import ( | |
| "sort" | ||
| "strconv" | ||
| "strings" | ||
| "sync/atomic" | ||
| "time" | ||
| "unicode/utf8" | ||
|
|
||
|
|
@@ -34,6 +35,7 @@ import ( | |
|
|
||
| "github.com/prometheus/statsd_exporter/pkg/clock" | ||
| "github.com/prometheus/statsd_exporter/pkg/mapper" | ||
| "github.com/prometheus/statsd_exporter/pkg/telemetry" | ||
| ) | ||
|
|
||
| const ( | ||
|
|
@@ -639,17 +641,67 @@ samples: | |
| } | ||
|
|
||
| type StatsDUDPListener struct { | ||
| conn *net.UDPConn | ||
| bufferWatcher *telemetry.BufferWatcher | ||
| bufferWarning uint32 | ||
| } | ||
|
|
||
| func (l *StatsDUDPListener) Listen(threadCount int, packetHandlers int, e chan<- Events) { | ||
| bufferWatcher, err := telemetry.NewBufferWatcher(l.conn) | ||
| l.bufferWatcher = bufferWatcher | ||
| shouldWatchBuffer := true | ||
|
|
||
| if err != nil { | ||
| shouldWatchBuffer = false | ||
| log.Debugf("Unable to watch UDP buffer size due to: %v", err) | ||
| } | ||
|
|
||
| l.bufferWarning = 0 | ||
|
|
||
| concurrentHandlersPerThread := packetHandlers / threadCount | ||
|
|
||
| for i := 0; i < threadCount; i++ { | ||
| go l.Listener(e, concurrentHandlersPerThread, shouldWatchBuffer) | ||
| } | ||
| } | ||
|
|
||
| func (l *StatsDUDPListener) Listener(e chan<- Events, concurrentPacketHandlers int, shouldWatchBuffer bool) { | ||
| var sem = make(chan struct{}, concurrentPacketHandlers) | ||
| buf := make([]byte, 65535) | ||
| for { | ||
| n, _, err := l.conn.ReadFromUDP(buf) | ||
| if err != nil { | ||
| log.Fatal(err) | ||
| } | ||
| if shouldWatchBuffer && atomic.LoadUint32(&l.bufferWarning) != 1 { | ||
| go l.watchBufferSize() | ||
| } | ||
|
|
||
| data := append([]byte(nil), buf[0:n]...) | ||
| select { | ||
| case sem <- struct{}{}: | ||
| { | ||
| go func() { | ||
| l.handlePacket(data[0:n], e) | ||
|
There was a problem hiding this comment. is it safe to call handlePacket concurrently? why a semaphore and goroutines per There was a problem hiding this comment. It is safe! At least, I ran this with a Re: Semaphore - I think that is a great suggestion, this was originally @AustinSchoen's work, lemme collab with him. The idea of reading UDP packets as fast as we can and dropping them if the processing can't keep up instead of blocking (but keeping track of that drop tally) is so elegant, very +++ to headed down that path. That should totally alleviate my need for procfs metrics. There was a problem hiding this comment. all credit for that idea goes to @rtreffer! There was a problem hiding this comment. I believe I went with a semaphore because it was the first example of a tunable queue of workers that I could find, there are multiple ways of handling this model. We were still early in the experimentation phase as to what amount and kind of concurrency we needed to handle the firehose of UDP packets coming at our exporters. Regardless I don't think we'd want to just drop data on the floor if we don't have the capacity to handle it in every situation. Those packets should back up into the protocol buffer while waiting to be read by the configured number of handlers. In situations where a system is overloaded only for a few (mili)seconds it would be better to block read off the protocol buffer and wait to handle them than to drop and log that we lost packets IMO. There was a problem hiding this comment.
I agree, and I don't propose that we do that. We can buffer again after the initial receive. However, at some point we have to drop, and we want to know when that happens. I would rather guarantee that that happens in a place where we can easily observe it (i.e. in user-lang Go code) than introduce complex platform-dependent monitoring of the socket buffers themselves. There was a problem hiding this comment. With "buffer again" I mean stuff them into a buffered channel, we don't have to get fancy about it. |
||
| <-sem | ||
| }() | ||
| } | ||
|
|
||
| default: | ||
| l.handlePacket(data[0:n], e) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| func (l *StatsDUDPListener) watchBufferSize() { | ||
| readBufferSize, err := l.bufferWatcher.GetSocketQueue() | ||
| if err != nil { | ||
| log.Debugf("Failed to watch UDP buffer size due to: %v", err) | ||
| } else { | ||
| if readBufferSize+65535 > l.bufferWatcher.ReadBuffer { | ||
| atomic.StoreUint32(&l.bufferWarning, 1) | ||
| log.Info("UDP buffer size is high enough that packet drops may have occurred. You may need to allocate more resources for the exporter, or increase the statsd.read-buffer parameter and the kernel parameter net.core.rmem_max") | ||
| } | ||
| } | ||
| } | ||
|
|
||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -10,6 +10,7 @@ require ( | |
| github.com/howeyc/fsnotify v0.0.0-20151003194602-f0c08ee9c607 | ||
| github.com/kr/pretty v0.1.0 // indirect | ||
| github.com/mattn/go-isatty v0.0.4 // indirect | ||
| github.com/mdlayher/netlink v0.0.0-20190313131330-258ea9dff42c | ||
|
There was a problem hiding this comment. I went with this package after it's use in https://github.com/prometheus/node_exporter |
||
| github.com/onsi/ginkgo v1.7.0 // indirect | ||
| github.com/onsi/gomega v1.4.3 // indirect | ||
| github.com/pmezard/go-difflib v1.0.0 // indirect | ||
|
|
@@ -19,7 +20,7 @@ require ( | |
| github.com/sergi/go-diff v1.0.0 // indirect | ||
| github.com/sirupsen/logrus v1.0.3 // indirect | ||
| github.com/stretchr/testify v1.2.2 // indirect | ||
| golang.org/x/sys v0.0.0-20190312061237-fead79001313 | ||
| gopkg.in/airbrake/gobrake.v2 v2.0.9 // indirect | ||
| gopkg.in/alecthomas/kingpin.v2 v2.2.5 | ||
| gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| package telemetry | ||
|
|
||
| // Copyright 2019 The Prometheus 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. | ||
|
|
||
| import ( | ||
| "net" | ||
| ) | ||
|
|
||
| type BufferWatcher struct { | ||
| uAddr *net.UDPAddr | ||
| ReadBuffer int | ||
| } |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,160 @@ | ||
| // +build linux | ||
|
|
||
| // Copyright 2019 The Prometheus 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 telemetry | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "encoding/binary" | ||
| "errors" | ||
| "net" | ||
| "strconv" | ||
| "syscall" | ||
| "unsafe" | ||
|
|
||
| "github.com/mdlayher/netlink" | ||
| "golang.org/x/sys/unix" | ||
| ) | ||
|
|
||
| const SOCK_DIAG_BY_FAMILY = 20 | ||
|
|
||
| type inetDiagSockId struct { | ||
| SourcePort uint16 | ||
| DestPort uint16 | ||
| SourceAddress [4]uint32 | ||
| DestAddress [4]uint32 | ||
| If uint32 | ||
| Cookie [2]uint32 | ||
| } | ||
|
|
||
| type inetDiagReqV2 struct { | ||
| Family uint8 | ||
| Protocol uint8 | ||
| Ext uint8 | ||
| _ uint8 | ||
| States uint32 | ||
| Id inetDiagSockId | ||
| } | ||
|
|
||
| type InetDiagMsgData struct { | ||
| Family uint8 | ||
| State uint8 | ||
| Timer uint8 | ||
| Retrans uint8 | ||
| Id inetDiagSockId | ||
| Expires uint32 | ||
| Rqueue uint32 | ||
| Wqueue uint32 | ||
| Uid uint32 | ||
| Inode uint32 | ||
| } | ||
|
|
||
| // MarshalBinary marshals a Message into a byte slice. | ||
| func (m inetDiagReqV2) MarshalBinary() ([]byte, error) { | ||
| var buf bytes.Buffer | ||
| if err := binary.Write(&buf, binary.LittleEndian, &m); err != nil { | ||
| return nil, err | ||
| } | ||
| return buf.Bytes(), nil | ||
| } | ||
|
|
||
| func convert_addr_to_int32(ip []byte) (ret [4]uint32) { | ||
|
There was a problem hiding this comment. please use camelCase instead of snake_case |
||
| buf := bytes.NewBuffer(ip) | ||
| binary.Read(buf, binary.BigEndian, &ret) | ||
| return | ||
| } | ||
|
|
||
| func convert_port_to_u16(port int) uint16 { | ||
| uport := uint16(port) | ||
| portdat := make([]byte, 6) | ||
| binary.LittleEndian.PutUint16(portdat, uport) | ||
| uport = binary.BigEndian.Uint16(portdat) | ||
| return uport | ||
| } | ||
|
|
||
| func (b *BufferWatcher) GetSocketQueue() (int, error) { | ||
| c, err := netlink.Dial(unix.NETLINK_SOCK_DIAG, nil) | ||
| if err != nil { | ||
| return 0, err | ||
| } | ||
| defer c.Close() | ||
|
|
||
| r := inetDiagReqV2{ | ||
| Family: unix.AF_INET6, | ||
| Protocol: unix.IPPROTO_UDP, | ||
| Ext: 255, | ||
| Id: inetDiagSockId{ | ||
| SourcePort: convert_port_to_u16(b.uAddr.Port), | ||
| SourceAddress: convert_addr_to_int32(b.uAddr.IP), | ||
| }, | ||
| States: 0xffffffff, | ||
| } | ||
|
|
||
| data, err := r.MarshalBinary() | ||
| if err != nil { | ||
| return 0, err | ||
| } | ||
|
|
||
| req := netlink.Message{ | ||
| Header: netlink.Header{ | ||
| Flags: netlink.Root | netlink.Match | | ||
| netlink.Request, | ||
| Type: SOCK_DIAG_BY_FAMILY, | ||
| }, | ||
| Data: data, | ||
| } | ||
|
|
||
| // Perform a request, receive replies, and validate the replies | ||
| msgs, err := c.Execute(req) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| msg_count := len(msgs) | ||
|
|
||
| if msg_count == 1 { | ||
| m := msgs[0] | ||
| var data *InetDiagMsgData = *(**InetDiagMsgData)(unsafe.Pointer(&m.Data)) | ||
|
|
||
| return int(data.Rqueue), nil | ||
| } else { | ||
| return 0, errors.New("Netlink returned an unexpected number of sockets: " + strconv.Itoa(msg_count)) | ||
| } | ||
| } | ||
|
|
||
| func getReadBuffer(uConn *net.UDPConn) (int, error) { | ||
| file, err := uConn.File() | ||
| if err != nil { | ||
| return 0, err | ||
| } | ||
|
|
||
| readBuffer, err := unix.GetsockoptInt(int(file.Fd()), syscall.SOL_SOCKET, syscall.SO_RCVBUF) | ||
| if err != nil { | ||
| return 0, err | ||
| } | ||
| return readBuffer, nil | ||
| } | ||
|
|
||
| func NewBufferWatcher(uConn *net.UDPConn) (*BufferWatcher, error) { | ||
| readBuffer, err := getReadBuffer(uConn) | ||
| if err != nil { | ||
| return &BufferWatcher{}, err | ||
| } | ||
|
|
||
| return &BufferWatcher{ | ||
| ReadBuffer: readBuffer, | ||
| uAddr: uConn.LocalAddr().(*net.UDPAddr), | ||
| }, nil | ||
| } | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,29 @@ | ||
| // +build !linux | ||
|
|
||
| // Copyright 2019 The Prometheus 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 telemetry | ||
|
|
||
| import ( | ||
| "errors" | ||
| "net" | ||
| ) | ||
|
|
||
| func NewBufferWatcher(uConn *net.UDPConn) (*BufferWatcher, error) { | ||
| return &BufferWatcher{}, errors.New("UDP Buffer watching unsupported on this OS") | ||
| } | ||
|
|
||
| func (b *BufferWatcher) GetSocketQueue() (int, error) { | ||
| return 0, errors.New("UDP Buffer watching unsupported on this OS") | ||
| } |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
would this always happen on non-Linux platforms?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Would this always log? Yes, it would. I figured a debug log would be a good place to quietly drop that to minimize confusion over why they can't get the buffer tracking to work, but if you think it's unnecessary, I would be happy to more quietly short circuit that.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
it's fine as it is, just trying to understand! Maybe add a comment to say how we expect the initialization to behave on non-Linux systems?