Architecture and data flow#

AIエージェントアクティビティは、エージェントが実行するSQLからPostgresログとLokiを介して、読み取った画面に移動します。パスを理解することは、ビューアーに表示できるものと、セッションデータが表示される前に同期が必要な場合がある理由を推論するのに役立ちます。

データパス#

AIエージェントが実行する各クエリは、ビューアに監査可能なセッションとして表示される前に4つのステージを通過します。

AI agent (Airman MCP)
  │  Tags each query: application_name = airman:<purpose>/<session-short>
  ▼
Postgres cluster
  │  Logs every statement with application_name, SQL, duration,
  │  error severity, database, role, and PID (JSON log format)
  ▼
Loki  (HM Loki pipeline, or a standalone Loki instance)
  │  Stores log lines, labeled by container and cluster, queryable via LogQL
  ▼
Viewer backend
  │  Queries Loki via LogQL, parses Postgres log lines into sessions,
  │  caches session summaries, and injects upstream credentials server-side
  ▼
Viewer (browser)
     Renders clusters, sessions, and session detail

コンポーネントの責任#

3つのコンポーネントはデータパスの異なる部分を処理し、バックエンドがすべての上流アクセスを仲介するため、資格情報がブラウザーに到達することはありません。

ビューアブラウザー シングルページアプリケーションは、クラスター、セッション、およびセッションの詳細画面をレンダリングし、クライアント側のルーティングを管理し、すべてのデータのバックエンドを呼び出します。

ビューアバックエンド 次の軽量サービス。

インスタンス構成と、関連するアップストリーム資格情報を保存します。
インスタンスタイプに基づいて、各要求を正しいHMまたはLokiエンドポイントにルーティングします。
生のPostgres JSONログ行を構造化されたセッションおよびステップデータに解析します。
セッションの概要をキャッシュし、段階的に同期して、冗長なクエリを回避します。
プロキシ化されるすべての要求に適切な認証ヘッダーを挿入するため、資格情報がブラウザーに到達することはありません。

アップストリームソース HM展開APIを介してプロジェクト、クラスター、およびLokiログクエリを公開し、マシンユーザーAPIキーで認証します。または、オプションのベアラートークン認証を使用して、直接照会されるスタンドアロンのLokiインスタンス。

エージェントアクティビティのタグ付け方法#

Airman MCPは、開くすべての接続にPostgres application_name セッション変数を自動的に設定します。

SET application_name = airman:<purpose>/<session-short>;

そこで

purpose — AIRMAN_MCP_PURPOSE たとえば billing 、support 、またはanalytics などを介してAirmanインスタンスごとに構成された静的ラベル。設定されていない場合は、デフォルトの_ になります。
session-short — MCPセッショントークンの最初の8つの16進文字。すべてのクエリーを1つのインタラクションにグループ化するために使用されます。

タグはPostgresセッションレベルで設定され、その接続の後続のすべてのステートメントに対して持続します。

ログエントリのフォーマット#

JSONロギングを有効にすると、Postgresは次のようなエントリを生成し、Lokiが取り込みます。

{
  "ts": "2026-05-22T14:30:45.123Z",
  "record": {
    "application_name": "airman:billing/a1b2c3d4",
    "message": "Duration: 45.2 ms  statement: SELECT customer_id, email FROM customers WHERE active = true",
    "database_name": "banking_db",
    "user_name": "agent_read",
    "command_tag": "SELECT",
    "error_severity": "",
    "log_time": "2026-05-22 14:30:45.123 UTC",
    "process_id": "12345",
    "session_id": "789",
    "sql_state_code": "",
    "query_id": "0"
  }
}

バックエンドはmessage フィールドからSQLテキストと期間を抽出し、残りのフィールドをステップメタデータに直接マッピングします。

クエリパス#

ビューアは、表示しているものに応じて、さまざまなLokiクエリを発行します。

セッションリスト クラスターを開くと、バックエンドは同期ウォーターマークを読み取り、Lokiからキャッシュされていない履歴をバックフィルし、キャッシュされたセッション概要をロードし、最新の日のライブクエリを追加し、結果をマージします — セッションの並べ替え最近見た人によって。

セッションの詳細 セッションを開くと、バックエンドはそのクラスターにスコープし、セッションショートにフィルタリングしたLogQLクエリーを発行し、返された各ログラインをステップの順序付けたリストに解析します。

バックエンドは、次の形式のLogQLクエリーを使用します。

{container="postgres", cnp_cluster_id=~"<clusterId>.*"} |= "airman:"

セッションの詳細では、session shortがラインフィルタとしてairman: を置き換えます。

バックグラウンド同期#

ページが読み込まれるたびに高価なフルレンジのLokiクエリを回避するため、バックエンドはセッション概要をキャッシュし、ウォーターマークベースのインクリメンタル同期で最新の状態に保ちます。

同期パイプライン#

クラスターのセッションリストが要求されると、バックエンドは次のことを行います。

ウォーターマークを読み取ります — 最も古い既知のエントリのインスタンスごと、クラスターごとのレコードと、キャッシュがどの程度まで移入されたか。最初の同期では、両方ともゼロです。
最も古いエントリを見つけます — 最初の同期で、30日間、90、次に365にわたって逆方向に検索して、キャッシュするフルタイム範囲を確立します。何も見つからない場合、今日が最も古いものとして扱われます。
バックフィル履歴 — 最初のエントリから今日までキャッシュを1日単位で埋め、セッションの概要をupsertし、ウォーターマークを進めます。進行状況は、ステータスエンドポイントを介して報告されます。
キャッシュされたセッションのロード — クラスターのキャッシュされたすべての概要を読み取ります。
今日のライブデータを取得します — 過去24時間に単一のクエリーを実行します。このデータは常に新しく、キャッシュされません。

次に、バックエンドは、キャッシュされたセッションとライブセッションをセッションショートごとにマージし、歩数を合計し、最初と最後に表示されたタイムスタンプを調整し、最後に表示された順に並べ替えます。

アダプティブチャンク分割#

高冗長クラスターは、Lokiのクエリごとの制限を超える場合があります。 1日分のチャンクが高すぎるとLokiが報告すると、バックエンドはウィンドウを半分に分割し、各半分を独立して再試行し、最小チャンクサイズ1時間まで下げます。 1時間のチャンクでもまだ失敗する場合は、エラーをログに記録し、そのウィンドウをスキップします。

再試行ポリシー#

アップストリームLoki呼び出しは、再試行ポリシーでラップされます。一時的なエラー502、503、504の場合、指数バックオフ2秒から開始、2倍、30秒で上限30秒で最大5回の再試行。再試行不可エラー400、401、404、500は、すぐに失敗します。

再同期#

Configureページから再同期をトリガーすると、バックエンドはインスタンスのキャッシュセッションとウォーターマークをクリアし、バックグラウンドジョブを開始します。Lokiインスタンスの場合、単一のスタブクラスターを再同期します。 HMインスタンスの場合、すべてのプロジェクトとクラスターを検出し、それらを並列同期します。要求はすぐに返され、ジョブが実行されると、インスタンスのステータスが Syncing から Done または Error に進みます。 Connecting data sources を参照してください。

Architecture and data flow

目次