前回まで
この記事では、Lambda × Jira API:最小の実装でチケットを自動作成する
―― Secrets Manager → urllib3 → Jira Issue Create
この記事で書くこと(第3回)
- LambdaでSecrets Managerを読み込む方法がわかる
- urllib3を使ってJira REST APIを呼び出せる
- 最小チケット作成でJiraに起票できる
- CloudWatchやStep Functions連携前に「コア部分」を理解
1.まずは最小のLambdaコードを用意する
今回は「文字列だけを送ってJiraにチケット作る」だけに絞る
後続の記事でアラーム解析・重複防止・コメント追加など入れていきます。
Jiraに1件だけ課題を作れることを確認するためのコードを書いていきます。
- Secrets ManagerからJira接続情報を読む
- Basic認証ヘッダを作る
- descriptionを最小限のADF(1段落テキスト)変換
- Jiraの
/rest/api/3/issueにPOST
lambda_function.py(最小動作版)
import json
import os
import base64
import logging
import urllib3
import boto3
# ===== ロガー設定 =====
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# ===== HTTP クライアント(Jira API 呼び出し用)=====
http = urllib3.PoolManager()
# ===== Secrets Manager から読み込むシークレット名 =====
JIRA_SECRET_NAME = os.getenv("JIRA_SECRET_NAME", "jira/dev")
_secrets_client = boto3.client("secretsmanager")
_secret_cache = None
# -------------------------------------------------------------------
# 1. Secrets Manager から Jira 設定(URL / Email / Token / ProjectKey)を取得
# -------------------------------------------------------------------
def load_jira_secret() -> dict:
"""
Lambda コンテナ再利用時の負荷を下げるため、
Secrets Manager の取得結果は _secret_cache にキャッシュする。
"""
global _secret_cache
if _secret_cache is not None:
return _secret_cache
# SecretString には JSON が格納されている想定
res = _secrets_client.get_secret_value(SecretId=JIRA_SECRET_NAME)
secret = json.loads(res["SecretString"])
_secret_cache = secret
return secret
# -------------------------------------------------------------------
# 2. Jira Basic 認証ヘッダを作成
# -------------------------------------------------------------------
def jira_auth_header(secret: dict) -> dict:
"""
Jira Cloud は Basic 認証で Email + API Token を base64 化して送る。
"""
token = f"{secret['JIRA_EMAIL']}:{secret['JIRA_API_TOKEN']}"
b64 = base64.b64encode(token.encode()).decode()
return {
"Authorization": f"Basic {b64}",
"Content-Type": "application/json",
}
# -------------------------------------------------------------------
# 3. description 用の「最小 ADF(1段落テキスト)」を生成
# -------------------------------------------------------------------
def to_adf(text: str) -> dict:
"""
Jira Cloud の description はプレーンテキスト不可。
最低限「paragraph」を含む ADF でラップする必要がある。
"""
return {
"type": "doc",
"version": 1,
"content": [
{
"type": "paragraph",
"content": [{"type": "text", "text": text}],
}
],
}
# -------------------------------------------------------------------
# 4. Jira 課題作成(最小限)
# -------------------------------------------------------------------
def create_issue(summary: str, description: str) -> str:
secret = load_jira_secret()
base_url = secret["JIRA_BASE_URL"].rstrip("/")
project_key = secret["JIRA_PROJECT_KEY"]
# Issue Type:Incident を基本とし、Secret の値で上書き可能にする
issuetype = {"name": secret.get("JIRA_ISSUE_TYPE", "Incident")}
payload = {
"fields": {
"project": {"key": project_key},
"summary": summary,
"description": to_adf(description), # ← ★ 最小 ADF を適用
"issuetype": issuetype,
}
}
url = f"{base_url}/rest/api/3/issue"
headers = jira_auth_header(secret)
resp = http.request(
"POST",
url,
headers=headers,
body=json.dumps(payload).encode(),
)
if resp.status >= 300:
# エラーが起きたらそのまま返し、デバッグしやすくする
raise Exception(f"Jira error {resp.status}: {resp.data}")
data = json.loads(resp.data.decode())
return data["key"]
# -------------------------------------------------------------------
# 5. Lambda メイン処理:テスト用 Incident を1件作成するだけ
# -------------------------------------------------------------------
def lambda_handler(event, context):
summary = "Test Incident from Lambda"
description = "This is a simple test issue created from my Lambda."
key = create_issue(summary, description)
logger.info(f"Created Jira issue: {key}")
return {"issue_key": key}
1-2.
なぜADFが必要なのか?
Jira Cloudのdescriptionフィールドはプレーンテキストでは400エラーになる
以下のようにテスト用に最小構成でADFを使ってる。
{
"type": "doc",
"version": 1,
"content": [
{
"type": "paragraph",
"content": [{"type": "text", "text": "ここに説明文"}]
}
]
}
1-2.
エラーハンドリングは今は基本無視
Lambda → Jiraの接続が正しく動くことを証明すること
1-3. urllib3 を使った HTTP リクエスト基礎
今回のLambdaではrequestsではなくurllib3を利用
http = urllib3.PoolManager()
resp = http.request(
"POST",
url,
headers=headers,
body=json.dumps(payload).encode(),
)
-
PoolManager()- HTTP接続のプールを管理するクライアント
- Lambdaの「コンテナ再利用」と相性がよい。毎回コネクションを張り直さなくていい
-
http.request(method, url, ...)-
methodはGET,POSTなどの文字列 -
headersにContent-TypeやAuthorizationを載せる -
bodyはbytesなので、json.dumps(...).encode()で渡す
-
-
resp.status/resp.data-
resp.status:HTTPステータスコード(201,400,401など) -
resp.data:レスポンスボディ →resp.data.decode()で文字列にする
-
1-4. Secrets Managerの読み取り
JiraのURLやAPI Tokenをコードにべた書きすると危険なので、
AWS Secrets ManagerにJSONで保存しておく。
Lambdaからは、次のような関数で読み取る
-
JIRA_SECRET_NAMEは環境変数で切り替え(jira/dev, jira/stg, jira/prod など) - Lambdaが起動するたびにSecrets Managerに行くとコスト、レイテンシが無駄なので
最初の1回だけ取得して_secret_cacheにキャッシュしている。
JIRA_SECRET_NAME = os.getenv("JIRA_SECRET_NAME", "jira/dev")
_secrets_client = boto3.client("secretsmanager")
_secret_cache = None
def load_jira_secret() -> dict:
global _secret_cache
# Lambda コンテナ再利用時に Secrets 取得を1回で済ませるためのキャッシュ
if _secret_cache is not None:
return _secret_cache
res = _secrets_client.get_secret_value(SecretId=JIRA_SECRET_NAME)
secret = json.loads(res["SecretString"])
_secret_cache = secret
return secret
2. Lambdaを実行してJiraを確認してみる
Lambdaをデプロイしたら
- 「テスト」を作成 → 空のJSONを渡す
- 実行 → Jiraのプロジェクトにチケットができていれば成功!
3. なぜ最小コードから始めるのか?
➀ 問題の切り分けが容易
一気にCloudWatchやSNSメッセージ解析のロジックを入れてしまうと
問題の特定が複雑化してしまうため、まずはJiraの呼び出しに集中して対応が吉
➁ Jira API x Secrets x Lmabdaの連携をクリアに理解
- Secrets Manager → JSONロード
- urllib3のPOST
- Basic認証
4. まとめ
第3回では最小のJira起票のLambdaを動かしました。
| 学んだ内容 | 一言まとめ |
|---|---|
| Secrets Manager の読み取り | JSON で Jira 認証情報を取得 |
| urllib3 で Jira REST API へ POST | Lambda から HTTPS を叩ける |
| 最小の Issue Create | 実際に Jira にチケットが作られる |
次回(第4回):SNS → CloudWatchアラームをパースしてJira化
- CloudWatch AlarmのSNSメッセージ構造
- Lambda内で「AlarmName」「Reason」などを抽出
- Jira summary / descriptionに整形
- ADFで(見出し/箇条書き/コードブロック)で読みやすく
- Step Functionsも同じ形式に正規化
- 優先度(High/Medium)を自動判定


