Grafana — Setup & Verification Guide

This skill teaches setup-gen and reward-gen how to create, manipulate, and verify Grafana state for monitoring/observability tasks. Grafana tasks involve HTTP state management through a shared Grafana instance.

• Libraries: requests, json, uuid, copy
• Grafana instance: https://cua-gym-grafana.xlang.ai
• Admin credentials: admin / cua-gym-admin-2024
• Grafana version: 12.4.1

---

0. Architecture Overview

Admin creates session org + user  →  User logs in via browser  →  CUA agent interacts with Grafana UI
         ↓                                    ↓                              ↓
  Inject dashboards/alerts            setup-gen configures state      reward-gen verifies via API
         ↓                                    ↓                              ↓
  TestData datasource               initial_setup.py runs              reward.py reads back state
         ↓                                                                   ↓
  csv_content = controlled data                                   DELETE org + user (cleanup)

Key design: Each training session gets its own Grafana Organization for complete isolation. Dashboards, datasources, alerts, and folders are org-scoped — a user in org N cannot see org M's resources.

Key difference from mock_websites: Grafana is a real application (not a mock). State is injected via Grafana's native REST API, not a custom /post?sid= endpoint. Verification reads state back via the same API.

---

1. Session Isolation via Organizations

Every training episode uses a dedicated Grafana Organization with its own user. This provides complete state isolation.

1.1 Session Setup Flow

import requests
import uuid
import json

GRAFANA_URL = 'https://cua-gym-grafana.xlang.ai'
ADMIN_AUTH = ('admin', 'cua-gym-admin-2024')

session_id = str(uuid.uuid4())[:8]
org_name = f'session_{session_id}'
user_login = f'agent_{session_id}'
user_password = f'agent_pass_{session_id}'
user_email = f'{user_login}@cua-gym.local'

# 1. Create organization
resp = requests.post(
    f'{GRAFANA_URL}/api/orgs',
    json={'name': org_name},
    auth=ADMIN_AUTH, timeout=15
)
assert resp.status_code == 200, f'Org creation failed: {resp.text}'
org_id = resp.json()['orgId']

# 2. Create user
resp = requests.post(
    f'{GRAFANA_URL}/api/admin/users',
    json={
        'name': f'Agent {session_id}',
        'login': user_login,
        'password': user_password,
        'email': user_email,
    },
    auth=ADMIN_AUTH, timeout=15
)
assert resp.status_code == 200, f'User creation failed: {resp.text}'
user_id = resp.json()['id']

# 3. Add user to org as Editor
resp = requests.post(
    f'{GRAFANA_URL}/api/orgs/{org_id}/users',
    json={'loginOrEmail': user_login, 'role': 'Editor'},
    auth=ADMIN_AUTH, timeout=15
)
assert resp.status_code == 200

# 4. Switch user's active org
resp = requests.post(
    f'{GRAFANA_URL}/api/users/{user_id}/using/{org_id}',
    auth=ADMIN_AUTH, timeout=15
)
assert resp.status_code == 200

# 5. Create TestData datasource in the new org
resp = requests.post(
    f'{GRAFANA_URL}/api/datasources',
    json={
        'name': 'TestData',
        'type': 'grafana-testdata-datasource',
        'access': 'proxy',
        'isDefault': True,
    },
    auth=ADMIN_AUTH,
    headers={'X-Grafana-Org-Id': str(org_id)},
    timeout=15
)
assert resp.status_code == 200
ds_uid = resp.json()['datasource']['uid']

1.2 Session Cleanup

def cleanup_session(org_id, user_id):
    """Delete org and user. Order matters: org first, then user."""
    requests.delete(f'{GRAFANA_URL}/api/orgs/{org_id}', auth=ADMIN_AUTH, timeout=15)
    requests.delete(f'{GRAFANA_URL}/api/admin/users/{user_id}', auth=ADMIN_AUTH, timeout=15)

1.3 Persisting Session Info

# Save session info for golden_patch.py and reward.py
session_info = {
    'grafana_url': GRAFANA_URL,
    'org_id': org_id,
    'org_name': org_name,
    'user_id': user_id,
    'user_login': user_login,
    'user_password': user_password,
    'ds_uid': ds_uid,
}
with open('/tmp/task_grafana_session', 'w') as f:
    json.dump(session_info, f)

---

2. Dashboard API

The primary setup and verification surface. Dashboards are JSON documents containing panels, variables, and layout.

2.1 Create Dashboard

def create_dashboard(dashboard_json, org_id, folder_uid=''):
    """Create or overwrite a dashboard."""
    payload = {
        'dashboard': dashboard_json,
        'overwrite': True,
        'folderUid': folder_uid,
    }
    resp = requests.post(
        f'{GRAFANA_URL}/api/dashboards/db',
        json=payload,
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    assert resp.status_code == 200, f'Dashboard creation failed: {resp.text}'
    return resp.json()  # {'uid', 'url', 'id', 'status', 'version'}

2.2 Read Dashboard (for reward verification)

def get_dashboard(uid, org_id):
    """Read back a dashboard. Returns full JSON model."""
    resp = requests.get(
        f'{GRAFANA_URL}/api/dashboards/uid/{uid}',
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    assert resp.status_code == 200, f'Dashboard read failed: {resp.text}'
    result = resp.json()
    return result['dashboard'], result['meta']
    # dashboard = full JSON model (panels, templating, time, tags, etc.)
    # meta = {folderUid, folderTitle, slug, url, version, ...}

2.3 Delete Dashboard

def delete_dashboard(uid, org_id):
    resp = requests.delete(
        f'{GRAFANA_URL}/api/dashboards/uid/{uid}',
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    return resp.status_code == 200

2.4 List Dashboards (search)

def search_dashboards(org_id, query='', tag=''):
    params = {}
    if query:
        params['query'] = query
    if tag:
        params['tag'] = tag
    resp = requests.get(
        f'{GRAFANA_URL}/api/search',
        params=params,
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    return resp.json()  # [{uid, title, url, tags, type, folderUid, ...}]

---

3. Controlled Data via csv_content

The TestData datasource's csv_content scenario embeds exact data inline in the dashboard JSON. This makes data fully deterministic and verifiable.

3.1 CSV Content Target

def make_csv_target(ref_id, csv_data):
    """Create a panel target with exact inline data.

    csv_data: str with header row + data rows.
    First column should be 'time' for timeseries panels.
    """
    return {
        'refId': ref_id,
        'scenarioId': 'csv_content',
        'csvContent': csv_data,
    }

# Example: CPU metrics for 3 servers over 8 hours
csv = (
    "time,web-server-01,web-server-02,db-primary\n"
    "2024-01-15T00:00:00Z,25,30,45\n"
    "2024-01-15T01:00:00Z,28,32,48\n"
    "2024-01-15T02:00:00Z,35,29,52\n"
    "2024-01-15T03:00:00Z,72,31,55\n"
    "2024-01-15T04:00:00Z,95,33,90\n"  # anomaly!
    "2024-01-15T05:00:00Z,88,35,85\n"
    "2024-01-15T06:00:00Z,42,30,50\n"
    "2024-01-15T07:00:00Z,30,28,47"
)
target = make_csv_target('A', csv)

3.2 Random Walk Target (for visual richness, non-verifiable data)

def make_random_walk_target(ref_id, alias, start_value=50, spread=10):
    """Random walk — produces live-updating data. NOT deterministic."""
    return {
        'refId': ref_id,
        'scenarioId': 'random_walk',
        'alias': alias,
        'seriesCount': 1,
        'startValue': start_value,
        'spread': spread,
    }

3.3 When to Use Which

Scenario	Use Case	Verifiable?
csv_content	Task requires specific data patterns (anomalies, trends)	Yes — data is in the JSON
random_walk	Visual richness, background panels	No — data changes on each load

---

4. Panel Types & Configuration

4.1 Supported Panel Types

Type Key	Display Name	Typical Use
timeseries	Time Series	Metrics over time (default)
stat	Stat	Single big number with sparkline
gauge	Gauge	Dial showing value vs threshold
bargauge	Bar Gauge	Horizontal/vertical bar
table	Table	Tabular data display
barchart	Bar Chart	Categorical comparison
piechart	Pie Chart	Proportional breakdown
heatmap	Heatmap	2D density visualization
histogram	Histogram	Value distribution
state-timeline	State Timeline	State changes over time
status-history	Status History	Periodic state display
geomap	Geomap	Geographic data
candlestick	Candlestick	Financial data
xychart	XY Chart	Arbitrary x/y plotting
canvas	Canvas	Free-form layout
text	Text	Markdown/HTML content
logs	Logs	Log data display
nodeGraph	Node Graph	Network/graph visualization
alertlist	Alert List	Active alerts display
dashlist	Dashboard List	Dashboard links
news	News	RSS feed

4.2 Panel JSON Structure

def make_panel(panel_id, title, panel_type, grid_x, grid_y, grid_w, grid_h,
               targets, ds_uid, field_config=None, options=None, transformations=None):
    """Build a complete panel JSON object."""
    panel = {
        'id': panel_id,
        'title': title,
        'type': panel_type,
        'gridPos': {'x': grid_x, 'y': grid_y, 'w': grid_w, 'h': grid_h},
        'datasource': {
            'type': 'grafana-testdata-datasource',
            'uid': ds_uid,
        },
        'targets': targets,
        'fieldConfig': field_config or {
            'defaults': {},
            'overrides': [],
        },
        'options': options or {},
    }
    if transformations:
        panel['transformations'] = transformations
    return panel

4.3 Field Config — Units, Thresholds, Overrides

# Standard field config with thresholds
field_config = {
    'defaults': {
        'unit': 'percent',        # bytes, ms, reqps, short, etc.
        'decimals': 1,
        'min': 0,
        'max': 100,
        'color': {'mode': 'palette-classic'},
        'custom': {
            'lineWidth': 2,
            'fillOpacity': 10,
            'stacking': {'mode': 'none'},  # 'normal' for stacked
        },
        'thresholds': {
            'mode': 'absolute',    # or 'percentage'
            'steps': [
                {'color': 'green', 'value': None},   # base (always None)
                {'color': 'yellow', 'value': 60},
                {'color': 'red', 'value': 80},
            ],
        },
        'links': [
            # data links
            {
                'title': 'View Details',
                'url': 'https://example.com/details?server=${__field.name}',
                'targetBlank': True,
            },
        ],
    },
    'overrides': [
        {
            'matcher': {'id': 'byName', 'options': 'web-server-01'},
            'properties': [
                {'id': 'color', 'value': {'fixedColor': 'red', 'mode': 'fixed'}},
                {'id': 'custom.lineWidth', 'value': 3},
            ],
        },
    ],
}

4.4 Panel Options (per visualization type)

# Time series options
timeseries_options = {
    'tooltip': {'mode': 'multi'},           # 'single', 'multi', 'none'
    'legend': {
        'displayMode': 'table',             # 'list', 'table', 'hidden'
        'placement': 'bottom',              # 'bottom', 'right'
        'calcs': ['mean', 'max', 'last'],   # legend calculations
    },
}

# Stat options
stat_options = {
    'graphMode': 'area',        # 'area', 'none'
    'colorMode': 'background',  # 'none', 'value', 'background'
    'textMode': 'value',        # 'auto', 'value', 'value_and_name', 'name', 'none'
    'orientation': 'auto',      # 'auto', 'horizontal', 'vertical'
}

# Gauge options
gauge_options = {
    'showThresholdLabels': True,
    'showThresholdMarkers': True,
    'orientation': 'auto',
}

# Pie chart options
pie_options = {
    'pieType': 'pie',           # 'pie', 'donut'
    'tooltip': {'mode': 'single'},
    'legend': {'displayMode': 'list', 'placement': 'right'},
}

# Bar chart options
bar_options = {
    'orientation': 'auto',      # 'auto', 'horizontal', 'vertical'
    'stacking': 'none',         # 'none', 'normal', 'percent'
    'showValue': 'auto',        # 'auto', 'always', 'never'
    'barWidth': 0.97,
    'groupWidth': 0.7,
}

# Table options
table_options = {
    'showHeader': True,
    'footer': {'show': False},
    'cellHeight': 'sm',         # 'sm', 'md', 'lg'
}

4.5 Transformations

# Organize fields (rename, reorder, hide)
transform_organize = {
    'id': 'organize',
    'options': {
        'excludeByName': {'Time': True},
        'renameByName': {'web-server-01': 'Web Server 1'},
        'indexByName': {'web-server-01': 0, 'web-server-02': 1},
    },
}

# Filter by value
transform_filter = {
    'id': 'filterByValue',
    'options': {
        'filters': [
            {
                'fieldName': 'web-server-01',
                'config': {
                    'id': 'greater',
                    'options': {'value': 50},
                },
            },
        ],
        'type': 'include',
        'match': 'any',
    },
}

---

5. Template Variables

Dashboard variables enable dynamic filtering via dropdown menus.

5.1 Custom Variable (static options)

templating = {
    'list': [
        {
            'type': 'custom',
            'name': 'server',
            'label': 'Server',
            'query': 'web-server-01,web-server-02,web-server-03',
            'current': {'text': 'web-server-01', 'value': 'web-server-01'},
            'options': [
                {'text': 'web-server-01', 'value': 'web-server-01', 'selected': True},
                {'text': 'web-server-02', 'value': 'web-server-02', 'selected': False},
                {'text': 'web-server-03', 'value': 'web-server-03', 'selected': False},
            ],
            'multi': False,          # True for multi-select
            'includeAll': False,     # True to add "All" option
            'hide': 0,              # 0=show, 1=hide label, 2=hide entirely
        },
        {
            'type': 'interval',
            'name': 'interval',
            'label': 'Interval',
            'query': '1m,5m,15m,30m,1h',
            'current': {'text': '5m', 'value': '5m'},
            'auto': False,
            'hide': 0,
        },
    ],
}

---

6. Alerting API

6.1 Prerequisites

Alert rules require a folder. Always create the folder first.

def create_folder(title, uid, org_id):
    resp = requests.post(
        f'{GRAFANA_URL}/api/folders',
        json={'title': title, 'uid': uid},
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    assert resp.status_code == 200, f'Folder creation failed: {resp.text}'
    return resp.json()

6.2 Create Alert Rule

def create_alert_rule(title, folder_uid, ds_uid, org_id,
                      threshold=80, pending_for='5m',
                      labels=None, annotations=None):
    """Create a threshold-based alert rule."""
    rule = {
        'title': title,
        'ruleGroup': 'cua-gym-alerts',
        'folderUID': folder_uid,
        'condition': 'C',
        'data': [
            {
                'refId': 'A',
                'relativeTimeRange': {'from': 600, 'to': 0},
                'datasourceUid': ds_uid,
                'model': {
                    'scenarioId': 'random_walk',
                    'seriesCount': 1,
                    'startValue': 50,
                    'spread': 20,
                    'refId': 'A',
                },
            },
            {
                'refId': 'C',
                'relativeTimeRange': {'from': 600, 'to': 0},
                'datasourceUid': '__expr__',
                'model': {
                    'type': 'threshold',
                    'expression': 'A',
                    'conditions': [
                        {
                            'evaluator': {
                                'type': 'gt',
                                'params': [threshold],
                            },
                        },
                    ],
                    'refId': 'C',
                },
            },
        ],
        'for': pending_for,
        'noDataState': 'NoData',
        'execErrState': 'Error',
        'labels': labels or {},
        'annotations': annotations or {},
    }
    resp = requests.post(
        f'{GRAFANA_URL}/api/v1/provisioning/alert-rules',
        json=rule,
        auth=ADMIN_AUTH,
        headers={
            'X-Grafana-Org-Id': str(org_id),
            'X-Disable-Provenance': 'true',
        },
        timeout=15,
    )
    assert resp.status_code == 201, f'Alert rule creation failed: {resp.text}'
    return resp.json()  # includes 'uid'

6.3 Read Alert Rule (for reward verification)

def get_alert_rule(rule_uid, org_id):
    resp = requests.get(
        f'{GRAFANA_URL}/api/v1/provisioning/alert-rules/{rule_uid}',
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    assert resp.status_code == 200
    return resp.json()
    # Fields: uid, title, condition, data, for, labels, annotations, isPaused, ...

6.4 List All Alert Rules

def list_alert_rules(org_id):
    resp = requests.get(
        f'{GRAFANA_URL}/api/v1/provisioning/alert-rules',
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    return resp.json()  # list of alert rule objects

6.5 Delete Alert Rule

def delete_alert_rule(rule_uid, org_id):
    requests.delete(
        f'{GRAFANA_URL}/api/v1/provisioning/alert-rules/{rule_uid}',
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )

---

7. Contact Points & Notification Policies

7.1 Create Contact Point

def create_contact_point(name, cp_type, settings, org_id):
    """Create a contact point.

    cp_type: 'email', 'webhook', 'slack', etc.
    settings: type-specific dict (see examples below).
    """
    resp = requests.post(
        f'{GRAFANA_URL}/api/v1/provisioning/contact-points',
        json={
            'name': name,
            'type': cp_type,
            'settings': settings,
            'disableResolveMessage': False,
        },
        auth=ADMIN_AUTH,
        headers={
            'X-Grafana-Org-Id': str(org_id),
            'X-Disable-Provenance': 'true',
        },
        timeout=15,
    )
    assert resp.status_code == 202, f'Contact point creation failed: {resp.text}'
    return resp.json()

# Email settings
email_settings = {'addresses': 'oncall@example.com;alerts@example.com'}

# Webhook settings
webhook_settings = {
    'url': 'https://hooks.example.com/grafana',
    'httpMethod': 'POST',
}

7.2 Read Contact Points (for verification)

def list_contact_points(org_id):
    resp = requests.get(
        f'{GRAFANA_URL}/api/v1/provisioning/contact-points',
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    return resp.json()  # list of {uid, name, type, settings, ...}

7.3 Notification Policy Tree

def get_notification_policies(org_id):
    resp = requests.get(
        f'{GRAFANA_URL}/api/v1/provisioning/policies',
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    return resp.json()  # full routing tree

def update_notification_policies(policy_tree, org_id):
    """WARNING: This replaces the ENTIRE policy tree.
    Always GET first to preserve the root receiver."""
    resp = requests.put(
        f'{GRAFANA_URL}/api/v1/provisioning/policies',
        json=policy_tree,
        auth=ADMIN_AUTH,
        headers={
            'X-Grafana-Org-Id': str(org_id),
            'X-Disable-Provenance': 'true',
        },
        timeout=15,
    )
    assert resp.status_code == 202

# Example: route critical alerts to a specific contact point
policy_tree = {
    'receiver': 'grafana-default-email',  # keep existing default
    'group_by': ['alertname'],
    'routes': [
        {
            'receiver': 'oncall-webhook',
            'object_matchers': [
                ['severity', '=', 'critical'],
            ],
            'continue': False,
        },
    ],
}

---

8. Mute Timings

def create_mute_timing(name, time_intervals, org_id):
    resp = requests.post(
        f'{GRAFANA_URL}/api/v1/provisioning/mute-timings',
        json={'name': name, 'time_intervals': time_intervals},
        auth=ADMIN_AUTH,
        headers={
            'X-Grafana-Org-Id': str(org_id),
            'X-Disable-Provenance': 'true',
        },
        timeout=15,
    )
    assert resp.status_code == 201
    return resp.json()

# Weekend mute
weekend_intervals = [
    {
        'weekdays': ['saturday', 'sunday'],
        'times': [{'start_time': '00:00', 'end_time': '23:59'}],
    },
]

# Maintenance window
maintenance_intervals = [
    {
        'weekdays': ['wednesday'],
        'times': [{'start_time': '02:00', 'end_time': '06:00'}],
    },
]

def list_mute_timings(org_id):
    resp = requests.get(
        f'{GRAFANA_URL}/api/v1/provisioning/mute-timings',
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    return resp.json()

---

9. Annotations

def create_annotation(text, tags, org_id, dashboard_uid=None, panel_id=None,
                      time_ms=None, time_end_ms=None):
    body = {'text': text, 'tags': tags}
    if dashboard_uid:
        body['dashboardUID'] = dashboard_uid
    if panel_id:
        body['panelId'] = panel_id
    if time_ms:
        body['time'] = time_ms
    if time_end_ms:
        body['timeEnd'] = time_end_ms
    resp = requests.post(
        f'{GRAFANA_URL}/api/annotations',
        json=body,
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    assert resp.status_code == 200
    return resp.json()  # {'id', 'message'}

def query_annotations(org_id, tags=None, dashboard_uid=None):
    """Query annotations. Tags use repeated params: tags=a&tags=b."""
    params = {}
    if dashboard_uid:
        params['dashboardUID'] = dashboard_uid
    resp = requests.get(
        f'{GRAFANA_URL}/api/annotations',
        params=params,
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    results = resp.json()
    if tags:
        results = [a for a in results if set(tags).issubset(set(a.get('tags', [])))]
    return results

---

10. Folders

def create_folder(title, uid, org_id, parent_uid=None):
    body = {'title': title, 'uid': uid}
    if parent_uid:
        body['parentUid'] = parent_uid
    resp = requests.post(
        f'{GRAFANA_URL}/api/folders',
        json=body,
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    assert resp.status_code == 200, f'Folder creation failed: {resp.text}'
    return resp.json()

def get_folder(uid, org_id):
    resp = requests.get(
        f'{GRAFANA_URL}/api/folders/{uid}',
        auth=ADMIN_AUTH,
        headers={'X-Grafana-Org-Id': str(org_id)},
        timeout=15,
    )
    return resp.json()