如何在OpenTelemetry Python中动态更新ObservableGauge

问题描述 #

我正在使用opentelemetry-api 1.14和opentelemetry-sdk 1.14，已掌握Counter与ObservableGauge的创建及使用方法。但我需要像Counter调用add方法那样，在应用中动态更新并设置gauge的值。当前代码中的gauge值固定为9，请问如何在main函数中更新simple_gauge？

原代码如下：

import time
""" API is the interface that you should interact with."""
from opentelemetry import metrics
"""
SDK is the implementation. Only access SDK during initialization, startup, and shutdown.
"""
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader, ConsoleMetricExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.resources import Resource

def initialize():
    resource = Resource(attributes={"service.name": "otel-test"})
    readers = []

    # Console Exporter
    exporter = ConsoleMetricExporter() 
    reader1 = PeriodicExportingMetricReader(exporter, export_interval_millis=5000) 
    readers.append(reader1)

    provider = MeterProvider(metric_readers=readers, resource=resource) 
    metrics.set_meter_provider(provider) 

initialize()

provider = metrics.get_meter_provider()
meter =  provider.get_meter("my-demo-meter")

simple_counter = meter.create_counter("simple_counter", description="simply increments each loop")
# Async Gauge
def observable_gauge_func(options):
    yield metrics.Observation(9, {})
simple_gauge = meter.create_observable_gauge("simple_gauge", [observable_gauge_func])

# How can I update simple_gauge in main
def main():
    loop_counter = 0
    while True:
        print(loop_counter)
        loop_counter += 1
        simple_counter.add(1)
        # How can I update simple_gauge here?
        time.sleep(5)

main()

解决方案 #

ObservableGauge是异步类型的指标，它的值由注册的观测函数在每次导出时动态返回，无法像Counter那样直接调用方法设置值。核心思路是让观测函数能访问到你在main中更新的状态变量，从而返回最新值。

方法1：使用全局变量（简单直接） #

定义一个全局变量存储gauge的当前值，观测函数读取该变量并返回，在main中修改全局变量即可更新值：

import time
from opentelemetry import metrics
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader, ConsoleMetricExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.resources import Resource

def initialize():
    resource = Resource(attributes={"service.name": "otel-test"})
    readers = []
    exporter = ConsoleMetricExporter() 
    reader1 = PeriodicExportingMetricReader(exporter, export_interval_millis=5000) 
    readers.append(reader1)
    provider = MeterProvider(metric_readers=readers, resource=resource) 
    metrics.set_meter_provider(provider) 

initialize()

provider = metrics.get_meter_provider()
meter = provider.get_meter("my-demo-meter")

simple_counter = meter.create_counter("simple_counter", description="simply increments each loop")

# 全局变量存储gauge当前值
gauge_value = 9

def observable_gauge_func(options):
    # 返回当前的gauge值
    yield metrics.Observation(gauge_value, {})

simple_gauge = meter.create_observable_gauge("simple_gauge", [observable_gauge_func])

def main():
    global gauge_value  # 声明使用全局变量
    loop_counter = 0
    while True:
        print(loop_counter)
        loop_counter += 1
        simple_counter.add(1)
        # 动态更新gauge的值
        gauge_value = loop_counter * 2  # 可替换为任意业务逻辑计算的值
        time.sleep(5)

main()

方法2：使用可变对象（避免全局变量） #

用列表这类可变对象存储状态，无需声明全局变量，观测函数直接读取对象内容：

import time
from opentelemetry import metrics
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader, ConsoleMetricExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.resources import Resource

def initialize():
    resource = Resource(attributes={"service.name": "otel-test"})
    readers = []
    exporter = ConsoleMetricExporter() 
    reader1 = PeriodicExportingMetricReader(exporter, export_interval_millis=5000) 
    readers.append(reader1)
    provider = MeterProvider(metric_readers=readers, resource=resource) 
    metrics.set_meter_provider(provider) 

initialize()

provider = metrics.get_meter_provider()
meter = provider.get_meter("my-demo-meter")

simple_counter = meter.create_counter("simple_counter", description="simply increments each loop")

# 用列表存储gauge状态（可变对象）
gauge_state = [9]

def observable_gauge_func(options):
    yield metrics.Observation(gauge_state[0], {})

simple_gauge = meter.create_observable_gauge("simple_gauge", [observable_gauge_func])

def main():
    loop_counter = 0
    while True:
        print(loop_counter)
        loop_counter += 1
        simple_counter.add(1)
        # 更新列表中的值
        gauge_state[0] = loop_counter * 2
        time.sleep(5)

main()

方法3：闭包封装（更优雅的状态管理） #

通过闭包封装状态和更新逻辑，避免全局变量，代码结构更清晰：

import time
from opentelemetry import metrics
from opentelemetry.sdk.metrics.export import PeriodicExportingMetricReader, ConsoleMetricExporter
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.resources import Resource

def initialize():
    resource = Resource(attributes={"service.name": "otel-test"})
    readers = []
    exporter = ConsoleMetricExporter() 
    reader1 = PeriodicExportingMetricReader(exporter, export_interval_millis=5000) 
    readers.append(reader1)
    provider = MeterProvider(metric_readers=readers, resource=resource) 
    metrics.set_meter_provider(provider) 

initialize()

provider = metrics.get_meter_provider()
meter = provider.get_meter("my-demo-meter")

simple_counter = meter.create_counter("simple_counter", description="simply increments each loop")

# 闭包封装gauge状态与观测函数
def create_dynamic_gauge(initial_value):
    current_value = initial_value
    def gauge_observable(options):
        yield metrics.Observation(current_value, {})
    def update_value(new_value):
        nonlocal current_value
        current_value = new_value
    return gauge_observable, update_value

# 创建观测函数和更新方法
gauge_observable, update_gauge = create_dynamic_gauge(9)
simple_gauge = meter.create_observable_gauge("simple_gauge", [gauge_observable])

def main():
    loop_counter = 0
    while True:
        print(loop_counter)
        loop_counter += 1
        simple_counter.add(1)
        # 调用更新方法设置gauge值
        update_gauge(loop_counter * 2)
        time.sleep(5)

main()

原理说明 #

ObservableGauge的设计是让SDK定期调用观测函数获取当前指标值，所以只要观测函数能访问到最新的状态，就能实现动态更新。以上三种方法本质都是将main中更新的状态传递给观测函数，区别在于状态的存储方式。

内容的提问来源于stack exchange，提问作者Matthew Thomas