🖋️ 开发散文：使用 Python 以 UDP 协议发送 Prometheus 指标数据

开发散文：以开发为主题的散文。基本完整复现了从遇到问题、分析问题、找到解决思路、编码实现思路的全流程，顺带着记录一些浅薄的知识点。

error: [Errno 90] Message too long

UDP 包是多大？

• Source Port：包来源端口信息。

• Destination Port：包目的端口信息。

• UDP Length：UDP 头信息+ Payload 的总长度。

• UDP Checksum：包的校验和，避免在数据在传输中被污染。

# IPv4
0xffff - (sizeof(IP Header) + sizeof(UDP Header)) = 65535-(20+8) = 65507

# IPv6
0xffff - sizeof(UDP Header) = 65535-8 = 65527

如何在 Python 计算数据大小？

getsizeof 的疑惑

from sys import getsizeof

getsizeof(b"")        👉 33
getsizeof(b"abcd")    👉 37

Empty
Bytes  type        scaling notes
28     int         +4 bytes about every 30 powers of 2
37     bytes       +1 byte per additional byte
49     str         +1-4 per additional character (depending on max width)
48     tuple       +8 per additional item
64     list        +8 for each additional
224    set         5th increases to 736; 21nd, 2272; 85th, 8416; 341, 32992
240    dict        6th increases to 368; 22nd, 1184; 43rd, 2280; 86th, 4704; 171st, 9320
136    func def    does not include default args and other attrs
1056   class def   no slots 
56     class inst  has a __dict__ attr, same scaling as dict above
888    class def   with slots
16     __slots__   seems to store in mutable tuple-like structure
                   first slot grows to 48, and so on.

简单点， len()

big_bytes: bytes
size = len(big_bytes)

如何对 Prometheus 指标数据切片？

# HELP sample_metric_foo_count 示例解释
# TYPE sample_metric_foo_count gauge
sample_metric_foo_count{foo="aa",bar="bb"} 14.0
sample_metric_foo_count{foo="cc",bar="dd"} 27.0
sample_metric_foo_count{foo="ee",bar="ff"} 12.0

# HELP sample_metric_bar_count 示例解释
# TYPE sample_metric_bar_count gauge
sample_metric_bar_count{baz="gg"} 660.0
sample_metric_bar_count{baz="hh"} 660.0

# HELP sample_metric_foo_count 示例解释
# TYPE sample_metric_foo_count gauge

// 以下为摘抄内容
func (p *TextParser) startOfLine() stateFn {
  ...
	switch p.currentByte {
	case '#':
		return p.startComment
	case '\n':
		return p.startOfLine // Empty line, start the next one.
	}
	return p.readingMetricName
}

func (p *TextParser) startComment() stateFn {
  ...
	keyword := p.currentToken.String()
	if keyword != "HELP" && keyword != "TYPE" {
		// Generic comment, ignore by fast forwarding to end of line.
		for p.currentByte != '\n' {
			if p.currentByte, p.err = p.buf.ReadByte(); p.err != nil {
				return nil // Unexpected end of input.
			}
		}
		return p.startOfLine
	}
	...
	switch keyword {
	case "HELP":
		return p.readingHelp
	case "TYPE":
		return p.readingType
	}
	panic(fmt.Sprintf("code error: unexpected keyword %q", keyword))
}

------ # HELP 作为开头，尽量作为数据包的开头 ------
# HELP sample_metric_foo_count 示例解释
# TYPE sample_metric_foo_count gauge
sample_metric_foo_count{foo="aa",bar="bb"} 14.0
sample_metric_foo_count{foo="cc",bar="dd"} 27.0
sample_metric_foo_count{foo="ee",bar="ff"} 12.0

# HELP sample_metric_bar_count 示例解释
# TYPE sample_metric_bar_count gauge
sample_metric_bar_count{baz="gg"} 660.0\n
------ \n 作为单行数据结尾，能够保证包数据被正常解析，最差情况作为数据包的结尾 ------
sample_metric_bar_count{baz="hh"} 660.0  由于缺少 HELP/TYPE 元信息，将无法被聚合处理

开码！

首先定义模型

@dataclass
class SlicedIndex:
    """数据切分索引"""
    start: int
    end: int
    # 如果不是以 # HELP 开头，会导致 metric 无法被识别合并
    valid_start: bool = True
    # 如果不是以 metric 内容完结，会让下一个分片包没有有效开头
    valid_end: bool = True

    def to_tuple(self) -> tuple:
        return self.start, self.end

    def __len__(self):
        return self.end - self.start


@dataclass
class SlicedIndexList:
    indexes: List[SlicedIndex] = field(default_factory=list)

    def append(self, start: int, end: int, valid_end: bool = True):
        valid_start = True
        if self.indexes and not self.indexes[-1].valid_end:
            valid_start = False

        self.indexes.append(SlicedIndex(start, end, valid_start, valid_end))

    def __iter__(self):
        for elem in self.indexes:
            yield elem

    def __getitem__(self, ii):
        """Get a list item"""
        return self.indexes[ii]

递归寻找切分点

def find_sliced_indexes(data: bytes, start: int, udp_package_max_size: int, sliced_index_list: SlicedIndexList):
    """递归找寻指标开头标志"""
    if start + udp_package_max_size >= len(data):
        sliced_index_list.append(start, len(data))
        return

    valid_start_index = data.rfind(b"# HELP", start, start + udp_package_max_size)
    if valid_start_index == -1 or valid_start_index == start:
        # 当某个 metrics 数据大小大于 UDP 协议最大包限制时，原则上我们无法通过 UDP 发送该数据
        # 但这里我们尝试以最大限制发送该数据，以一个超过最大限制的数据为例
        # +-----------------+ 协议最大长度
        # +-----------------+  +-------------------+  +-------------------+
        # 有效            有效  (缺失元信息部分丢弃) 有效  有效              有效
        # 可以看到中间段数据会被丢弃，因为没有有效的 metrics 开头，但是我们尽可能保证了后续的包是有效开头和有效结尾
        # 所以当前的做法更利于我们单 metrics 数据比 65535 稍大的场景，这样丢弃的内容少，保留的内容多
        # 反之，当单个 metrics 的数据比最大限制大的越多，丢弃的越多

        # sample 肯定会转行，只要有转行就能保证起码前一个包的内容格式有效
        valid_end_index = data.rfind(b"\n", start, start + udp_package_max_size)
        new_start = valid_end_index + 1
        sliced_index_list.append(start, new_start, valid_end=False)
    else:
        sliced_index_list.append(start, valid_start_index)
        new_start = valid_start_index

    # 尝试继续向后查找
    find_sliced_indexes(data, new_start, udp_package_max_size, sliced_index_list)
    return

切！

def slice_metrics_udp_data(data: bytes, sliced_indexes: SlicedIndexList) -> Generator[memoryview, None, None]:
    """拆分 metrics UDP data"""
    # Q: 为什么不直接切分?
    # A: 直接按照大小切分会降 bytes 中的字面值切断，让服务端对于指标无法理解，所以要按照字面值的内容做切分

    # memoryview 无需额外拷贝
    mview = memoryview(data)
    for index in sliced_indexes:
        if not index.valid_start:
            logger.warning("data<len:%s> has no valid start, may not be parsed", len(index))
        yield mview[index.start : index.end]

收尾

def find_udp_data_sliced_indexes(data: bytes, udp_package_max_size: int = 65507, mtu: int = 1500) -> SlicedIndexList:
    """对 UDP 发送数据进行切片处理，保证每次发送成功
    :param data: 预发送数据
    :param udp_package_max_size: 当前系统支持的最大 UDP 发送包大小，以 bytes 计算，默认为 65535 (在 macOS 下默认为 9126)
    :param mtu: Maximum Transmission Unit

    udp_package_max_size = 0xffff - (sizeof(IP Header) + sizeof(UDP Header)) = 65535-(20+8) = 65507
    ref to: https://en.wikipedia.org/wiki/User_Datagram_Protocol
    """
    length = len(data)
    if length > mtu:
        # TODO: 当前我们暂不考虑处理 MTU 的问题，先解决 UDP 包过大的情况
        logger.debug("UDP packages is larger than MTU, not safe for single push.")

    if length <= udp_package_max_size:
        return SlicedIndexList(indexes=[SlicedIndex(0, length)])

    sliced_index_list = SlicedIndexList()
    try:
        find_sliced_indexes(data, 0, udp_package_max_size, sliced_index_list)
    except RecursionError:
        logger.warning("data has no valid format, drop it...")

    return sliced_index_list

效果

# 原来的 UDP 直接发送
# 将抛出 error: [Errno 90] Message too long
udp_socket.sendto(data, (address, port))

# 循环调用 UDP 请求
for sliced_data in slice_metrics_udp_data(data, find_udp_data_sliced_indexes(data)):
	  # 所有 UDP 包都可以发送出去
    # 同时取决于数据特性，绝大多数的 UDP 数据都能在服务端得到解析
    udp_socket.sendto(sliced_data, (address, port))

尾声

进一步优化？

参考

• https://www.ietf.org/rfc/rfc768.txt

• https://erg.abdn.ac.uk/users/gorry/course/inet-pages/udp.html

• https://www.rfc-editor.org/rfc/rfc2675.html

• https://stackoverflow.com/questions/1098897/what-is-the-largest-safe-udp-packet-size-on-the-internet

• https://stackoverflow.com/questions/449560/how-do-i-determine-the-size-of-an-object-in-python

• https://docs.python.org/zh-cn/3/c-api/memoryview.html