PrometheusとAlertmanagerを連携させ、Slack経由でパトライトを制御するまで

この記事では、Prometheusで取得したサーバーのメトリクス（例：ディスク使用率）に基づいてアラートを発報し、そのアラートをAlertmanager経由でSlackに通知し、さらにSlackメッセージをフックしてパトライトを制御するまでの一連のシステム構築手順を解説します。

複雑な複数のコンポーネントを連携させるため、各手順を順番に実行することで、スムーズにシステムのセットアップが完了できるように構成しています。

全体像の把握

まず、今回構築するシステムの全体像を理解しましょう。データの流れは以下の通りです。

1. Prometheus: サーバーのディスク使用率を監視します。
   設定した閾値を超えるとアラートを発報します。
2. Alertmanager: Prometheusから受け取ったアラートを、Inhibition Rule（より深刻なアラートが発報されたら、軽いアラートの通知を抑制するルール）に基づいて整理し、Slackへ通知します。
3. Slack: Alertmanagerから届いたアラートメッセージを受け取ります。
4. slack_patlite_app.py: Slackに届いたアラートメッセージをリアルタイムで検知し、その内容を解析します。
5. パトライト: スクリプトからの命令に応じて点灯・点滅し、サーバーの状態を視覚的に表示します。

1. PrometheusとAlertmanagerのインストール

今回は、既にインストールが完了している前提で進めます。
まだの方は、以前の記事、公式サイトの手順に従ってPrometheusとAlertmanagerをインストールしてください。

• Prometheus公式ドキュメント: https://prometheus.io/docs/prometheus/latest/installation/
• Alertmanager公式ドキュメント: https://prometheus.io/docs/alerting/latest/alertmanager/

2. Slackアプリの作成と設定

Slackでアラート通知を受け取るためのアプリケーションを作成し、Webhook URLを取得します。

Slackアプリの作成

Slack APIサイト にアクセスし、「Create New App」をクリックします。

「From an app manifest」を選択し、以下をYAMLファイルとして入力します。
display_name と bot_user の名前は任意です。

display_name: サーバー監視
description: PrometheusとAlertmanagerの通知を受け取るためのアプリです。
features:
  bot_user:
    display_name: Prometheus Alert Bot
    always_online: false
  outgoing_webhooks: []
  slash_commands: []
  shortcuts: []
  actions: []
oauth_config:
  scopes:
    bot:
      - chat:write
      - channels:read
settings:
  event_subscriptions:
    request_url: ''
    bot_events: []
  interactivity:
   is_enabled: false
  org_deploy_enabled: false
  socket_mode_enabled: true
  token_rotation_enabled: false

Webhook URLの取得

作成したアプリのページで、「Incoming Webhooks」を有効にします。

「Add New Webhook to Workspace」をクリックし、通知先のチャンネルを選択して「許可する」をクリックします。

生成された Webhook URL をメモしておきます。
これは後ほどAlertmanagerの設定で使用します。

App Tokenの取得

アプリのページで「Basic Information」から「App-Level Tokens」セクションに移動します。

「Generate tokens and scopes」をクリックし、以下のスコープを持つトークンを作成します。connections:write

生成された App Token（xapp-で始まる文字列） をメモしておきます。
これは後ほどPythonスクリプトの設定で使用します。

Bot User Tokenの取得

アプリのページで「OAuth & Permissions」に移動します。

「Install to Workspace」をクリックし、アプリをワークスペースにインストールします。

生成された Bot User OAuth Token（xoxb-で始まる文字列） をメモしておきます。
これもPythonスクリプトの設定で使用します。

3. PrometheusとAlertmanagerの設定

PrometheusとAlertmanagerの設定ファイルを修正し、アラートの発報とSlackへの通知を連携させます。

Prometheusの設定 (/etc/prometheus/prometheus.yml)

ディスク使用率のアラートルールを定義します。

# prometheus.yml

alerting:
  alertmanagers:
  - static_configs:
    - targets: ['localhost:9093'] # Alertmanagerのアドレス

rule_files:
  - "alert_disk_space.yml"

# ... (その他設定)


Prometheusのアラートルール(/etc/prometheus/alert_disk_space.yml)

3段階のSeverityを持つアラートを定義します。

# alert_disk_space.yml

groups:
  - name: disk_usage_alerts
    rules:
      - alert: DiskUsage
        expr: 100 - (node_filesystem_free_bytes{fstype="ext4"} / node_filesystem_size_bytes{fstype="ext4"}) * 100 > 80
        for: 1m
        labels:
          severity: critical
        annotations:
          summary: ディスクの使用率が80%を超えました
          description: ホスト {{ $labels.instance }} のディスク使用率が 80%を超えています。直ちに対応が必要です。
      
      - alert: DiskUsage
        expr: 100 - (node_filesystem_free_bytes{fstype="ext4"} / node_filesystem_size_bytes{fstype="ext4"}) * 100 > 70
        for: 1m
        labels:
          severity: warning
        annotations:
          summary: ディスクの使用率が70%を超えました
          description: ホスト {{ $labels.instance }} のディスク使用率が 70%を超えています。容量増加を検討してください。

      - alert: DiskUsage
        expr: 100 - (node_filesystem_free_bytes{fstype="ext4"} / node_filesystem_size_bytes{fstype="ext4"}) * 100 > 50
        for: 1m
        labels:
          severity: alert
        annotations:
          summary: ディスクの使用率が50%を超えました
          description: ホスト {{ $labels.instance }} のディスク使用率が 50%を超えています。定期的な監視をお勧めします。


Alertmanagerの設定 (/etc/prometheus/alertmanager.yml)

Slackへの通知と、Severityによる抑制ルールを定義します。
特にgroup_byは重要です。

# alertmanager.yml

global:
  resolve_timeout: 5m

route:
  group_by: ['alertname', 'instance', 'cluster', 'service']
  group_wait: 30s
  group_interval: 5m
  repeat_interval: 3h
  receiver: default-receiver

inhibit_rules:
  - source_matchers: [severity="critical"]
    target_matchers: [severity="warning"]
    equal: [alertname, instance, cluster, service]
  - source_matchers: [severity="warning"]
    target_matchers: [severity="alert", severity="info"]
    equal: [alertname, instance, cluster, service]

receivers:
  - name: "default-receiver"
    slack_configs:
      - channel: "YOUR_SLACK_CHANNEL"
        api_url: "YOUR_SLACK_WEBHOOK_URL"
        title: '[{{ .Status | toUpper }}] Prometheus Alert Notification'
        text: |
          {{ range .Alerts }}
          [{{ .Status | toUpper }}]
          - AlertName: {{ .Labels.alertname }}
            Instance: {{ .Labels.instance }}
            Severity: {{ .Labels.severity | toUpper }}
            Summary: {{ .Annotations.summary }}
            Description: {{ .Annotations.description }}
          {{ end }}
        fallback: '[{{ .Status | toUpper }}] Alerts from Prometheus'
        send_resolved: true


※ YOUR_SLACK_CHANNEL は、Slack内の投稿するチャンネル名に置き換えてください。
※ YOUR_SLACK_WEBHOOK_URL は、Slackアプリ作成時に取得したWebhook URLに置き換えてください。

設定ファイルを変更したら、PrometheusとAlertmanagerを再起動します。

sudo systemctl restart prometheus
sudo systemctl restart prometheus-alertmanager


4. Slackメッセージを解析するPythonスクリプトの作成

Slackから届いたアラートメッセージを解析し、パトライトを制御するPythonスクリプトを作成します。
今回の制御されるパトライトはNHPFB2です。

スクリプトの準備

Pythonの仮想環境を作成し、必要なライブラリをインストールします。

mkdir patlite_controller
cd patlite_controller
python3 -m venv venv
source venv/bin/activate
pip install slack_bolt requests

slack_patlite_app.py というファイルを作成し、以下のコードを貼り付けます。

import os
import logging
import requests
import json
import re
import threading
import time
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler

# ロギング設定
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# 環境変数からSlackトークンとパトライトIPを取得
SLACK_BOT_TOKEN = os.environ.get("SLACK_BOT_TOKEN")
SLACK_APP_TOKEN = os.environ.get("SLACK_APP_TOKEN")
PATLITE_IP_ADDRESS = os.environ.get("PATLITE_IP_ADDRESS")

# 環境変数が設定されているか確認
if not SLACK_BOT_TOKEN:
    logger.error("Error: SLACK_BOT_TOKEN environment variable is not set.")
    exit(1)
if not SLACK_APP_TOKEN:
    logger.error("Error: SLACK_APP_TOKEN environment variable is not set.")
    exit(1)
if not PATLITE_IP_ADDRESS:
    logger.error("Error: PATLITE_IP_ADDRESS environment variable is not set.")
    exit(1)

app = App(token=SLACK_BOT_TOKEN)

# パトライトのベースURL (環境変数から取得したIPアドレスを使用)
PATLITE_BASE_URL = f"http://{PATLITE_IP_ADDRESS}/api/control"

# アクティブなアラートを格納する辞書
active_alerts = {}

# パトライト制御ロック: 複数スレッドからの同時制御を防ぐ
patlite_lock = threading.Lock()

# パトライト制御関数 (GETメソッドに対応)
def control_patlite(alert_code):
    url = f"{PATLITE_BASE_URL}?alert={alert_code}"
    with patlite_lock:
        try:
            logger.info(f"Attempting to control Patlite with code: {alert_code} (URL: {url})")
            response = requests.get(url, timeout=5)
            response.raise_for_status()
            logger.info(f"Patlite control successful. Status: {response.status_code}")
            return True
        except requests.exceptions.RequestException as e:
            logger.error(f"Failed to control Patlite at {url}: {e}")
            return False

# パトライトの状態を更新する関数
def update_patlite_state():
    red_status = '0'
    yellow_status = '0'
    green_status = '0'
    for alert_key, severity in active_alerts.items():
        if severity == "critical":
            red_status = '1'
        elif severity == "warning":
            yellow_status = '1'
        elif severity == "alert":
            green_status = '2'
    patlite_code = f"{red_status}{yellow_status}{green_status}000"
    logger.info(f"Calculated Patlite code based on active alerts: {patlite_code}")
    control_patlite(patlite_code)

@app.event("message")
def handle_message_events(body, logger_slack):
    event = body.get('event', {})
    slack_message_text = ""
    attachments = event.get('attachments', [])
    if attachments and len(attachments) > 0 and 'text' in attachments[0]:
        slack_message_text = attachments[0]['text']

    if not slack_message_text:
        return

    parsed_alerts_in_message = []
    alert_parser_regex = re.compile(
        r'\[(FIRING|RESOLVED)\]\s+'
        r'- AlertName: ([\w\s\.]+)\s+'
        r'Instance: ([^\n]+)\s+'
        r'Severity: (CRITICAL|WARNING|ALERT|INFO)\s+'
        r'Summary: ([^\n]+)\s+'
        r'Description: ([^\n]+)',
        re.MULTILINE
    )

    for match in alert_parser_regex.finditer(slack_message_text):
        alert_status_individual = match.group(1).strip().lower()
        alert_name = match.group(2).strip()
        instance_full = match.group(3).strip()
        severity_str = match.group(4).strip().lower()
        instance_base = instance_full.split(':')[0]
        parsed_alerts_in_message.append({
            'alert_name': alert_name,
            'instance': instance_base,
            'status': alert_status_individual,
            'severity': severity_str
        })

    if parsed_alerts_in_message:
        temp_alert_states = {}
        severity_order = {"critical": 0, "warning": 1, "alert": 2, "info": 3, "resolved": 999}
        for alert_data in parsed_alerts_in_message:
            alert_key = f"{alert_data['alert_name']}-{alert_data['instance']}"
            current_status = alert_data['status']
            current_severity = alert_data['severity']
            existing_effective_severity = temp_alert_states.get(alert_key, None)
            if current_status == "resolved":
                if existing_effective_severity is None or \
                   severity_order.get(existing_effective_severity, 1000) >= severity_order["resolved"]:
                    temp_alert_states[alert_key] = "resolved"
            elif current_status == "firing":
                current_severity_rank = severity_order.get(current_severity, 1000)
                existing_severity_rank = severity_order.get(existing_effective_severity, 1000)
                if existing_effective_severity is None or \
                   current_severity_rank < existing_severity_rank or \
                   existing_effective_severity == "resolved":
                    temp_alert_states[alert_key] = current_severity
            else:
                logger.warning(f"Unknown alert status '{current_status}' for alert {alert_key}")

        for alert_key, final_status_or_severity in temp_alert_states.items():
            logger.info(f"Applying parsed alert to active_alerts: {alert_key}, Status/Severity: {final_status_or_severity}")
            if final_status_or_severity == "resolved":
                if alert_key in active_alerts:
                    del active_alerts[alert_key]
                    logger.info(f"Removed resolved alert: {alert_key}")
                else:
                    logger.warning(f"Resolved alert {alert_key} not found in active list. Current active: {active_alerts.keys()}")
            else:
                active_alerts[alert_key] = final_status_or_severity
                logger.info(f"Added/Updated active alert: {alert_key} -> {final_status_or_severity}")
        logger.info(f"Current active alerts after processing Slack message: {active_alerts}")
        update_patlite_state()
    else:
        logger.warning(f"Could not parse alert information from Slack message.")

if __name__ == "__main__":
    logger.info("Starting Patlite Slack Controller (Socket Mode)...")
    SocketModeHandler(app, SLACK_APP_TOKEN).start()

5. systemdサービスとして起動

このスクリプトをデーモン化し、OS起動時に自動実行されるようにします。

systemdサービスファイルの作成

/etc/systemd/system/patlite-slack-controller.service というファイルを作成します。

sudo vi /etc/systemd/system/patlite-slack-controller.service

以下の内容を貼り付けます。

[Unit]
Description=Patlite Slack Controller
After=network.target

[Service]
ExecStart=/opt/patlite_controller/venv/bin/python3  /opt/patlite_controller/slack_patlite_app.py
WorkingDirectory=/opt/patlite_controller
Restart=always
User=RUN_USER
Environment=SLACK_BOT_TOKEN="xoxb-YOUR_BOT_TOKEN"
Environment=SLACK_APP_TOKEN="xapp-YOUR_APP_TOKEN" Environment=PATLITE_IP_ADDRESS="xxx.xxx.xxx.xxx"

[Install] WantedBy=multi-user.target

ExecStart と WorkingDirectory は、作成したPythonスクリプトのパスに合わせて変更してください。

User はスクリプトを実行するユーザー名に変更してください。

Environment の各トークンとIPアドレスは、取得した情報に置き換えてください。

サービスの有効化と起動

サービスファイルをリロードし、有効化して起動します。

sudo systemctl daemon-reload 
sudo systemctl enable patlite-slack-controller.service
sudo systemctl start patlite-slack-controller.service

ログの確認

以下のコマンドでログを確認し、スクリプトが正常に動作しているか確認します。

sudo journalctl -u patlite-slack-controller.service -f

これで、PrometheusとAlertmanagerでディスク使用率を監視し、Slack経由でパトライトを制御するシステムが完成しました。
ディスク使用率を段階的に変化させて、パトライトの点灯・点滅が正しく変化することを確認してください。