udp

README

The UDP Input

A note on UDP/IP OS Buffer sizes

Some OSes (most notably, Linux) place very restricive limits on the performance of UDP protocols. It is highly recommended that you increase these OS limits to at least 25MB before trying to run UDP traffic to your instance. 25MB is just a recommendation, and should be adjusted to be inline with your read-buffer plugin setting.

Linux

Check the current UDP/IP receive buffer default and limit by typing the following commands:

sysctl net.core.rmem_max
sysctl net.core.rmem_default

If the values are less than 26214400 bytes (25MB) you should add the following lines to the /etc/sysctl.conf file:

net.core.rmem_max=26214400
net.core.rmem_default=26214400

Changes to /etc/sysctl.conf do not take effect until reboot. To update the values immediately, type the following commands as root:

sysctl -w net.core.rmem_max=26214400
sysctl -w net.core.rmem_default=26214400

BSD/Darwin

On BSD/Darwin systems you need to add about a 15% padding to the kernel limit socket buffer. Meaning if you want an 25MB buffer (8388608 bytes) you need to set the kernel limit to 26214400*1.15 = 30146560. This is not documented anywhere but happens in the kernel here.

Check the current UDP/IP buffer limit by typing the following command:

sysctl kern.ipc.maxsockbuf

If the value is less than 30146560 bytes you should add the following lines to the /etc/sysctl.conf file (create it if necessary):

kern.ipc.maxsockbuf=30146560

Changes to /etc/sysctl.conf do not take effect until reboot. To update the values immediately, type the following commands as root:

sysctl -w kern.ipc.maxsockbuf=30146560

Using the read-buffer option for the UDP listener

The read-buffer option allows users to set the buffer size for the UDP listener. It Sets the size of the operating system's receive buffer associated with the UDP traffic. Keep in mind that the OS must be able to handle the number set here or the UDP listener will error and exit.

read-buffer = 0 means to use the OS default, which is usually too small for high UDP performance.

Configuration

Each UDP input allows the binding address, target database, and target retention policy to be set. If the database does not exist, it will be created automatically when the input is initialized. If the retention policy is not configured, then the default retention policy for the database is used. However if the retention policy is set, the retention policy must be explicitly created. The input will not automatically create it.

Each UDP input also performs internal batching of the points it receives, as batched writes to the database are more efficient. The default batch size is 1000, pending batch factor is 5, with a batch timeout of 1 second. This means the input will write batches of maximum size 1000, but if a batch has not reached 1000 points within 1 second of the first point being added to a batch, it will emit that batch regardless of size. The pending batch factor controls how many batches can be in memory at once, allowing the input to transmit a batch, while still building other batches.

Processing

The UDP input can receive up to 64KB per read, and splits the received data by newline. Each part is then interpreted as line-protocol encoded points, and parsed accordingly.

UDP is connectionless

Since UDP is a connectionless protocol there is no way to signal to the data source if any error occurs, and if data has even been successfully indexed. This should be kept in mind when deciding if and when to use the UDP input. The built-in UDP statistics are useful for monitoring the UDP inputs.

Config Examples

One UDP listener

# influxd.conf
...
[[udp]]
  enabled = true
  bind-address = ":8089" # the bind address
  database = "telegraf" # Name of the database that will be written to
  batch-size = 5000 # will flush if this many points get buffered
  batch-timeout = "1s" # will flush at least this often even if the batch-size is not reached
  batch-pending = 10 # number of batches that may be pending in memory
  read-buffer = 0 # UDP read buffer, 0 means to use OS default
...

Multiple UDP listeners

# influxd.conf
...
[[udp]]
  # Default UDP for Telegraf
  enabled = true
  bind-address = ":8089" # the bind address
  database = "telegraf" # Name of the database that will be written to
  batch-size = 5000 # will flush if this many points get buffered
  batch-timeout = "1s" # will flush at least this often even if the batch-size is not reached
  batch-pending = 10 # number of batches that may be pending in memory
  read-buffer = 0 # UDP read buffer size, 0 means to use OS default

[[udp]]
  # High-traffic UDP
  enabled = true
  bind-address = ":80891" # the bind address
  database = "mymetrics" # Name of the database that will be written to
  batch-size = 5000 # will flush if this many points get buffered
  batch-timeout = "1s" # will flush at least this often even if the batch-size is not reached
  batch-pending = 100 # number of batches that may be pending in memory
  read-buffer = 8388608 # (8*1024*1024) UDP read buffer size
...

Documentation

Overview

Package udp provides the UDP input service for InfluxDB.

Index

• Constants
• type Config
  • func NewConfig() Config
  • func (c *Config) WithDefaults() *Config
• type Service
  • func NewService(c Config) *Service
  • func (s *Service) Addr() net.Addr
  • func (s *Service) Close() error
  • func (s *Service) Closed() bool
  • func (s *Service) Open() (err error)
  • func (s *Service) Statistics(tags map[string]string) []models.Statistic
  • func (s *Service) WithLogger(log zap.Logger)
• type Statistics

Constants

const (
	// DefaultBindAddress is the default binding interface if none is specified.
	DefaultBindAddress = ":8089"

	// DefaultDatabase is the default database for UDP traffic.
	DefaultDatabase = "udp"

	// DefaultRetentionPolicy is the default retention policy used for writes.
	DefaultRetentionPolicy = ""

	// DefaultBatchSize is the default UDP batch size.
	DefaultBatchSize = 5000

	// DefaultBatchPending is the default number of pending UDP batches.
	DefaultBatchPending = 10

	// DefaultBatchTimeout is the default UDP batch timeout.
	DefaultBatchTimeout = time.Second

	// DefaultPrecision is the default time precision used for UDP services.
	DefaultPrecision = "n"

	// DefaultReadBuffer is the default buffer size for the UDP listener.
	// Sets the size of the operating system's receive buffer associated with
	// the UDP traffic. Keep in mind that the OS must be able
	// to handle the number set here or the UDP listener will error and exit.
	//
	// DefaultReadBuffer = 0 means to use the OS default, which is usually too
	// small for high UDP performance.
	//
	// Increasing OS buffer limits:
	//     Linux:      sudo sysctl -w net.core.rmem_max=<read-buffer>
	//     BSD/Darwin: sudo sysctl -w kern.ipc.maxsockbuf=<read-buffer>
	DefaultReadBuffer = 0
)

const (

	// MAX_UDP_PAYLOAD is largest payload size the UDP service will accept.
	MAX_UDP_PAYLOAD = 64 * 1024
)

Types

type Config

type Config struct {
	Enabled     bool   `toml:"enabled"`
	BindAddress string `toml:"bind-address"`

	Database        string        `toml:"database"`
	RetentionPolicy string        `toml:"retention-policy"`
	BatchSize       int           `toml:"batch-size"`
	BatchPending    int           `toml:"batch-pending"`
	ReadBuffer      int           `toml:"read-buffer"`
	BatchTimeout    toml.Duration `toml:"batch-timeout"`
	Precision       string        `toml:"precision"`
	// contains filtered or unexported fields
}

Config holds various configuration settings for the UDP listener.

func NewConfig

func NewConfig() Config

NewConfig returns a new instance of Config with defaults.

func (*Config) WithDefaults

func (c *Config) WithDefaults() *Config

WithDefaults takes the given config and returns a new config with any required default values set.

type Service

type Service struct {
	PointsWriter interface {
		WritePoints(database, retentionPolicy string, consistencyLevel models.ConsistencyLevel, points []models.Point) error
	}

	MetaClient interface {
		CreateDatabase(name string) (*meta.DatabaseInfo, error)
	}

	Logger zap.Logger
	// contains filtered or unexported fields
}

Service is a UDP service that will listen for incoming packets of line protocol.

func NewService

func NewService(c Config) *Service

NewService returns a new instance of Service.

func (*Service) Addr

func (s *Service) Addr() net.Addr

Addr returns the listener's address.

func (*Service) Close

func (s *Service) Close() error

Close closes the service and the underlying listener.

func (*Service) Closed

func (s *Service) Closed() bool

Closed returns true if the service is currently closed.

func (*Service) Open

func (s *Service) Open() (err error)

Open starts the service.

func (*Service) Statistics

func (s *Service) Statistics(tags map[string]string) []models.Statistic

Statistics returns statistics for periodic monitoring.

func (*Service) WithLogger

func (s *Service) WithLogger(log zap.Logger)

WithLogger sets the logger on the service.

type Statistics

type Statistics struct {
	PointsReceived      int64
	BytesReceived       int64
	PointsParseFail     int64
	ReadFail            int64
	BatchesTransmitted  int64
	PointsTransmitted   int64
	BatchesTransmitFail int64
}

Statistics maintains statistics for the UDP service.