New RelicのアラートからBacklog課題を自動起票してみた ～インシデント管理×AI活用への第一歩～

はじめに

こんにちは、サーバーワークスの福田です。

今回は、New Relicで発生したアラート（インシデント）をBacklogの課題として自動起票してみました。 本記事では、New Relic → AWS EventBridge → Lambda → Backlog APIという構成で、アラート発生時の自動起票と復旧時の自動クローズを実現する方法を解説します。

なぜやろうと思ったか

インシデント管理にAIを活用したいなーと思ったからです。

AIを活用するには、AIが利用するデータソースを確かなものにしないと、正確な回答が返ってこない場合があります。そのため、システムの現状を把握するためのオブザーバビリティの実装や、構成情報のIaC化などが重要になります。

これらに共通しているのは「正しいデータを使用する」ということです。

なぜ情報整備（オブザーバビリティ導入やIaC化）が必要なのか？

オブザーバビリティが整備されていない場合の問題

例えば、AIに「このエラーの根本原因を特定して」と依頼したとします。

オブザーバビリティが未整備の場合

• アプリケーションのログが出力されていない、または必要な情報が含まれていない
• メトリクス（CPU、メモリ、レイテンシなど）が収集されていない
• 分散システムでリクエストの流れを追跡できない（トレースがない）

→ AIに渡せるデータがそもそも存在しない、または不完全なため原因特定が不可能

オブザーバビリティが整備されている場合

• アプリケーションログに十分なコンテキスト（ユーザーID、リクエストIDなど）が含まれている
• メトリクスが時系列で記録され、異常値を検出できる
• 分散トレーシングでリクエストの流れを可視化し、ボトルネックを特定できる

→ 「このエラーは毎週月曜9時のバッチ処理で発生し、DB接続数の枯渇が原因」とAIが分析できる

IaC化されていない場合の問題

AIに「このサーバーの設定を確認して、セキュリティリスクを指摘して」と依頼したとします。

IaC化されていない場合

• 手作業で構築されたため、構成情報がドキュメント化されていない（または古いまま放置）
• AWSコンソールやCLIで直接変更されており、変更履歴が追跡できない
• 同じ「本番環境」でも、担当者によって微妙に設定が異なる

→ AIが参照できる「正しい構成情報」が存在しない

IaC化されている場合

• TerraformやCloudFormationのコードが信頼できる情報源となり、実際のインフラ構成と一致している
• Gitの履歴から「いつ・誰が・なぜ」設定を変更したかを追跡できる

→AIがコードを読むだけで構成を把握し、「このセキュリティグループのインバウンドルールは0.0.0.0/0からのアクセスを許可しています。意図した設定ですか？」と指摘できる

インシデント管理・分析も同様

インシデント管理・分析にAIを活用する場合も同様に、正しい情報（現状）が必要になります。そのためには情報の一元化が重要です。

バラバラな情報管理の例

情報	管理場所
アラート通知	Slack #alerts チャンネル
対応状況	Excelファイル（誰かのローカル）
原因分析	Confluenceの個人ページ
再発防止策	口頭で共有

→ AIに「過去1年の障害から再発防止策の実施状況を分析して」と聞いても、データが散在していて不可能

検証作業としてBacklogを選んだ理由

そこで、アラート情報やアラート対応状況の管理場所としてBacklogを選びました。また、AIの利用もなるべく楽にしたいという思いがあり、Backlog MCPや最近発表されたBacklog AIアシスタントを使えば楽に分析できるのではないかという思惑もあります。

つまり、

• データの一元管理
• 楽にAIを活用したい（運用・保守・開発作業を増やしたくない）

という2つの思いから、今回の仕組みを構築しました。

nulab.com

backlog.com

この仕組みで何ができるのか

Backlogにアラート情報を集約し、課題内で対応内容を管理する運用にすると、以下のようなことが可能になります..なると思います
(まだBacklogMCP等は動作検証していないですが出来ると信じています）

過去の対応履歴をAIで検索・分析

Backlog MCPやAIアシスタントを活用することで、過去にどのような対応をしていたかを簡単に検索・分析できます。

「前にも同じアラートが出たとき、どう対応したっけ...？」

という悩みがなくなり、効率よく対応できるようになると思います。

対応手順の改善・ナレッジ化

• 過去の対応内容から改善点を発見
• 対応手順書の作成・更新をAIがサポート
• ベストプラクティスの蓄積

インシデント傾向の可視化

• 頻発しているアラートの集計
• 対応状況（未対応・対応中・完了）の把握
• 対応時間の分析

つまり、Backlogをインシデント管理ツール（AI分析付き）として活用できるのではないかと考えています。ポストモーテムの実施や改善活動にも役立てられるはずです。

今回の構成

今回構築するシステムの全体像は以下の通りです。

処理の流れ

1. アラート発生（ACTIVATED）: New RelicのWorkflowからEventBridge経由でLambdaが起動し、Backlogに新規課題を作成
2. アラート復旧（CLOSED）: 同様にLambdaが起動し、対応する課題を自動で「完了」ステータスに変更

ポイント: 同じアラートで二重起票されないよう、New RelicのIssue IDをキーにして重複チェックを行っています。

---

事前準備

今回CloudFormationにより設定をしましたのでテンプレートを流す前に、以下情報を確認しました。

1. Backlog APIキーの発行

1. Backlog画面右上のアイコン → 「個人設定」をクリック
2. 左メニュー「API」をクリック
3. 「メモ」欄に「NewRelic連携用」と入力して「登録」
4. 表示されたAPIキーを控えておきます

2. Backlogの各種IDの確認

項目	確認方法
SpaceId	BacklogのURL（例: xxx.backlog.jp の xxx 部分）
Domain	backlog.jp または backlog.com
ProjectId	課題検索画面のURLに含まれる projectId=XXXXX の数字
IssueTypeId	検索条件で種別を選択した際のURLに含まれる issueTypeId=XXXXX
PriorityId	検索条件で優先度を選択した際のURLに含まれる priorityId=X（高:2, 中:3, 低:4）
ClosedStatusId	プロジェクト設定 > 状態 で「完了」ステータスのIDを確認（通常は4）
AssigneeId（任意）	検索条件で担当者を選択した際のURLに含まれる assigneeId=XXXXX
NotifiedUserIds（任意）	通知先ユーザーのID。複数指定する場合はカンマ区切り（例: 12345,67890）

3. New RelicでAWS EventBridge Destinationを設定

1. New Relicの [Alerts & AI] > [Destinations] を開く
2. [New destination] > [AWS EventBridge] を選択
3. AWSアカウントIDとリージョン（例: ap-northeast-1）を入力して保存

4. AWSでパートナーイベントソースを関連付け

Step 1: New Relic側での設定

1. New Relicの [Alerts] > [Destinations] を開く
2. [Add a destination] をクリックし、[AWS EventBridge] を選択
3. 以下の情報を入力:
   • Destination name: 任意の名前（例: Backlog連携用EventBridge）
   • AWS region: イベントを送信するリージョン（例: ap-northeast-1）
   • AWS account ID: AWSアカウントID
   • Event source: 任意の名前を入力、または既存のパートナーイベントソースを選択
4. [Create destination] をクリック

補足: Event sourceに新しい名前を入力し、「Create新規イベントソース名」をクリックするとAWS側に新しいパートナーイベントソースが作成されます。既存のソースを再利用する場合は、ドロップダウンから選択してください。

Step 2: AWS側でのイベントバス関連付け

New Relicで設定を保存すると、AWS側にパートナーイベントソースが作成されます（ステータス: 保留中）。

1. AWSコンソール > [Amazon EventBridge] > [パートナーイベントソース] を開く
2. ステータスが 「保留中 (Pending)」 の aws.partner/newrelic.com/... を選択
3. [イベントバスに関連付ける] をクリック
4. 確認画面で [関連付ける] をクリック
5. ステータスが 「アクティブ (Active)」 に変わったことを確認
6. 関連付けられた イベントバス名（aws.partner/newrelic.com/XXXXX/XXXX など）を控えておく

設定した情報サンプル

私の環境では動作しておりますが動作保証は出来ないのであくまでもサンプル情報として参考にしてください。

Lambda関数のソースコード（クリックで展開）

import json
import os
import time
import boto3
import urllib.request
import urllib.parse
import urllib.error
import logging
from typing import Dict, Any, Optional, List
from datetime import datetime, timedelta, timezone
from botocore.config import Config

logger = logging.getLogger()
logger.setLevel(os.environ.get('LOG_LEVEL', 'INFO'))

# キャッシュ
_cache: Dict[str, Any] = {}
_cache_time: float = 0.0


# =============================================================================
# 設定・ユーティリティ
# =============================================================================

def get_config() -> Dict[str, Any]:
    """Secrets Managerから設定を取得（5分キャッシュ）"""
    global _cache, _cache_time
    
    if _cache and (time.monotonic() - _cache_time) < 300:
        return _cache

    secret_arn = os.environ.get('SECRET_ARN')
    if not secret_arn:
        raise ValueError("SECRET_ARN is not set")

    client = boto3.client('secretsmanager', config=Config(
        retries={'max_attempts': 3, 'mode': 'adaptive'},
        connect_timeout=5, read_timeout=10
    ))
    
    config = json.loads(client.get_secret_value(SecretId=secret_arn)['SecretString'])
    
    # 通知ユーザーIDをリストに変換
    raw = config.get('notified_user_ids', '')
    config['notified_user_ids_list'] = [
        int(uid.strip()) for uid in str(raw).split(',') 
        if uid.strip() and uid.strip().isdigit()
    ]
    
    _cache, _cache_time = config, time.monotonic()
    logger.info("Config loaded")
    return config


def to_jst(timestamp_ms: Optional[int]) -> str:
    """ミリ秒タイムスタンプをJST文字列に変換"""
    if not timestamp_ms:
        return "-"
    try:
        jst = timezone(timedelta(hours=9))
        return datetime.fromtimestamp(timestamp_ms / 1000.0, tz=timezone.utc).astimezone(jst).strftime('%Y-%m-%d %H:%M:%S')
    except (ValueError, OSError, OverflowError):
        return "-"


def sanitize(text: str) -> str:
    """4バイト文字（絵文字等）を除去"""
    return ''.join(c for c in (text or '') if len(c.encode('utf-8')) <= 3)


def get_alert_name(data: Dict[str, Any]) -> str:
    """アラート名を取得"""
    names = data.get('alertConditionNames')
    if isinstance(names, list) and names:
        return ", ".join(str(n) for n in names if n)
    if names:
        return str(names)
    return data.get('title', '不明なアラート')


# =============================================================================
# Backlog API
# =============================================================================

def call_api(method: str, endpoint: str, params: Dict[str, Any], config: Dict[str, Any]) -> Optional[Dict]:
    """Backlog APIを呼び出す（リトライ付き）"""
    base = f"https://{config['space_id']}.{config['domain']}/api/v2"
    api_key = {'apiKey': config['api_key']}
    
    if method == 'GET':
        url = f"{base}{endpoint}?{urllib.parse.urlencode({**api_key, **params}, doseq=True)}"
        data = None
    else:
        url = f"{base}{endpoint}?{urllib.parse.urlencode(api_key)}"
        data = urllib.parse.urlencode(params, doseq=True).encode('utf-8')

    for attempt in range(3):
        try:
            req = urllib.request.Request(url, data=data, method=method)
            req.add_header('Content-Type', 'application/x-www-form-urlencoded')
            with urllib.request.urlopen(req, timeout=30) as resp:
                return json.loads(resp.read().decode('utf-8'))
        except urllib.error.HTTPError as e:
            if e.code == 429:
                time.sleep(min(2 ** attempt, 8))
                continue
            logger.error(f"API error: {e.code} on {endpoint}")
            return None
        except Exception:
            logger.error(f"Error on {endpoint}")
            return None
    
    logger.error(f"Max retries: {endpoint}")
    return None


def find_issue(identifier: str, config: Dict[str, Any]) -> Optional[Dict]:
    """識別子で既存課題を検索"""
    logger.info(f"Searching: {identifier}")
    results = call_api('GET', '/issues', {
        'projectId[]': config['project_id'],
        'keyword': identifier,
        'count': 100, 'sort': 'created', 'order': 'desc'
    }, config)
    
    if not results:
        return None
    
    for issue in results:
        if identifier in issue.get('description', ''):
            logger.info(f"Found: {issue.get('issueKey')}")
            return issue
    return None


def add_optional_params(params: Dict[str, Any], config: Dict[str, Any]) -> Dict[str, Any]:
    """担当者・通知先を追加"""
    if config.get('assignee_id'):
        params['assigneeId'] = config['assignee_id']
    if config.get('notified_user_ids_list'):
        params['notifiedUserId[]'] = config['notified_user_ids_list']
    return params


# =============================================================================
# 課題処理
# =============================================================================

def build_description(data: Dict[str, Any], identifier: str, issue_id: str) -> str:
    """課題の説明文を生成"""
    lines = [
        f"識別子: {identifier}", "",
        "| 項目 | 内容 |", "|---|---|",
        f"| **コンディション名** | [{sanitize(get_alert_name(data))}]({data.get('issueUrl', '#')}) |",
        f"| **ステータス** | {data.get('state', 'UNKNOWN')} |",
        f"| **優先度** | {data.get('priority', '-')} |",
        f"| **Issue ID** | {issue_id} |",
    ]
    
    if 'createdAt' in data:
        lines.append(f"| **発生日時** | {to_jst(data.get('createdAt'))} |")
    if 'updatedAt' in data:
        lines.append(f"| **更新日時** | {to_jst(data.get('updatedAt'))} |")
    lines.append("")
    
    for label, key in [("対象エンティティ", "impactedEntities"), ("アラートポリシー", "alertPolicyNames"), ("ワークフロー", "workflowName")]:
        if data.get(key):
            lines.append(f"**{label}:** `{sanitize(str(data[key]))}`" if key == "impactedEntities" else f"**{label}:** {sanitize(str(data[key]))}")
            lines.append("")
    
    lines.extend([
        "---", "<details><summary>生データ(JSON)</summary>", "",
        "```json", json.dumps(data, ensure_ascii=False, indent=2), "```", "", "</details>"
    ])
    return "\n".join(lines)


def create_issue(data: Dict[str, Any], config: Dict[str, Any], identifier: str, issue_id: str) -> Dict[str, str]:
    """新規課題を作成"""
    existing = find_issue(identifier, config)
    if existing:
        key = existing.get('issueKey', 'unknown')
        logger.info(f"Already exists: {key}")
        return {'action': 'skipped', 'reason': 'duplicate', 'issueKey': key}

    params = add_optional_params({
        'projectId': config['project_id'],
        'summary': sanitize(f"[New Relic] {get_alert_name(data)}"),
        'description': build_description(data, identifier, issue_id),
        'issueTypeId': config['issue_type_id'],
        'priorityId': config['priority_id']
    }, config)

    result = call_api('POST', '/issues', params, config)
    if result:
        key = result.get('issueKey', 'unknown')
        logger.info(f"Created: {key}")
        return {'action': 'created', 'issueKey': key}
    
    logger.error("Failed to create")
    return {'action': 'error', 'reason': 'api_failed'}


def close_issue(data: Dict[str, Any], config: Dict[str, Any], identifier: str) -> Dict[str, str]:
    """課題をクローズ"""
    target = find_issue(identifier, config)
    if not target:
        logger.warning(f"Not found: {identifier}")
        return {'action': 'skipped', 'reason': 'not_found'}

    key = target.get('issueKey', 'unknown')
    
    if str(target.get('status', {}).get('id')) == str(config.get('closed_status_id')):
        logger.info(f"Already closed: {key}")
        return {'action': 'skipped', 'reason': 'already_closed', 'issueKey': key}

    logger.info(f"Closing: {key}")
    
    params = add_optional_params({
        'statusId': config['closed_status_id'],
        'comment': f"【復旧】New Relic側で復旧(Closed)しました\n\n復旧日時: {to_jst(data.get('updatedAt'))}\nこの課題を「完了」ステータスに変更します。"
    }, config)

    result = call_api('PATCH', f"/issues/{key}", params, config)
    if result:
        logger.info(f"Closed: {key}")
        return {'action': 'closed', 'issueKey': key}
    
    logger.error(f"Failed to close: {key}")
    return {'action': 'error', 'reason': 'api_failed', 'issueKey': key}


# =============================================================================
# メインハンドラー
# =============================================================================

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """Lambda メインハンドラー"""
    detail = event.get('detail', {})
    state = detail.get('state')
    issue_id = detail.get('issueId') or detail.get('id')
    
    logger.info(f"Event: state={state}, id={issue_id}")
    
    if not issue_id:
        logger.error("Missing issueId")
        return {'statusCode': 400, 'body': json.dumps({'error': 'Missing issueId'})}

    try:
        config = get_config()
        identifier = f"NR-{issue_id}"
        
        handlers = {'ACTIVATED': create_issue, 'CLOSED': close_issue}
        handler = handlers.get(state)
        
        if handler:
            result = handler(detail, config, identifier, issue_id) if state == 'ACTIVATED' else handler(detail, config, identifier)
        else:
            logger.info(f"Ignored: {state}")
            result = {'action': 'ignored', 'reason': f'unhandled_state: {state}'}

        return {'statusCode': 200, 'body': json.dumps({'status': 'done', 'result': result})}

    except ValueError as e:
        logger.error(f"Config error: {e}")
        return {'statusCode': 500, 'body': json.dumps({'error': 'Configuration error'})}
    except Exception:
        logger.exception("Unexpected error")
        return {'statusCode': 500, 'body': json.dumps({'error': 'Internal error'})}

Secrets Managerの設定内容（クリックで展開）

キー	説明	必須
api_key	Backlog APIキー	○
domain	backlog.jp または backlog.com	○
space_id	スペースID	○
project_id	プロジェクトID（数値）	○
issue_type_id	課題種別ID（数値）	○
priority_id	優先度ID（数値: 高=2, 中=3, 低=4）	○
closed_status_id	完了ステータスID（数値、通常は4）	○
assignee_id	担当者ID（数値）	-
notified_user_ids	通知先ユーザーID（カンマ区切り）	-

---

New Relic Workflowの設定

1. Workflowの作成

1. New Relic > [Alerts & AI] > [Workflows] > [Add a workflow]
2. Name: 任意の名前（例: Backlog連携用）
3. Filter data: 通知したいアラート条件を指定
   • 例: Issue priority is High（高優先度のみ通知）
4. Notify: 作成済みの EventBridge Destination を選択
5. Notification Triggers:
   • Activated: オン（これがないと起票されません）
   • Closed: オン（これがないと自動完了されません）

2. Payloadのカスタマイズ（任意）

初期設定でも問題ないと思います。 参考までに実際に設定した内容は以下になります。

{
    "id": {{ json issueId }},
    "issueUrl": {{ json issuePageUrl }},
    "title": {{ json annotations.title.[0] }},
    "priority": {{ json priority }},
    "impactedEntities": {{ json entitiesData.names }},
    "totalIncidents": {{ json totalIncidents }},
    "state": {{ json state }},
    "trigger": {{ json triggerEvent }},
    "isCorrelated": {{ json isCorrelated }},
    "createdAt": {{ createdAt }},
    "updatedAt": {{ updatedAt }},
    "sources": {{ json accumulations.source }},
    "alertPolicyNames": {{ json accumulations.policyName }},
    "alertConditionNames": {{ json accumulations.conditionName }},
    "workflowName": {{ json workflowName }}
}

---

動作確認

1. テスト通知の送信

New RelicのWorkflow編集画面で [Send test notification] をクリックします。

2. Backlogの確認

課題一覧に「[New Relic]...」というタイトルの課題が作成されていれば成功です。

3. ログの確認

起票されない場合は、AWS Lambda > [モニタリング] > [CloudWatch ログを表示] でエラー内容を確認してください。

実際に起票された課題

New Relic側でクローズ状態になるとBacklogのコメントにクローズする旨の記載とステータスが完了に自動変更される

---

ハマったポイント

Issue already exists と出て起票されない

同じNew Relic Issue IDで何度もテストすると、重複チェックにより起票がスキップされます。処理としては問題ないですが動作確認の際にテストする際に詰まりました。 新しい課題として起票したい場合は、New RelicのPayloadでIDを一時的に書き換えてテストしてください。

• "id": {{ json issueId }}ではなく"id": "テスト用ID"に変更するなど

復旧時に課題がクローズされない

• Workflowの Notification Triggers で Closed がオンになっているか確認してください

懸念点・注意事項

Backlog APIのレート制限（Rate Limit）について

Backlog API は各ユーザーに対して、1分間に受付可能なリクエスト数を制限しております。 現在の上限は レート制限情報の取得API で確認できます。

今回の仕組みでは、1回のアラートで最大2回のAPIコール（検索1回 + 作成or更新1回）が発生します。通常の運用では問題になることはほぼありませんが、頻発するアラート（フラッピングなど）が懸念される場合は、New Relic側でミューティングルールやアラート閾値の調整を検討してください。

既存課題が大量にある場合の課題紐づけについて

Backlogの課題一覧取得APIは1回のリクエストで最大100件までしか結果を返しません（課題一覧の取得 ）。
ただし、以下の理由から、既存課題が大量に存在する場合でも復旧通知が元の課題と紐づかないことはない想定です。

1. 識別子は一意: 識別子はNew Relicのアラートごとに固有
2. 検索結果は0件か1件: 同じ識別子を持つ課題は1件しか存在しないため、キーワード検索で該当課題がヒットすれば必ず見つかる
3. 説明文で完全一致チェック: 検索結果から識別子が完全に含まれる課題のみを対象にするため、誤検出を防止

運用してみないとわからない

実際にアラートが発生してみないと使用感や挙動がわからないというのが正直なところです。

• 頻発アラート発生時に処理が追いつくか
• 通常のアラート通知で漏れが発生しないか
• Backlog課題としての管理のしやすさ

これらは実運用を通じて検証し、必要に応じてLambdaの改修やNew Relic側のアラート設定調整するしかないなと思います。

まとめ

今回の内容をまとめます。

• New Relic → EventBridge → Lambda → Backlog API の構成で自動起票を実現
• アラート発生時に自動起票、復旧時に自動クローズ
• 重複起票を防ぐためIssue IDで既存課題をチェック
• Secrets Managerで設定情報（APIキー等）をセキュアに管理
• 担当者・通知先ユーザーの自動設定にも対応

AIは「与えられたデータ」をベースに回答します。 AIを効率よく利用するためにには、まず正しいデータを正しい場所に蓄積する仕組みを作ることが第一歩です。

今回の仕組みにより今後のAI活用の土台作りだけでなく手動での起票作業がなくなり、対応漏れのリスクを減らすことができます。監視運用の効率化に参考にしていただければと思います。 私も今回使用したLambdaを見直したり、BacklogMCP等を試したいと思っています。

宣伝

弊社では、お客様環境のオブザーバビリティを加速するためのNew Relicアカウント開設を含めた伴走型のNew Relic導入支援サービスをご提供しております。もしご興味をお持ちの方は、こちらのNew Relic導入支援サービスのページよりお問合せ頂けましたら幸いでございます。

・福田 圭(記事一覧)

・マネージドサービス部　所属
・X（Twitter）：@soundsoon25

2023 New Relic Partner Trailblazer。New Relic Trailblazer of the Year 2025受賞。New Relic User Group運営。

クラウド活用の「ちょっとした不安」や
「もっと良くしたい」に応えます

サーバーワークスは、AWSの構築から運用まで
幅広く支援しています

資料ダウンロード相談する

#New Relic #Backlog #BacklogApi

2026-02-04 21:46