PostgreSQLのapplication_nameを4段階フォールバックで解決する：env→launch_label→script.{pid}→fallback

PostgreSQL の pg_stat_activity を見ると、複数のジョブ・スクリプト・REPL が同じユーザー名・同じホストから接続していて、どのプロセスが何をしているか区別できない問題に何度かぶつかりました。

application_name 接続パラメータをきちんと設定すれば識別できるのですが、「環境ごとに何を入れるか」の設計が意外に難しい。最終的に 4 段階のフォールバック に落ち着いた経緯を書きます。

何が困るのか

pg_stat_activity を見ても誰だかわからない

1
SELECT pid, usename, application_name, state, query_start
2
  FROM pg_stat_activity
3
 WHERE state != 'idle';

1
 pid  | usename | application_name |  state  |     query_start
2
------+---------+------------------+---------+---------------------
3
 1234 | app     | psql             | active  | 2026-04-30 10:15:00
4
 1567 | app     |                  | active  | 2026-04-30 10:15:30
5
 1890 | app     |                  | idle... | 2026-04-30 10:14:00

application_name が空 or psql。遅いクエリを誰が投げたかわからず、調査時に困ります。

何を入れたいか

最低限ほしい情報：

どのジョブか（launchd label / スクリプト名）
環境（dev / staging / prod）
インスタンス識別（pid や hostname）

これを application_name に詰めれば、pg_stat_activity を見るだけで犯人が特定できます。

4 段階のフォールバック設計

接続コードがすべての環境で動くよう、以下の順で解決します。

1
Priority 1: APP_NAME 環境変数が明示的にセットされている
2
Priority 2: launchd の LAUNCH_LABEL から推測
3
Priority 3: __main__ モジュールから "script.{pid}" を生成
4
Priority 4: ハードコードされたフォールバック "unknown_app.{pid}"

実装

1
import os
2
import sys
3
from pathlib import Path
4

5
def resolve_application_name() -> str:
6
    """4 段階フォールバックで application_name を決定する"""
7

8
    # Priority 1: 明示的な環境変数
9
    if name := os.getenv("APP_NAME"):
10
        return name
11

12
    # Priority 2: launchd 経由なら label から
13
    if label := os.getenv("LAUNCH_LABEL"):
14
        # com.example.daily_pipeline → daily_pipeline.1234
15
        short = label.split(".")[-1]
10 collapsed lines
16
        return f"{short}.{os.getpid()}"
17

18
    # Priority 3: __main__ から推測
19
    main_module = sys.modules.get("__main__")
20
    if main_module and hasattr(main_module, "__file__"):
21
        script = Path(main_module.__file__).stem
22
        return f"{script}.{os.getpid()}"
23

24
    # Priority 4: 最終フォールバック
25
    return f"unknown_app.{os.getpid()}"

各フォールバックがカバーする状況

Priority	想定環境	例
1. APP_NAME	テスト・特殊実行	`APP_NAME=migration_2026_04 python ...`
2. LAUNCH_LABEL	launchd ジョブ	plist で `LAUNCH_LABEL=com.example.daily`
3. main	python -m / 直接実行	`python -m pipeline.runner`
4. fallback	REPL / Jupyter	`unknown_app.12345`

実装の落とし穴

罠 1: main が None のケース

pytest 実行中などは sys.modules["__main__"] が pytest 自身を指したり、__file__ 属性を持たなかったりします。

1
main_module = sys.modules.get("__main__")
2
if main_module and hasattr(main_module, "__file__"):
3
    script = Path(main_module.__file__).stem

hasattr ガードを必ず入れる。さもないと AttributeError で接続前に死にます。接続コードで例外を投げると本体ロジックに入る前に落ちるので、フォールバックは絶対に死なせません。

罠 2: launchd の LAUNCH_LABEL を自分で渡す必要がある

launchd は標準で LAUNCH_LABEL を環境変数に入れてくれません。plist 側で明示する必要があります。

1
<key>EnvironmentVariables</key>
2
<dict>
3
    <key>LAUNCH_LABEL</key>
4
    <string>com.example.daily_pipeline</string>
5
</dict>

これを忘れると Priority 2 が空振りして Priority 3 にフォールバックします。動きはしますが「launchd 由来か CLI 由来か」が区別できなくなります。

罠 3: application_name の文字数制限

PostgreSQL の application_name は デフォルト 64 文字制限 です（NAMEDATALEN - 1）。

長い launchd label を入れるとサイレントに切り詰められます。

1
def resolve_application_name() -> str:
2
    raw = _resolve_raw()
3
    return raw[:63]  # 念のため明示的に切る

切り詰めても pid が末尾に残るように、短い識別子 + pid の順序にします。

罠 4: pid が再利用される

pid を入れても、長期間動いているプロセスでは pid が他のプロセスから再利用されることがあります。完全な一意性は望めません。

「いま pg_stat_activity を見たときに識別できる」目的では十分ですが、過去ログに突き合わせる用途には使えません。長期トレース用には別途 session_id を持つ必要があります。

接続コード側に組み込む

psycopg / SQLAlchemy 等で application_name を渡します。

1
from sqlalchemy import create_engine
2

3
app_name = resolve_application_name()
4
engine = create_engine(
5
    "postgresql://app@db.example.com/mydb",
6
    connect_args={"application_name": app_name},
7
)

connection pool を使う場合、接続が再利用されても application_name は引き継がれるので、起動時に決めれば十分です。

マルチプロセスの注意

multiprocessing で worker を fork する場合、parent の application_name がそのまま継承されます。worker ごとに pid が違うので、fork 後に application_name を作り直す必要があります。

1
def worker_init():
2
    """multiprocessing worker 起動時に呼ばれる"""
3
    app_name = resolve_application_name()  # ここで子プロセスの pid が拾われる
4
    # 既存の engine を捨てて作り直す
5
    global engine
6
    engine.dispose()
7
    engine = create_engine(..., connect_args={"application_name": app_name})

fork 後の DB 接続は親から引き継がない（各 worker で作り直す）のが原則ですが、application_name もその一環として作り直します。

効果：pg_stat_activity が一気に読みやすくなる

1
SELECT application_name, count(*), max(state_change)
2
  FROM pg_stat_activity
3
 GROUP BY application_name
4
 ORDER BY 2 DESC;

1
       application_name        | count |     max
2
-------------------------------+-------+-----------------------
3
 daily_pipeline.1234           |     5 | 2026-04-30 10:15:00
4
 hourly_ingestion.5678         |     3 | 2026-04-30 10:14:30
5
 migration_2026_04.9012        |     1 | 2026-04-30 10:13:00
6
 unknown_app.3456              |     1 | 2026-04-30 10:12:00

unknown_app.* が出ていたら「これはどこから来た接続だ？」という調査の起点になります。fallback の存在自体が**「設計漏れの早期発見**」につながりました。

監視への展開

pg_stat_activity から application_name 別の長時間クエリを集計できます。

1
SELECT application_name,
2
       count(*) FILTER (WHERE state = 'active' AND now() - query_start > interval '5 minutes') AS slow_count
3
  FROM pg_stat_activity
4
 GROUP BY application_name;

「どのジョブで slow query が出ているか」が分かるので、最適化の優先順位がつけやすくなります。

まとめ

Priority	解決手段	カバー範囲
1	`APP_NAME` 環境変数	明示制御（テスト・移行作業など）
2	`LAUNCH_LABEL` から推測	launchd ジョブ
3	`sys.modules["__main__"].__file__`	python -m / スクリプト直接実行
4	`unknown_app.{pid}`	REPL / 想定外環境

ポイントは以下の 3 つでした。

接続コードでは絶対に例外を投げない。フォールバックは死なない設計にする。
launchd 用には plist で LAUNCH_LABEL を明示する。標準では入っていない。
unknown_app.* を「設計漏れシグナル」として扱う。出ていたらどこかで application_name の解決が抜けている。

application_name を入れる作業自体は 1 時間で終わりますが、4 段階フォールバックの設計をきちんとやると、運用観測の解像度が一段階上がる効果がありました。