IBM Cloud Internet Servicesのアクセスログを検索する方法

:::note warn
IBM Cloud Data Engine(旧SQL Query)は2024年1月18日をもって非推奨となり、2024年2月18日以降、新規インスタンスの作成が出来なくなりました。さらに、2025年1月18日をもってEoSとなるため、現在ご利用のインスタンスから後継サービスであるAnalytics Engineなどの利用へ変更する必要があります。
:::

# はじめに

IBM Cloudが提供するCDNサービスである、IBM Cloud Internet Services(CIS)のアクセスログは、ログプッシュの設定により、IBM Cloud Object Storage(ICOS)に保管されます。

そのままでは参照しづらいので、Watson StudioとSQL Queryを用いた検索・参照方法をご紹介します。

# 前提

+ IBM Cloudアカウント
+ CISインスタンスがあり、ログプッシュの設定済み
+ 上記ログと、検索結果を保管するためのICOSバケット（別々を推奨）
+ Watson Studioインスタンス（ライト版でOK）
+ SQL Queryインスタンス（ライト版でOK）

# ログのパスについて

CISのログプッシュは、改行区切りのJSONファイルがgzip圧縮されたものが、5分ごとに1ファイル作成されます。

ICOS上のパスは、例えば2021年12月2日0時0分から0時5分(UTC)までのログを含むファイルでは以下のようになります。また、ログプッシュ設定時、「ログを日付別のサブフォルダーに整理する」 を選択した場合は、この例では冒頭に `20211202/` が付与されたパスになります。

```
20211202T000000Z_20211202T000500Z_90abcdef.log.gz
```

# ログの形式について

JSONの形式ですが、ログプッシュで設定したログ・フィールド名がそのままキーになる単純な形です。

どのようなログ・フィールドがあるのかは、以下を参照してください。

+ [ログ・フィールドへのアクセス - 使用可能なフィールド](https://cloud.ibm.com/docs/cis?topic=cis-log-fields#http-requests)

# タイムスタンプ形式の違い

ログ・フィールドには、EdgeStartTimestampなどのタイムスタンプが入るフィールドがあるのですが、ここにどのような形式で保管されるかは、ログプッシュ設定時に指定することが出来ます。

以下がその設定と、実際に格納されるデータになりますので、扱いやすい設定を選択すると良いです。（個人的には、JSTに変換しやすいUnixが扱いやすいかなと思いますが、より厳密なタイムスタンプが必要ならばUnixNanoでしょうか）

|ログプッシュの設定|実際に格納される型と値|
|:-----------|:------------|
|Unix|整数：UNIX時間(秒)|
|UnixNano|整数：UNIX時間(ナノ秒)|
|RFC3339|文字列：`YYYY-MM-DDThh:mm:ssZ`|

# ログ参照Notebook概要

まず、SQL Queryのモジュールを導入、importします。エラーメッセージが出力されることがありますが、無視して構いません。

```Python:cell-1
# SQLQuery導入
!pip -qqq install ibmcloudsql pyarrow
import ibmcloudsql
```

次に、入力元のICOS、出力先のICOS、SQL Queryへの必要な権限を持つAPI Keyと、使用するSQL QueryのCRNを指定して、クエリー実行の準備を実施します。CRNは、リソース一覧で該当のインスタンスを選択した際に表示されるダイアログから確認出来ます。

```Python:cell-2
# API Key
cloud_api_key = '**************'
# SQL Query設定
sql_crn = '**************'
# クエリー実行準備
sqlClient = ibmcloudsql.SQLQuery(cloud_api_key, sql_crn)
sqlClient.logon()
```

次にクエリーを組み立てます。例えば、2021年11月25日のAM5時台(UTC)のWAFブロックイベントのログをCSV形式で抽出するためのクエリーは以下のような感じになります。「ログを日付別のサブフォルダーに整理する」を設定していて、タイムスタンプはUnixを指定している環境を想定しています。

```Python:cell-3.1
# クエリー組み立て
query = """
SELECT cast(edgestarttimestamp + 32400 as timestamp) as DATETIME, WAFRuleID, WAFRuleMessage, ClientIP, ClientCountry
FROM cos://INPUT-BUCKETNAME/20211125/20211125T05* STORED AS JSON
WHERE WAFRuleMessage != ""
ORDER BY edgestarttimestamp
INTO cos://jp-tok/OUTPUT-BUCKETNAME STORED AS CSV
"""
```

別の例として、特定のパスヘのアクセスをとにかく全部見たい場合は、以下のようなクエリーになります。なぜ `SELECT *` としないのか、ですが、ログ・フィールドを全て保存している環境では、FirewallMatchesActionsなどの配列を含むキーがJSONファイルに含まれており、そのままではCSV形式に単純に変換出来ず、単純にそれらを除いた全てのログ・フィールドを指定することにしています。配列を含むキーも抽出したい場合は、CSVではなく、JSON形式で出力するか、SQL Queryの[EXPLODE](https://cloud.ibm.com/docs/sql-query?topic=sql-query-sqlfunctions#explode)という関数を用いて展開するなどの工夫が必要になります。

```Python:cell-3.2
query = """
SELECT cast(edgestarttimestamp + 32400 as timestamp) as DATETIME, cachecachestatus, cacheresponsebytes, cacheresponsestatus, cachetieredfill, clientasn, clientcountry, clientdevicetype, clientip, clientipclass, clientrequestbytes, clientrequesthost, clientrequestmethod, clientrequestpath, clientrequestprotocol, clientrequestreferer, clientrequesturi, clientrequestuseragent, clientsslcipher, clientsslprotocol, edgecoloid, edgeendtimestamp, edgepathingop, edgepathingsrc, edgepathingstatus, edgeratelimitaction, edgeratelimitid, edgerequesthost, edgeresponsebytes, edgeresponsecompressionratio, edgeresponsecontenttype, edgeresponsestatus, edgeserverip, edgestarttimestamp, originip, originresponsebytes, originresponsehttpexpires, originresponsehttplastmodified, originresponsestatus, originresponsetime, originsslprotocol, parentrayid, rayid, securitylevel, wafaction, wafflags, wafmatchedvar, wafprofile, wafruleid, wafrulemessage, workercputime, workerstatus, workersubrequest, workersubrequestcount, zoneid
FROM cos://jp-tok/INPUT-BUCKETNAME/20211125/20211125T05* STORED AS JSON
WHERE clientrequestpath = '/ready'
ORDER BY edgestarttimestamp
INTO cos://jp-tok/OUTPUT-BUCKETNAME/PATH STORED AS CSV
"""
```

次はクエリーを実行する部分です。クエリーの実行方法はいくつかありますが、クエリー実行の完了を待つタイプの関数（run_sql, wait_for_job）では、クエリーの実行が長時間に渡る場合にタイムアウトしてしまうことがあるので、以下のようにクエリーを投入したあと、定期的にその状況を確認して、終了していたら抜けるようなロジックで実行するようにします。

正常に完了すれば、「completed」と表示されます。

```Python:cell-4
import time, sys

# クエリ投入
jobid = sqlClient.submit_sql(query)
# 実行中になるまで待つ
time.sleep(10)
# 実行中ではなくなるまで待つ
while sqlClient.get_job(jobid)['status'] == 'running':
    time.sleep(10)
# 結果がfailedならば異常終了
if sqlClient.get_job(jobid)['status'] == 'failed':
    print('ERROR')
    sys.exit(1)
# 結果ステータス表示
print(sqlClient.get_job(jobid)['status'])
```

最後に、以下のように実行すると、SQL Queryへのリンク（`https://us.sql-query.cloud.ibm.com/sqlquery/?instance_crn=〜〜〜`）が表示されるので、リンク先のSQL Query画面からクエリーの実行結果を確認することができます。

```Python:cell-5
# SQL Queryへのリンクを表示
print(sqlClient.sql_ui_link())
```

また、PixieDustを用いてNotebook上で結果を確認することも可能です。

```Python:cell-6
# pixiedustでの結果内容表示準備
!pip -qqq install pixiedust
import pixiedust as px

# pixiedustでの結果内容表示
resultdf = sqlClient.get_result(jobid)
px.display(resultdf)
```

# まとめ

ふいにアクセス分析したくなることがあるかもしれないので、1つテンプレートを用意しておくといいかなと思います。

# 参考

+ [ibmcloudsql](https://github.com/IBM-Cloud/sql-query-clients/tree/master/Python)