1. はじめに

皆さんこんにちは。

最近DQVをクリアしたのですが、イオナズンを覚えるからって理由で嫁に「フローラ」を選んだ男、佐藤晃樹です。

それはさておき、皆さんどんなサービスを使っていても、「この機能が実現できるリソースやアプリがないなぁ」と思う事がありませんか？

今回は「これできないのかぁ」の様な壁に業務中にあたり、こんな感じで回避したよっていうのを記事にしていきます。

2. 業務中にぶつかった壁

- 余裕の監視設計（のはずだった）

とあるサイトの監視設計を担当することになりました。

環境：

クラウドサービス: Azure
監視サービス: Grafana

最初は簡単だと思っていました。

「データソース追加してダッシュボードとアラート設定して視覚化したら終わりかな」と・・・

- 要件定義で見つけた爆弾

可用性テスト周りの要件定義を読み進めていると...

要件：「監視サービスを用い、1分間隔で外形監視を行う事」

ここで嫌な予感がしました。

「Azureで外形監視といえばApplication Insightsだけど、最低5分間隔じゃなかったっけ...？」

- ドキュメント確認で絶望

公式ドキュメントを確認すると、微妙な記載を発見：

「既定の間隔が 5 分で、テストの場所が 5 か所の場合、サイトは平均して毎分テストされます」

　"平均して"毎分テスト…

これでは1分間隔での外形監視という要件を満たしているとは言い切れません。

- 発想の転換

じゃあどうしよう...となったときに、ふと思いました。

「ないなら作ればいいんじゃね？」

外形監視なんて、結局はURLにアクセスしてレスポンスを取るだけ。

AIを使ってコーディングすれば、すぐに実装できそうです。

これが今回の経緯にあたります。

3. AIに死活監視を行うPythonコードを書かせる

実際に生成AIに以下の要件でプロンプトを作成し、コードを生成させました。

使用するLLM：Claude 4.1 Opus

作成してほしいもの：定期的にURLの監視を行い、レスポンス結果をAzure SQL Databaseに記録するAzure Functionsアプリ

要件：

Timer Triggerで定期実行（スケジュールは環境変数から）
複数URLの監視（環境変数MONITORING_URLSからカンマ区切りで取得）
各URLに対してHTTPリクエスト送信
- タイムアウト: 接続10秒、読み取り30秒
- リトライ: 5xx系エラーで3回まで
監視結果をAzure SQL Databaseに保存
- pyodbc使用
- テーブル名: UrlMonitoring
- カラム: MonitoringTime, TargetUrl, StatusCode, ResponseTime, IsSuccess, ErrorMessage
2xx系ステータスコードを成功として扱う
エラーハンドリング、ログ出力も含めて実装

といった感じでプロンプトを作成して、AIに投げてみました。

出力されたコード（土台）の状態から修正を加えて出来たコードがこちらです。

Pythonimport os
import datetime
import logging
import requests
from urllib3.util import Retry
from requests.adapters import HTTPAdapter
import azure.functions as func
import pyodbc

# 環境変数から接続文字列を取得
sql_connection_string = os.getenv("SQL_CONNECTION_STRING")
schedulesetting = os.getenv("ScheduleAppSetting", "0 * * * * *")  # デフォルト値を設定

# 環境変数から監視対象URLを取得
def get_monitoring_urls():
    """環境変数から監視対象URLのリストを取得"""
    urls_string = os.getenv("MONITORING_URLS", "")
    if not urls_string:
        logging.warning("MONITORING_URLS not found in environment variables")
        return []
    
    # カンマ区切りのURLを分割してリストに変換
    urls = [url.strip() for url in urls_string.split(",") if url.strip()]
    logging.info(f"Monitoring URLs: {urls}")
    return urls

app = func.FunctionApp()

def save_to_sql_database(url, status_code, response_time, error_message=""):
    """監視結果をSQL Databaseに保存"""
    if not sql_connection_string:
        logging.error("SQL_CONNECTION_STRING is not set")
        return

    try:
        # SQL Databaseに接続
        conn = pyodbc.connect(sql_connection_string)
        cursor = conn.cursor()
        
        # データを挿入
        insert_query = """
        INSERT INTO UrlMonitoring 
        (MonitoringTime, TargetUrl, StatusCode, ResponseTime, IsSuccess, ErrorMessage)
        VALUES (?, ?, ?, ?, ?, ?)
        """
        
        # 現在時刻を取得
        monitoring_time = datetime.datetime.now(datetime.timezone.utc)
        is_success = 1 if 200 <= status_code < 300 else 0
        
        cursor.execute(insert_query, 
                    monitoring_time, 
                    url, 
                    status_code, 
                    response_time, 
                    is_success, 
                    error_message if error_message else None)
        
        conn.commit()
        cursor.close()
        conn.close()
        
        logging.info(f"Saved to SQL: {url} - Status: {status_code}, Response: {response_time}ms")
        
    except Exception as e:
        logging.error(f"Failed to save to SQL Database: {e}")

@app.timer_trigger(schedule=schedulesetting, arg_name="myTimer", run_on_startup=False,
            use_monitor=False) 
def main(myTimer: func.TimerRequest) -> None:
    logging.info('URL monitoring function started')
    
    # 監視対象URLを取得
    targets = get_monitoring_urls()
    
    if not targets:
        logging.error("No URLs to monitor. Please set MONITORING_URLS in local.settings.json")
        return

    # HTTPセッションの設定
    session = requests.Session()
    retries = Retry(total=3,
                    backoff_factor=1,
                    status_forcelist=[500, 502, 503, 504])
    session.mount("http://", HTTPAdapter(max_retries=retries))
    session.mount("https://", HTTPAdapter(max_retries=retries))

    # 各URLを監視
    for url in targets:
        status_code = 0
        response_time = 0
        error_message = ''
        
        try:
            # レスポンス時間を計測
            start_time = datetime.datetime.now()
            response = session.get(url, timeout=(10.0, 30.0))
            end_time = datetime.datetime.now()
            
            status_code = response.status_code
            response_time = int((end_time - start_time).total_seconds() * 1000)  # ミリ秒
            
        except requests.exceptions.ReadTimeout:
            error_message = 'Timeout'
        except requests.exceptions.ConnectionError:
            error_message = 'Connection error'
        except Exception as e:
            error_message = str(e)
        
        # 結果をログ出力
        if status_code > 0 and status_code < 400:
            logging.info(f'[OK]: {url} - Status: {status_code}, Response: {response_time}ms')
        else:
            logging.error(f'[ERR]: {url} - Status: {status_code}, Error: {error_message}')
        
        # SQL Databaseに保存
        save_to_sql_database(url, status_code, response_time, error_message)

このコードができるまで約1時間でした（個人的にはめちゃくちゃ早いと思いました）

軽くコーディングする際は、生成AIを使うことで要件を洗い出せたりコードの土台がすぐ出来るなど、

一発で完成までは持ってけずとも、すごく時短になりました。

4. 実際に動かして監視してみた

コードをローカルで動かしてみる

SQLDB立てる
- まずは適当なリソースグループを作成する
- リソースグループにSQLDBを作成する（設定等は割愛）
- 作成したSQLDBでレスポンス結果の格納用テーブルを作成する
ローカルで実行するために接続文字列の取得や、監視するURLやスケジュール等の設定（Credentialな情報を含むため割愛）
実際にテストする
- VSCodeでコードを開き、Azure Functions プロジェクトを作成する
- azuriteを実行
- F5を押してデバッグを開始（今回は毎分実行するようにしました）
SQL DBにレコードが追加されていることを確認する