6.2 グラフとチャート ​

#!/usr/bin/env python3
"""Zabbix カスタムグラフ生成スクリプト"""

import json
import requests
from typing import List, Dict, Any
from datetime import datetime, timedelta

class ZabbixGraphGenerator:
    def __init__(self, zabbix_url: str, auth_token: str):
        self.zabbix_url = zabbix_url
        self.auth_token = auth_token
    
    def create_performance_graph(self, host_name: str, metrics: List[Dict[str, Any]]) -> str:
        """パフォーマンス監視グラフ作成"""
        
        # ホストID取得
        host_data = {
            "jsonrpc": "2.0",
            "method": "host.get",
            "params": {
                "filter": {"host": [host_name]},
                "output": ["hostid"]
            },
            "auth": self.auth_token,
            "id": 1
        }
        
        response = requests.post(self.zabbix_url, json=host_data)
        host_id = response.json()["result"][0]["hostid"]
        
        # アイテム情報取得
        graph_items = []
        for idx, metric in enumerate(metrics):
            item_data = {
                "jsonrpc": "2.0",
                "method": "item.get",
                "params": {
                    "hostids": [host_id],
                    "search": {"key_": metric["key"]},
                    "output": ["itemid", "name", "units"]
                },
                "auth": self.auth_token,
                "id": 1
            }
            
            response = requests.post(self.zabbix_url, json=item_data)
            item = response.json()["result"][0]
            
            graph_items.append({
                "itemid": item["itemid"],
                "color": metric.get("color", self._get_default_color(idx)),
                "drawtype": metric.get("drawtype", 0),  # 0=line
                "sortorder": idx,
                "yaxisside": metric.get("yaxisside", 0),  # 0=left, 1=right
                "calc_fnc": metric.get("calc_fnc", 2)  # 2=avg
            })
        
        # グラフ作成
        graph_data = {
            "jsonrpc": "2.0",
            "method": "graph.create",
            "params": {
                "name": f"Performance Overview - {host_name}",
                "width": 900,
                "height": 200,
                "ymin_type": 0,  # 0=calculated
                "ymax_type": 0,
                "show_work_period": 1,
                "show_triggers": 1,
                "graphtype": 0,  # 0=normal
                "show_legend": 1,
                "show_3d": 0,
                "percent_left": 0,
                "percent_right": 0,
                "ymin_itemid": 0,
                "ymax_itemid": 0,
                "gitems": graph_items
            },
            "auth": self.auth_token,
            "id": 1
        }
        
        response = requests.post(self.zabbix_url, json=graph_data)
        result = response.json()
        
        if "result" in result:
            return result["result"]["graphids"][0]
        else:
            raise Exception(f"Graph creation failed: {result}")
    
    def create_stacked_graph(self, host_name: str, title: str, items: List[str]) -> str:
        """積み重ねグラフ作成"""
        
        # ホストID取得（省略...上記と同様）
        
        graph_items = []
        colors = ["#FF6B6B", "#4ECDC4", "#45B7D1", "#96CEB4", "#FECA57"]
        
        for idx, item_key in enumerate(items):
            # アイテムID取得（省略...）
            
            graph_items.append({
                "itemid": item_id,
                "color": colors[idx % len(colors)],
                "drawtype": 2,  # 2=filled area
                "sortorder": idx,
                "yaxisside": 0,
                "calc_fnc": 2
            })
        
        # 積み重ねグラフ設定
        graph_data = {
            "jsonrpc": "2.0",
            "method": "graph.create",
            "params": {
                "name": title,
                "width": 900,
                "height": 200,
                "graphtype": 1,  # 1=stacked
                "show_legend": 1,
                "percent_left": 95.0,  # 95%線表示
                "gitems": graph_items
            },
            "auth": self.auth_token,
            "id": 1
        }
        
        response = requests.post(self.zabbix_url, json=graph_data)
        return response.json()["result"]["graphids"][0]
    
    def _get_default_color(self, index: int) -> str:
        """デフォルト色セット"""
        colors = [
            "#1f77b4", "#ff7f0e", "#2ca02c", "#d62728", "#9467bd",
            "#8c564b", "#e377c2", "#7f7f7f", "#bcbd22", "#17becf"
        ]
        return colors[index % len(colors)]
    
    def update_graph_threshold_lines(self, graph_id: str, thresholds: List[Dict[str, Any]]):
        """グラフに閾値線追加"""
        
        # 現在のグラフ設定取得
        get_data = {
            "jsonrpc": "2.0",
            "method": "graph.get",
            "params": {
                "graphids": [graph_id],
                "selectGraphItems": "extend"
            },
            "auth": self.auth_token,
            "id": 1
        }
        
        response = requests.post(self.zabbix_url, json=get_data)
        graph = response.json()["result"][0]
        
        # 閾値線アイテム追加
        for threshold in thresholds:
            # 仮想アイテム（閾値用）作成は複雑なため、
            # 実際の実装では既存のアイテムを使用するか、
            # 計算アイテムを活用することを推奨
            pass
        
        return graph_id

# 使用例
def main():
    generator = ZabbixGraphGenerator(
        zabbix_url="https://zabbix.example.com/api_jsonrpc.php",
        auth_token="auth_token_here"
    )
    
    # パフォーマンスグラフ作成
    performance_metrics = [
        {
            "key": "system.cpu.util",
            "color": "FF6B6B",
            "yaxisside": 0
        },
        {
            "key": "vm.memory.utilization",
            "color": "4ECDC4",
            "yaxisside": 0
        },
        {
            "key": "system.cpu.load[percpu,avg5]",
            "color": "45B7D1",
            "yaxisside": 1
        }
    ]
    
    graph_id = generator.create_performance_graph("web-server-01", performance_metrics)
    print(f"Created performance graph: {graph_id}")
    
    # 積み重ねグラフ作成
    memory_items = [
        "vm.memory.size[used]",
        "vm.memory.size[buffers]",
        "vm.memory.size[cached]"
    ]
    
    stack_graph_id = generator.create_stacked_graph(
        "web-server-01",
        "Memory Usage Breakdown",
        memory_items
    )
    print(f"Created stacked graph: {stack_graph_id}")

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""マルチホスト比較グラフ生成"""

def create_cluster_comparison_graph(hosts: List[str], metric_key: str):
    """クラスター比較グラフ作成"""
    
    graph_config = {
        "name": f"Cluster {metric_key} Comparison",
        "width": 1200,
        "height": 400,
        "show_legend": 1,
        "gitems": []
    }
    
    colors = [
        "#FF6B6B", "#4ECDC4", "#45B7D1", "#96CEB4", "#FECA57",
        "#FF8E7B", "#6C5CE7", "#A29BFE", "#FD79A8", "#FDCB6E"
    ]
    
    for idx, host in enumerate(hosts):
        # ホスト別アイテム追加
        graph_config["gitems"].append({
            "itemid": get_item_id(host, metric_key),
            "color": colors[idx % len(colors)],
            "drawtype": 0,  # 線グラフ
            "sortorder": idx,
            "yaxisside": 0,
            "calc_fnc": 2  # 平均
        })
    
    return create_graph(graph_config)

def create_percentile_graph(hosts: List[str], metric_key: str):
    """パーセンタイルグラフ作成"""
    
    # 各ホストのデータを取得してパーセンタイル計算
    all_data = []
    for host in hosts:
        data = get_historical_data(host, metric_key, period="24h")
        all_data.extend(data)
    
    # パーセンタイル計算
    percentiles = {
        "P50": numpy.percentile(all_data, 50),
        "P75": numpy.percentile(all_data, 75),
        "P90": numpy.percentile(all_data, 90),
        "P95": numpy.percentile(all_data, 95),
        "P99": numpy.percentile(all_data, 99)
    }
    
    # パーセンタイル線をグラフに追加
    graph_config = {
        "name": f"Cluster {metric_key} Percentiles",
        "annotations": []
    }
    
    for label, value in percentiles.items():
        graph_config["annotations"].append({
            "type": "line",
            "yMin": value,
            "yMax": value,
            "borderColor": get_percentile_color(label),
            "label": {
                "content": f"{label}: {value:.2f}",
                "enabled": True
            }
        })
    
    return graph_config

#!/usr/bin/env python3
"""Zabbixグラフ画像エクスポート"""

import requests
import base64
from datetime import datetime, timedelta

class ZabbixGraphExporter:
    def __init__(self, zabbix_url: str, username: str, password: str):
        self.base_url = zabbix_url.replace('/api_jsonrpc.php', '')
        self.session = requests.Session()
        self.login(username, password)
    
    def login(self, username: str, password: str):
        """Webインターフェースログイン"""
        login_data = {
            'name': username,
            'password': password,
            'enter': 'Sign in'
        }
        
        response = self.session.post(
            f"{self.base_url}/index.php",
            data=login_data
        )
        
        if "Dashboard" not in response.text:
            raise Exception("Login failed")
    
    def export_graph_image(self, graph_id: str, width: int = 900, height: int = 200,
                          period: int = 3600, output_format: str = 'PNG') -> bytes:
        """グラフ画像エクスポート"""
        
        # 時間範囲設定
        to_time = datetime.now()
        from_time = to_time - timedelta(seconds=period)
        
        params = {
            'graphid': graph_id,
            'width': width,
            'height': height,
            'from': int(from_time.timestamp()),
            'to': int(to_time.timestamp()),
            'profileIdx': 'web.graphs.filter',
            'profileIdx2': graph_id
        }
        
        response = self.session.get(
            f"{self.base_url}/chart2.php",
            params=params
        )
        
        return response.content
    
    def export_custom_graph(self, items: List[str], title: str,
                           width: int = 900, height: int = 200,
                           period: int = 3600) -> bytes:
        """カスタムグラフ画像生成"""
        
        # アイテムIDをカンマ区切りで連結
        itemids = ','.join(items)
        
        params = {
            'itemids[0]': itemids,
            'type': 0,  # 線グラフ
            'period': period,
            'width': width,
            'height': height,
            'name': title,
            'legend': 1,
            'graph_type': 'png'
        }
        
        response = self.session.get(
            f"{self.base_url}/chart.php",
            params=params
        )
        
        return response.content
    
    def create_dashboard_snapshot(self, dashboard_id: str,
                                 width: int = 1920, height: int = 1080) -> bytes:
        """ダッシュボードスナップショット作成"""
        
        # Selenium WebDriverを使用してブラウザ制御
        from selenium import webdriver
        from selenium.webdriver.chrome.options import Options
        
        chrome_options = Options()
        chrome_options.add_argument('--headless')
        chrome_options.add_argument(f'--window-size={width},{height}')
        chrome_options.add_argument('--disable-gpu')
        
        driver = webdriver.Chrome(options=chrome_options)
        
        try:
            # ダッシュボードページにアクセス
            dashboard_url = f"{self.base_url}/zabbix.php?action=dashboard.view&dashboardid={dashboard_id}"
            driver.get(dashboard_url)
            
            # ログイン（必要に応じて）
            # ...
            
            # スクリーンショット撮影
            screenshot = driver.get_screenshot_as_png()
            return screenshot
            
        finally:
            driver.quit()
    
    def generate_report_images(self, report_config: Dict[str, Any]) -> Dict[str, bytes]:
        """レポート用画像一括生成"""
        
        images = {}
        
        for section in report_config['sections']:
            if section['type'] == 'graph':
                image_data = self.export_graph_image(
                    graph_id=section['graph_id'],
                    width=section.get('width', 900),
                    height=section.get('height', 200),
                    period=section.get('period', 3600)
                )
                images[section['name']] = image_data
                
            elif section['type'] == 'custom':
                image_data = self.export_custom_graph(
                    items=section['items'],
                    title=section['title'],
                    width=section.get('width', 900),
                    height=section.get('height', 200),
                    period=section.get('period', 3600)
                )
                images[section['name']] = image_data
        
        return images

# 使用例
def main():
    exporter = ZabbixGraphExporter(
        zabbix_url="https://zabbix.example.com",
        username="admin",
        password="password"
    )
    
    # 単一グラフエクスポート
    graph_image = exporter.export_graph_image(
        graph_id="123",
        width=1200,
        height=300,
        period=86400  # 24時間
    )
    
    with open("performance_graph.png", "wb") as f:
        f.write(graph_image)
    
    # レポート用画像一括生成
    report_config = {
        'sections': [
            {
                'type': 'graph',
                'name': 'cpu_usage',
                'graph_id': '123',
                'width': 900,
                'height': 200,
                'period': 86400
            },
            {
                'type': 'custom',
                'name': 'memory_overview',
                'items': ['item1', 'item2', 'item3'],
                'title': 'Memory Usage Overview',
                'width': 900,
                'height': 200,
                'period': 86400
            }
        ]
    }
    
    images = exporter.generate_report_images(report_config)
    
    for name, image_data in images.items():
        with open(f"report_{name}.png", "wb") as f:
            f.write(image_data)

if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""グラフデータCSVエクスポート"""

import csv
import json
from datetime import datetime, timedelta

class ZabbixDataExporter:
    def __init__(self, zabbix_api):
        self.api = zabbix_api
    
    def export_graph_data_csv(self, graph_id: str, period_hours: int = 24,
                             output_file: str = None) -> str:
        """グラフデータをCSV形式でエクスポート"""
        
        # グラフ情報取得
        graph_data = self.api.graph.get(
            graphids=[graph_id],
            selectGraphItems="extend"
        )[0]
        
        # 時間範囲設定
        to_time = datetime.now()
        from_time = to_time - timedelta(hours=period_hours)
        
        # アイテムデータ取得
        csv_data = []
        headers = ['timestamp']
        
        for graph_item in graph_data['gitems']:
            item_id = graph_item['itemid']
            
            # アイテム情報取得
            item_info = self.api.item.get(
                itemids=[item_id],
                output=['name', 'units']
            )[0]
            
            headers.append(f"{item_info['name']} ({item_info['units']})")
            
            # 履歴データ取得
            history_data = self.api.history.get(
                itemids=[item_id],
                time_from=int(from_time.timestamp()),
                time_till=int(to_time.timestamp()),
                sortfield='clock',
                sortorder='ASC'
            )
            
            # データポイント追加
            for point in history_data:
                timestamp = datetime.fromtimestamp(int(point['clock']))
                
                # 既存の行を検索または新規作成
                row_found = False
                for row in csv_data:
                    if row[0] == timestamp:
                        row.append(float(point['value']))
                        row_found = True
                        break
                
                if not row_found:
                    new_row = [timestamp] + [None] * (len(headers) - 1)
                    new_row[len(headers) - 1] = float(point['value'])
                    csv_data.append(new_row)
        
        # 時刻でソート
        csv_data.sort(key=lambda x: x[0])
        
        # CSV出力
        if not output_file:
            output_file = f"graph_{graph_id}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        
        with open(output_file, 'w', newline='', encoding='utf-8') as csvfile:
            writer = csv.writer(csvfile)
            writer.writerow(headers)
            
            for row in csv_data:
                formatted_row = [row[0].strftime('%Y-%m-%d %H:%M:%S')] + row[1:]
                writer.writerow(formatted_row)
        
        return output_file
    
    def export_multi_host_comparison(self, hosts: List[str], metric_key: str,
                                   period_hours: int = 24) -> str:
        """マルチホスト比較データエクスポート"""
        
        # ホスト情報取得
        host_data = self.api.host.get(
            filter={'host': hosts},
            output=['hostid', 'host', 'name']
        )
        
        # 時間範囲設定
        to_time = datetime.now()
        from_time = to_time - timedelta(hours=period_hours)
        
        csv_data = []
        headers = ['timestamp'] + [f"{h['name']} ({h['host']})" for h in host_data]
        
        # 各ホストのデータ取得
        for host in host_data:
            items = self.api.item.get(
                hostids=[host['hostid']],
                search={'key_': metric_key},
                output=['itemid']
            )
            
            if not items:
                continue
                
            item_id = items[0]['itemid']
            
            history_data = self.api.history.get(
                itemids=[item_id],
                time_from=int(from_time.timestamp()),
                time_till=int(to_time.timestamp()),
                sortfield='clock',
                sortorder='ASC'
            )
            
            # データ統合
            for point in history_data:
                timestamp = datetime.fromtimestamp(int(point['clock']))
                
                # 既存の行を検索
                row_found = False
                for row in csv_data:
                    if row[0] == timestamp:
                        # ホストのインデックス計算
                        host_index = next(i for i, h in enumerate(host_data) if h['hostid'] == host['hostid'])
                        row[host_index + 1] = float(point['value'])
                        row_found = True
                        break
                
                if not row_found:
                    new_row = [timestamp] + [None] * len(host_data)
                    host_index = next(i for i, h in enumerate(host_data) if h['hostid'] == host['hostid'])
                    new_row[host_index + 1] = float(point['value'])
                    csv_data.append(new_row)
        
        # CSV出力
        output_file = f"multi_host_{metric_key}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
        
        with open(output_file, 'w', newline='', encoding='utf-8') as csvfile:
            writer = csv.writer(csvfile)
            writer.writerow(headers)
            
            # 時刻でソート
            csv_data.sort(key=lambda x: x[0])
            
            for row in csv_data:
                formatted_row = [row[0].strftime('%Y-%m-%d %H:%M:%S')] + row[1:]
                writer.writerow(formatted_row)
        
        return output_file