Grafana 仪表盘变量让同一组面板能够切换环境、服务或实例，而不必复制多份仪表盘。查询变量从数据源获得选项，自定义变量则可以维护一份有意固定的列表。这两种方式没有绝对优劣：当选项应跟随数据源发现结果变化时，可以使用查询变量；当选择范围就是一份较小且稳定的约定时，维护静态列表同样合理。变量是否合适，取决于它如何表达实际使用需求。

数据源执行查询之前，变量会先被代入查询表达式，所以变量格式本身属于查询正确性的一部分。Prometheus 多选场景应使用正则标签匹配，例如 environment=~"$environment"。Grafana 的 Prometheus 插值逻辑会把选中值处理为正则候选项，并执行所需转义。如果改用精确等号匹配，组合后的文本只会被当成一个标签值，并不会自动变成希望表达的多个环境选择。变量编辑器里的 label_values 是发现选项的辅助语法，不是可以直接复制到面板表达式中使用的 PromQL 函数。

检查最终送出的表达式

这里存在两层语法：正则表达式，以及包裹它的 PromQL 字符串。production.eu 中的点号应表示字面字符，不能成为同时匹配 productionXeu 的通配符。应让数据源按其约定完成插值；使用显式格式覆盖时，要检查真正发送出去的查询。raw 格式会绕过数据源格式化，并不是解决复杂变量值的通用办法。表面上省掉反斜线，可能只是把错误隐藏在返回的数字中。

全部选项也有明确语义。本例没有设置自定义的全部值，因此会展开经过筛选的可用选项。菜单只包含 production.eu 和 staging，全部选择就应覆盖这两个环境。如果把全部值改成点星号等自定义表达式，范围则可能扩大到后台正则允许匹配的其他标签。需要先决定这里的全部究竟指当前菜单中的值，还是更广的查询范围，再用刻意排除的数据验证这种决定。

坏掉的样子

查询变量不会保证永远新鲜，刷新设置决定了 Grafana 何时重新执行选项查询。选项依赖时间范围时，通常还需要随时间范围变化刷新。存在链式变量，也不保证每组上游选择都能找到下游组合；数据本来不存在时，结果完全可能为空。应当区分真实的空集合、已经陈旧的选择和查询错误，不能看到无数据提示，就直接把它解释为服务故障。

变量筛选不是权限控制。菜单里隐藏一个环境，并不会从数据源删除它，也不会阻止其他获准查询请求这些数据。访问边界需要在对应的 Grafana 配置与后台数据服务中落实。分享链接携带的同样只是选中状态，不是权限声明。多个环境通过重复的 var-environment 参数传递，不能随意发明一个逗号拼接的权限列表，再期望后台自动按访问规则理解它。

Grafana 过滤选项、正则及 PromQL 字符串层、合成预期总和、授权边界，以及尚未验证的下拉框行为。
Grafana 查询格式化：机制与证据边界。

快速验证

保存代码为 grafana-variables.py，准备 Python、本机 Docker daemon 和预先加载的两个镜像后，运行 python3 grafana-variables.py --serve。脚本在专用桥接网络中创建可删除的 Grafana 12.1.0 与 Prometheus 3.5.1 容器，仅通过本机回环地址暴露 Grafana，不挂载宿主机目录。打开输出的仪表盘地址。服务模式默认保留 120 秒，--serve-seconds 接受 1–300；控制键加 C 会进入清理，并保留中断，不把它当作正常结束。Docker 调用各有 30 秒超时，就绪重试采用 20 秒截止时间，最后一个正在执行的请求仍可能耗尽自身超时。清理按确切自有名称检查，删除失败、清单无法确认或命令超时或调用进行中被中断后的状态不确定都会报错。不带服务参数时，只检查手写后台查询。Python 已禁用代理发现及重定向；浏览器代理策略仍须单独控制。

import argparse
import json
import secrets
import subprocess
import time
import uuid
from http.client import HTTPException
from threading import Event
from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import HTTPRedirectHandler, ProxyHandler, build_opener

parser = argparse.ArgumentParser()
parser.add_argument("--serve", action="store_true")
parser.add_argument("--serve-seconds", type=int, default=120)
args = parser.parse_args()
if not 1 <= args.serve_seconds <= 300:
    parser.error("--serve-seconds must be between 1 and 300")

def require(condition, message):
    if not condition:
        raise AssertionError(message)

uncertain_commands = []

prefix = "xbcatm-variable-" + uuid.uuid4().hex[:10]
network, prom, grafana = [prefix + suffix for suffix in ("-net", "-prom", "-grafana")]
print(json.dumps({"owned_containers": [grafana, prom], "owned_network": network}), flush=True)


def command(*args):
    try:
        result = subprocess.run(["docker", *args], capture_output=True, text=True, timeout=30)
    except subprocess.TimeoutExpired as error:
        uncertain_commands.append(args)
        raise RuntimeError("Docker " + args[0] + " timed out") from error
    except KeyboardInterrupt:
        uncertain_commands.append(args)
        raise
    if result.returncode:
        raise RuntimeError(result.stderr.strip())
    return result.stdout.strip()


class NoRedirect(HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):
        raise RuntimeError("Unexpected redirect in local fixture")


opener = build_opener(ProxyHandler({}), NoRedirect())


def get(path):
    with opener.open(origin + path, timeout=5) as response:
        require(response.status == 200, "Unexpected HTTP status")
        return json.load(response)


def eventually(check):
    deadline = time.monotonic() + 20
    while time.monotonic() < deadline:
        try:
            value = check()
            if value:
                return value
        except (HTTPError, URLError, TimeoutError, ConnectionError, HTTPException):
            pass
        time.sleep(0.1)
    raise AssertionError("Local lab did not become ready")


def cleanup_owned(groups):
    # Inspect exact owned names; absence is not inferred from a failed inspect.
    errors = []
    for kind, names, format_string in groups:
        try:
            listing = command(kind, "ls", *( ["--all"] if kind == "container" else [] ),
                              "--format", format_string).splitlines()
            present = sorted(set(names) & set(listing))
            if present:
                command(kind, "rm", *( ["--force"] if kind == "container" else [] ), *present)
        except Exception as error:
            errors.append(f"{kind} removal: {error}")
        try:
            remaining = command(kind, "ls", *( ["--all"] if kind == "container" else [] ),
                                "--format", format_string).splitlines()
            if set(names) & set(remaining):
                errors.append(f"{kind} remains: {sorted(set(names) & set(remaining))}")
        except Exception as error:
            errors.append(f"{kind} inventory unverified: {error}")
    if uncertain_commands:
        errors.append("A Docker command timed out or was interrupted; daemon-side completion remains uncertain")
    require(not errors, "Cleanup not verified: " + "; ".join(errors))
    print(json.dumps({"cleanup": "owned names absent in final inventories"}), flush=True)


source = {"type": "prometheus", "uid": "lab_prom"}
dashboard = {
    "uid": "variable-lab", "title": "Variable escaping lab", "schemaVersion": 41,
    "version": 1, "timezone": "utc", "refresh": "", "time": {"from": "now-5m", "to": "now"},
    "templating": {"list": [{
        "name": "environment", "label": "Environment", "type": "query", "datasource": source,
        "query": {"query": "label_values(lab_active_workers, environment)", "refId": "VariableQuery"},
        "definition": "label_values(lab_active_workers, environment)",
        "regex": "/^(production[.]eu|staging)$/", "refresh": 1, "sort": 1,
        "multi": True, "includeAll": True, "allValue": "", "options": [],
        "current": {"text": ["production.eu"], "value": ["production.eu"], "selected": True},
    }]},
    "panels": [{
        "id": number, "title": title, "type": "stat", "datasource": source,
        "gridPos": {"h": 7, "w": 12, "x": (number - 1) * 12, "y": 0},
        "targets": [{"refId": "A", "expr": expression, "instant": True, "range": False}],
        "options": {"reduceOptions": {"calcs": ["lastNotNull"], "fields": "", "values": False}},
        "fieldConfig": {"defaults": {"decimals": 0}, "overrides": []},
    } for number, title, expression in (
        (1, "Selected workers", 'sum(lab_active_workers{environment=~"$environment"})'),
        (2, "Unescaped control", 'sum(lab_active_workers{environment=~"${environment:raw}"})'),
    )],
}
prom_script = """set -eu
mkdir -p /tmp/www
cat > /tmp/www/metrics.txt <<'METRICS'
# HELP lab_active_workers Synthetic active worker count.
# TYPE lab_active_workers gauge
lab_active_workers{environment="production.eu"} 10
lab_active_workers{environment="staging"} 2
lab_active_workers{environment="productionXeu"} 100
METRICS
cat > /tmp/prom.yml <<'CONFIG'
global:
  scrape_interval: 1s
scrape_configs:
  - job_name: lab
    metrics_path: /metrics.txt
    static_configs:
      - targets: ['127.0.0.1:8080']
CONFIG
/bin/httpd -f -p 127.0.0.1:8080 -h /tmp/www &
exec /bin/prometheus --config.file=/tmp/prom.yml --storage.tsdb.path=/tmp/data
"""
grafana_script = """set -eu
mkdir -p /tmp/data /tmp/provisioning/datasources /tmp/provisioning/dashboards /tmp/dashboards
mkdir -p /tmp/provisioning/plugins /tmp/provisioning/alerting
cat > /tmp/provisioning/datasources/lab.yml <<'DATASOURCE'
apiVersion: 1
datasources:
  - name: Lab Prometheus
    type: prometheus
    uid: lab_prom
    access: proxy
    url: http://metrics:9090
    isDefault: true
    jsonData:
      timeInterval: 1s
DATASOURCE
cat > /tmp/provisioning/dashboards/lab.yml <<'PROVIDER'
apiVersion: 1
providers:
  - name: lab
    type: file
    options:
      path: /tmp/dashboards
PROVIDER
cat > /tmp/dashboards/lab.json <<'DASHBOARD'
""" + json.dumps(dashboard) + """
DASHBOARD
exec /run.sh
"""
try:
    # Use preloaded images; this is a local disposable fixture, not a download step.
    for image in ("prom/prometheus:v3.5.1", "grafana/grafana:12.1.0"):
        command("image", "inspect", image)
    command("network", "create", network)
    command("run", "--pull=never", "-d", "--rm", "--name", prom, "--network", network, "--network-alias", "metrics",
            "--entrypoint", "/bin/sh", "prom/prometheus:v3.5.1", "-c", prom_script)
    command("run", "--pull=never", "-d", "--rm", "--name", grafana, "--network", network,
            "-p", "127.0.0.1::3000", "-e", "GF_PATHS_DATA=/tmp/data",
            "-e", "GF_PATHS_PROVISIONING=/tmp/provisioning",
            "-e", "GF_AUTH_ANONYMOUS_ENABLED=true", "-e", "GF_AUTH_ANONYMOUS_ORG_ROLE=Viewer",
            "-e", "GF_AUTH_DISABLE_LOGIN_FORM=true", "-e", "GF_AUTH_BASIC_ENABLED=false",
            "-e", "GF_SECURITY_ADMIN_PASSWORD=" + secrets.token_urlsafe(24),
            "-e", "GF_ANALYTICS_REPORTING_ENABLED=false", "-e", "GF_ANALYTICS_CHECK_FOR_UPDATES=false",
            "-e", "GF_ANALYTICS_CHECK_FOR_PLUGIN_UPDATES=false", "-e", "GF_PLUGINS_PREINSTALL_DISABLED=true",
            "--entrypoint", "/bin/sh", "grafana/grafana:12.1.0", "-c", grafana_script)
    origin = "http://" + command("port", grafana, "3000/tcp").splitlines()[0]
    health = eventually(lambda: get("/api/health"))
    require(health["version"] == "12.1.0" and health["database"] == "ok", "Unexpected Grafana health")
    loaded = eventually(lambda: get("/api/dashboards/uid/variable-lab"))
    require(loaded["dashboard"]["templating"]["list"][0]["multi"] is True, "Dashboard provisioning mismatch")
    checks = {}
    cases = ((r"production\\.eu", 10), ("staging", 2),
             (r"(production\\.eu|staging)", 12), ("production.eu", 110),
             ("production.eu,staging", None), (".*", 112))
    # Wait for the synthetic series to be scraped before asserting empty results.
    def ready_series():
        result = get("/api/datasources/proxy/uid/lab_prom/api/v1/query?query=" +
                     quote("count(lab_active_workers)", safe=""))["data"]["result"]
        return len(result) == 1 and float(result[0]["value"][1]) == 3
    eventually(ready_series)
    for pattern, expected in cases:
        expression = 'sum(lab_active_workers{environment=~"' + pattern + '"})'
        result = get("/api/datasources/proxy/uid/lab_prom/api/v1/query?query=" +
                     quote(expression, safe=""))["data"]["result"]
        value = None if not result else float(result[0]["value"][1])
        require(len(result) <= 1 and value == expected, (expression, result))
        checks[expression] = value
    print(json.dumps({"app_url": origin + "/d/variable-lab/variable-escaping-lab",
                      "grafana_version": health["version"],
                      "backend_handwritten_query_checks": checks,
                      "dashboard_interpolation": "NOT CHECKED",
                      "serve_seconds": args.serve_seconds if args.serve else 0}), flush=True)
    if args.serve:
        Event().wait(args.serve_seconds)
finally:
    cleanup_owned((("container", (grafana, prom), "{{.Names}}"),
                   ("network", (network,), "{{.Name}}")))

被中断或超时的命令即使随后遇到空清单，状态仍然不确定。清理失败会通过异常链保留原始异常。再次中断或进程突然死亡可能阻止清理完成；最终清单无法排除延迟完成的 daemon 操作或并发变化。

显式检查在 Python 优化模式下也会执行。Python 检查通过数据源代理提交手写 PromQL，并未操作下拉框或验证变量插值。逗号连接与通配符对照分别要求无序列及 112，这两项都不是下拉框测试。合成数据中，production.eu 有十个工作实例，staging 有两个，productionXeu 有一百个。只选择 production.eu 时，正确面板应显示十，刻意不转义的对照面板则显示一百一十。选择两个可用环境后，再选择全部选项，正确面板都应显示十二。此时对照面板会为空，因为原样格式生成的是逗号连接文本，不能匹配预期的两个正则候选项。这两类失败都有真实数据作对照，而不是只检查表达式看起来是否合理。

浏览器每次切换后，应同时保留 Query Inspector 中送出的表达式、返回值及当前显示的选择；多选时还须检查重复 URL 参数。后台输出明确标记仪表盘插值尚未检查，不能据此升级 UI 证据。早先作者报告使用 Chrome 152 检查三种选择、送出查询、返回值及重复 URL 参数。本轮没有找回对应的精确浏览器捕获，也没有重跑容器或浏览器。这些历史观察针对合成仪表盘，不证明租户隔离或动态发现下的新鲜度。修改仪表盘时，可以保留被排除的环境作为回归数据；当变量类型、格式覆盖、刷新策略或数据源改变，都应重新检查最终表达式，而不只检查下拉框还能否打开。

延伸阅读

来源