运维

概述

既然开始使用VictoriaMetrics了那么自然也要了解一下指标怎么写入和读取,比如目前我想采集一个新的指标,应该怎么写入.

本文将使用 python 脚本做个简单的演示,并且详细讲解各种指标的写法.

写入方法较多,下面会挨个介绍和展示,如果文档有不足之处还请见谅,可多参考参考官网的文档.

指标数组写入

为什么需要指标数组这种东西?

实际上这种指标写法叫Batch Write（批量写入）,即一次性写入多个指标的行为.

假设场景: 单台服务器有 4 张网卡,我们需要采集每张网卡的流量数据上报到VictoriaMetrics,且均采用同一指名.

那么对于这种数据我们肯定不能一个个请求写入,当写入规模很大时效率太慢不说还占用服务端资源.

所以这个时候就可以用到指标数组批量写入数据,一起写入提升效率,具体操作思路如下

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
import requests
import time
import random
from requests.auth import HTTPBasicAuth

# VictoriaMetrics 配置
VM_URL = "http://localhost:8428"  # 替换为你的 VictoriaMetrics 地址
WRITE_ENDPOINT = f"{VM_URL}/api/v1/import"
USERNAME = "your_username"  # 替换为你的用户名
PASSWORD = "your_password"  # 替换为你的密码

# 生成指标数组
def generate_metrics_array(num_metrics=5):
    metrics = []
    base_timestamp = int(time.time())

    for i in range(num_metrics):
        metric = {
            "metric_name": "custom_cpu_usage",
            "labels": {
                "env": "prod",
                "instance": f"server{i+1}",
                "job": "test_job"
            },
            "value": random.uniform(0, 100),  # 模拟 CPU 使用率
            "timestamp": base_timestamp + i  # 递增时间戳
        }
        metrics.append(metric)

    return metrics

# 将指标数组转换为 Prometheus 文本格式
def format_metrics_to_prometheus(metrics):
    metric_lines = []
    for metric in metrics:
        metric_name = metric["metric_name"]
        labels = metric["labels"]
        value = metric["value"]
        timestamp = metric["timestamp"]

        # 构造标签字符串,例如：env="prod",instance="server1",job="test_job"
        labels_str = ",".join([f'{key}="{value}"' for key, value in labels.items()])
        # 构造完整指标行,例如：custom_cpu_usage{env="prod",instance="server1",job="test_job"} 42.5 1697054700
        metric_line = f'{metric_name}{{{labels_str}}} {value} {timestamp}'
        metric_lines.append(metric_line)

    # 用换行符连接所有指标行
    return "\n".join(metric_lines)

# 写入指标到 VictoriaMetrics
def write_to_victoriametrics(metric_data):
    headers = {
        "Content-Type": "text/plain"
    }
    try:
        response = requests.post(
            WRITE_ENDPOINT,
            data=metric_data,
            headers=headers,
            auth=HTTPBasicAuth(USERNAME, PASSWORD)
        )
        if response.status_code == 204:
            print("成功写入指标数组到 VictoriaMetrics")
        else:
            print(f"写入失败,状态码: {response.status_code}, 响应: {response.text}")
    except Exception as e:
        print(f"写入时发生错误: {str(e)}")

# 主程序
if __name__ == "__main__":
    # 生成指标数组
    metrics_array = generate_metrics_array(num_metrics=5)
    print("生成的指标数组:")
    for metric in metrics_array:
        print(metric)

    # 转换为 Prometheus 格式
    metric_data = format_metrics_to_prometheus(metrics_array)
    print(f"\n格式化的 Prometheus 数据:\n{metric_data}")

    # 写入 VictoriaMetrics
    write_to_victoriametrics(metric_data)

生成的指标数组:

1
2
3
4
5
{'metric_name': 'custom_cpu_usage', 'labels': {'env': 'prod', 'instance': 'server1', 'job': 'test_job'}, 'value': 67.123456, 'timestamp': 1697054700}
{'metric_name': 'custom_cpu_usage', 'labels': {'env': 'prod', 'instance': 'server2', 'job': 'test_job'}, 'value': 23.456789, 'timestamp': 1697054701}
{'metric_name': 'custom_cpu_usage', 'labels': {'env': 'prod', 'instance': 'server3', 'job': 'test_job'}, 'value': 89.012345, 'timestamp': 1697054702}
{'metric_name': 'custom_cpu_usage', 'labels': {'env': 'prod', 'instance': 'server4', 'job': 'test_job'}, 'value': 45.678901, 'timestamp': 1697054703}
{'metric_name': 'custom_cpu_usage', 'labels': {'env': 'prod', 'instance': 'server5', 'job': 'test_job'}, 'value': 12.345678, 'timestamp': 1697054704}

格式化的 Prometheus 数据:

1
2
3
4
5
custom_cpu_usage{env="prod",instance="server1",job="test_job"} 67.123456 1697054700
custom_cpu_usage{env="prod",instance="server2",job="test_job"} 23.456789 1697054701
custom_cpu_usage{env="prod",instance="server3",job="test_job"} 89.012345 1697054702
custom_cpu_usage{env="prod",instance="server4",job="test_job"} 45.678901 1697054703
custom_cpu_usage{env="prod",instance="server5",job="test_job"} 12.345678 1697054704

单个指标写入

当然有时候你也想一次只写入一个指标,那么可以参考下面这份代码.

至于他们的区别,在看完上面指标数组的介绍相信你应该也有所了解了,这里就不在过多解释了,直接上代码.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
import requests
import time
import random
from requests.auth import HTTPBasicAuth

# VictoriaMetrics 配置
VM_URL = "http://localhost:8428"  # 替换为你的 VictoriaMetrics 地址
WRITE_ENDPOINT = f"{VM_URL}/api/v1/import"
USERNAME = "your_username"  # 替换为你的用户名
PASSWORD = "your_password"  # 替换为你的密码

# 生成模拟指标数据
def generate_metric():
    metric_name = "custom_cpu_usage"
    labels = {
        "env": "prod",
        "instance": "server1",
        "job": "test_job"
    }
    value = random.uniform(0, 100)  # 模拟 CPU 使用率（0-100%）
    timestamp = int(time.time())  # 当前时间戳（秒）

    # 构造 Prometheus 文本格式
    labels_str = ",".join([f'{key}="{value}"' for key, value in labels.items()])
    metric_line = f'{metric_name}{{{labels_str}}} {value} {timestamp}'
    return metric_line

# 写入指标到 VictoriaMetrics
def write_to_victoriametrics(metric_data):
    headers = {
        "Content-Type": "text/plain"
    }
    try:
        response = requests.post(
            WRITE_ENDPOINT,
            data=metric_data,
            headers=headers,
            auth=HTTPBasicAuth(USERNAME, PASSWORD)  # 添加 Basic Auth
        )
        if response.status_code == 204:
            print("成功写入指标到 VictoriaMetrics")
        else:
            print(f"写入失败,状态码: {response.status_code}, 响应: {response.text}")
    except Exception as e:
        print(f"写入时发生错误: {str(e)}")

# 主程序
if __name__ == "__main__":
    # 模拟每 10 秒写入一次指标
    for _ in range(5):  # 写入 5 次作为示例
        metric_data = generate_metric()
        print(f"生成的指标数据: {metric_data}")
        write_to_victoriametrics(metric_data)
        time.sleep(10)  # 每 10 秒生成一次指标

指标追加全局标签

有时候写入指标时希望附带一些附加的全局标签,这些标签往往是全局固定不变的,但是每个指标又得带上它.

举个简单的例子,比如我目前有服务器的流量,磁盘,CPU 使用率指标,那么我怎么知道这个指标是哪台服务器?

所以这个时候是不是应该给他加一个ipv4标签去记录他的 IP 地址?或者uuid标签记录他的设备 SN 唯一标识?

但是这个标签你怎么加入呢?可参考下面代码的例子,在 url 的后面我们追加了一个extra_label参数,里面写了我们的标签名和值.

就是通过这种方式直接添加的,而无需再写入指标里面设置.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
import requests
import time
import random
from requests.auth import HTTPBasicAuth

# VictoriaMetrics 配置
VM_URL = "http://localhost:8428"
WRITE_ENDPOINT = f"{VM_URL}/api/v1/import?extra_label=source=python_script&extra_label=tenant=team_a&extra_label=region=us-east"
USERNAME = "your_username"
PASSWORD = "your_password"

# 生成指标
def generate_metrics(num_metrics=3):
    metrics = []
    base_timestamp = int(time.time())
    for i in range(num_metrics):
        metric = {
            "name": "custom_cpu_usage",
            "labels": {
                "env": "prod",
                "instance": f"server{i + 1}",
                "job": "test_job"
            },
            "value": random.uniform(0, 100),
            "timestamp": base_timestamp + i
        }
        metrics.append(metric)
    return metrics

# 格式化为 Prometheus 文本
def format_metrics(metrics):
    lines = []
    for metric in metrics:
        name = metric["name"]
        labels = metric["labels"]
        value = metric["value"]
        timestamp = metric["timestamp"]
        labels_str = ",".join([f'{key}="{value}"' for key, value in labels.items()])
        lines.append(f'{name}{{{labels_str}}} {value} {timestamp}')
    return "\n".join(lines)

# 写入 VictoriaMetrics
def write_metrics(metric_data):
    headers = {"Content-Type": "text/plain"}
    try:
        response = requests.post(
            WRITE_ENDPOINT,
            data=metric_data,
            headers=headers,
            auth=HTTPBasicAuth(USERNAME, PASSWORD)
        )
        if response.status_code == 204:
            print("写入成功,所有指标已添加 extra_label")
        else:
            print(f"写入失败,状态码: {response.status_code}, 响应: {response.text}")
    except Exception as e:
        print(f"写入错误: {str(e)}")

# 主程序
if __name__ == "__main__":
    metrics = generate_metrics(num_metrics=3)
    metric_data = format_metrics(metrics)
    print(f"Prometheus 格式数据:\n{metric_data}")
    write_metrics(metric_data)

其实还有一个办法,参考代码如下:

定义一个函数根据环境变量或者脚本变量动态为标签赋值然后 return 给指标合并,最后上报.

1
2
3
4
5
6
7
8
9
10
11
12
import os

# 动态生成全局标签
def get_global_labels():
    return {
        "source": "python_script",
        "tenant": os.getenv("TENANT_ID", "team_a"),
        "region": os.getenv("REGION", "us-east"),
        "app_version": "1.0.0"  # 动态添加版本
    }

GLOBAL_LABELS = get_global_labels()

二进制高性能写入

VictoriaMetrics 支持 Prometheus 的 remote_write 协议,允许以二进制格式（Protobuf）写入数据,相比文本格式更高效.

Python 客户端可以使用 prometheus_client 或自定义 Protobuf 序列化.

场景一般就是高性能写入场景,减少带宽和解析开销,反正按需使用吧.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
from prometheus_client import CollectorRegistry, Gauge, remote_write
import time
import random
from requests.auth import HTTPBasicAuth

# VictoriaMetrics 配置
VM_URL = "http://localhost:8428"
REMOTE_WRITE_ENDPOINT = f"{VM_URL}/api/v1/write"
USERNAME = "your_username"
PASSWORD = "your_password"

# 创建 Prometheus 注册表和指标
registry = CollectorRegistry()
cpu_usage = Gauge(
    "custom_cpu_usage",
    "CPU usage percentage",
    ["env", "instance", "job"],
    registry=registry
)

# 生成并写入指标
def write_metrics(num_metrics=3):
    for i in range(num_metrics):
        cpu_usage.labels(
            env="prod",
            instance=f"server{i + 1}",
            job="test_job"
        ).set(random.uniform(0, 100))
    
    try:
        remote_write(
            REMOTE_WRITE_ENDPOINT,
            registry,
            headers={"Content-Type": "application/x-protobuf"},
            auth=(USERNAME, PASSWORD)
        )
        print("Remote write 写入成功")
    except Exception as e:
        print(f"写入错误: {str(e)}")

# 主程序
if __name__ == "__main__":
    write_metrics(num_metrics=3)