Splunk Enterpriseの基本をゼロから丁寧に。

この記事は約22分で読めます。
sponsored link

Transforming

transforming は「生データをテーブルに変換するコマンド」で可視化の際に用います。(e.g. top, rare, chart, timechart, stats, geostats)

chart

chart は連続したデータをプロットするコマンドです。(x軸:任意,  y軸:数値)

index=security
| chart count over host

index=security
| chart count over host by vendor_action

chart で作成したグラフは [Format] > [General] > [Stack Mode] から スタックチャート, 100%スタックチャート を選択することができます。

index=security
| chart count over host by vendor_action usenull=f

timechart

timechart は _time に沿った時系列データをプロットするコマンドです。主な用途は Line chart, Area chart です。(x軸:_time,  y軸:任意)

index=network sourcetype=cisco*
| timechart count by usage 

チャートの時間幅は span で指定することができ、指定しない場合はTime Rangeにより自動で決まります。(e.g. 直近60m→span=1m, 直近24h→span=30m)

また、[Format] > [General] > [Multi-series Mode] をYesにすると、それぞれの凡例別に軸を独立させることができます。これによりスパゲッティチャート (ごちゃごちゃしたグラフ) になることを防げます。なお、Treills Layout での代替も可能です。

index=network sourcetype=cisco* 
| timechart span=6h count by usage usenull=f

フィルタリング & テーブル形式

eval

eval は集計した結果を新しいカラム (フィールド) として追加するコマンドです。なお追加したいフィールド名が既に存在する場合、既存のフィールドの値が上書きされます。

eval でインデックスデータが編集されることはなく、新しいデータもインデックスには上書きされません。生成されたフィールドは検索で再利用可能です。(フィールド値は case-sensitive)

例えば action ごとの合計 bytes から帯域を計算し新規カラムとして追加したい場合、以下のように表せます。

index=network sourcetype=cisco*
| stats max(sc_bytes) as max_sc_bytes, sum(sc_bytes) as Bytes by action
| eval "Bandwidth (MB)" = round(Bytes/(1024*1024), 2)

※Splunkではパイプライン ” | ” を追加する毎に中間テーブルが生成/更新されるイメージです。

ここでは stats, eval で役割に違いがないように見えますが両者には以下の違いがあります。

  • stats:既存フィールドを集計して新規フィールドを生成。by, over で分析軸を指定。
  • eval:既存フィールドを集計/非統計処理 (string, boolean) して新規フィールドを生成。同じ検索ラインでの再帰処理も可能。

※「tstats」という爆速サーチを可能にするコマンドも存在しますが、ここでは割愛します。生ログではなく tsidx ファイルに直接統計処理を行うというもので、定型レポート作成など実運用でよく使われます。

stats の処理後に sort で数値順に並び替えて、eval でその数値を文字列変換したフィールドで上書きするといった使われ方もされます。少し複雑にはなりますが以下のような表現も可能です。

index=web sourcetype=access_combined action=purchase
| eval cnt_fail = if(status=200,0,1), rate = case(productId LIKE "WC%",3, productId LIKE "FS%",1, 0=0,0)
| stats sum(price) as tp, sum(sale_price) as tsp, sum(cnt_fail) as "Number of fail", avg(rate) as ar range(_time) as pt by product_name
| sort 5 -"Number of fail" useother=f
| eval Discount = round((tp-tsp)/tp*100, 1)."%", "Total prices" = "$".tostring(tp, "commas"), "Average rate" = ar."pt", "Purchasing time" = tostring(pt, "duration")
| fields - tp, tsp, pt
| rename product_name as "Product"

  • if(status=200, 0, 1):status が200であれば “0”,  そうでなければ “1” を返す。
  • case(productId LIKE “WC%”, 3, productId LIKE “FS%”, 1, 0=0, 0):productId の先頭文字がWCであれば “3”,  FSであれば “1”,  その他であれば “0” を返す。なお、0=0は必ずtrueになるのでその他を表現できる。
  • range(_time):_time の Max〜Min の差分。つまり時間範囲。
  • sort 5 -“Number of fail”:イベント数を5に絞って Number of fail で降順にソート。名前にスペースが含まれるので””が必要。
  • useother=f:イベント数を5に絞った際に生成される”その他”イベントを非表示。
  • round:引数を指定しないことで数値を整数 (=少数0桁) に丸めている。
  • tostring(tp, “commas”):tp を小数のまま文字列に変換。
  • tostring(pt, “duration”):pt の書式を数値→hh:mm:ssに変換。(e.g. 175[s]→00:02:55)

また evalcount と組み合わせてフィールド内の特定文字のカウントにも使われます。

index=web sourcetype=access_combined
| stats count(eval(action="view")) as View, count(eval(action="addtocart")) as "Add to cart", count(eval(action="purchase")) as Purchase by host
host View Add to cart Purchase
www1 1073 1166 1151
www2 1163 1178 1183
www3 1137 1119 1137

search vs. where

search, where は両方とも結果をフィルタリングしますが、利用する上で若干の違いがあります。

search where
主な用途 1つのフィールドが対象のとき 複数フィールドを比較したいとき
検索パイプラインの位置 どこでも利用可 2つ目以降でのみ利用可
大文字小文字の区別 (Sensitive) Insensitive Sensitive
ワイルドカード × 一文字(_), 複数文字(*) に対応
関数の利用 × ◎ (e.g. isnum, isnull, len, etc.)

fillnull

fillnull は結果のテーブルに含まれる NULL をどの文字で埋めるかを決めるコマンドです。”value=” で文字を指定しますが、何も指定しなければ 0 埋めされます。

index=sales
| chart sum(price) over fruits by country
| fillnull
fruits JAPAN CHINA USA
Orange 200 200 200
Banana 0 100 100
Apple 0 0 150
index=sales
| chart sum(price) over fruits by country
| fillnull value="None"
fruits JAPAN CHINA USA
Orange 200 200 200
Banana None 100 100
Apple None None 150

イベントの相関

transaction

transaction は複数のソース (e.g. App, Host) で発生したイベントを関連づけて1つのイベントとして表示するコマンドです。

例えば、オンラインストアで買い物をする際には、Application Server, Database, E-commerse Engine などを経由するため、その過程で多くのイベントが発生します。transaction を用いるとこれらのイベントを共通のキー (e.g. session_id) で紐づけることができるのです。

index=web
| transaction session_id

また transaction は元データに「duration, eventcount」を自動で付与するので、これを用いた新規フィールドを作成することも可能です。もちろんレポート作成にも対応しています。

ただ transaction を使うと1行に長々とイベントが表示されてしまうので、table を使って表構造に変換してあげます。可読性UPが狙いです。

index=web sourcetype=access_combined
| transaction session_id maxspan=5m maxpause=30s startswith=eval(action="view") endswith=eval(action="purchase")
| eval duration = tostring(duration, "duration")
| sort 10 -duration
| table _time session_id duration eventcount action
_time session_id duration eventcount action
2021-07-06 17:41:06 21SFAB32FD 00:03:32 15 view
addtocart
purchase
2021-06-22 05:31:26 1SAS2F2SA2 00:02:58 15 view
purchase
2021-07-13 11:21:23 AF8DG91S8 00:02:26 21 view
addtocart
changequantity
purchase

transaction では上記のように、maxspan, maxpause, startswith, endswith などを指定できます。

  • maxspan=5m:1つのトランザクションの最大時間範囲を5分に設定。
  • maxpause=30s:30秒以上時間がたてば別のトランザクションとする。
  • startswith=eval(action=”view”):action=”view” で始まるトランザクションを抽出。
  • endswith=eval(action=”purchase”):action=”purchase” で終わるトランザクションを抽出。

transaction は複数の送受信データを解析してバグの原因を特定するという用途にも向いており、例えば、何らかの通信が REJECT されたログのみをまとめて見る場合、以下のようになります。

index=network | transaction id_a id_b id_c | search REJECT

transactionコマンド vs. statsコマンド

transaction stats
処理の速さ × ◎ (実運用向き)
主な用途 単一イベントの情報量が少ないとき
相関があるログをまとめて見たいとき
start/end値, timeを基にグループ化したいとき
集計項目が予め決まっているとき
集計結果をテーブル形式で見たいとき
フィールド値でグループ化したいとき
イベント数の制限 1000 [default] なし (無制限)

※transaction で表示されるイベント数を1000より大きくしたい場合、管理者が limit.conf ファイルのmax_events_per_buckets を編集する必要があります。

フィールドの作成/管理

Splunk では生データ内の sourcetype, key/valueペア などは自動で抽出されますが、Field Extractor (FX) を用いてオリジナルなフィールドを抽出することも可能です。なお、メタフィールド (host, source, sourcetype), _time, _raw は検索前にインデックスとして生成されています。

FX は「Settings」「Fields Sidebar」「Event Actions」の3箇所から起動することができ、抽出方法には「Regex (正規表現)」「Delimiter (区切り文字)」の2種類があります。抽出後は knowledge object として保存することで他ユーザーと共有可能です。

  • Settings:[Settings] > [Fields] > [Field extractions] > [Open Field Extractor](画面右上)
  • Fields Sidebar:画面左のサイドバー (下部) から [+ Extract New Fields]
  • Event Actions:検索後のイベントを1つ開き [Event Actions ] > [Extract Fields]

※FXは基本的に sourcetype に紐づいた抽出を行いますが、[Settings]からFXを起動した場合のみ source に紐づいた抽出を行えます。

無事に起動できたら次は表示されるイベントを1つ適当にクリックして次へ進むと、[Regular Expression](正規表現),  [Delimiters](区切り文字) の選択項目が表示されるので、自分が選びたい方を選択します。

Field Extractor (FX) – 正規表現抽出

Regex (正規表現) 抽出はシステムログファイルのような「非構造化データ」に最適で、指定した表現にマッチする/しないデータを抽出できます。

イベント内の該当フィールドをマウスで囲むことで抽出フィールドを選択でき、新規フィールド名を設定してあげれば後は Splunk が自動で正規表現を作成してくれます。以下では port 番号を新規フィールドとして作成しました。

抽出されたフィールドは青でハイライトされ、画面下部にプレビューが出るので、その抽出で問題が無いかをユーザーが確認しやすい設計になっています。

[Next] ボタンを押すと最終レビュー画面が表示されるので、そこで問題がなければ [Next] ボタンを押してパーミッションを決定、保存すれば完成となります。

Field Extractor (FX) – 区切り文字抽出

Delimiters (区切り文字) 抽出はCSVファイルのような「構造化データ」に最適です。指定できる区切り文字には commas( , ), pipes ( | ), spaces, tabs, other character があります。

Delimiters も基本的に Regex と同じフローなのでここでは割愛します。

エイリアス, 定義済みフィールド

フィールドエイリアス

フィールドエイリアスとは「既存フィールドのエイリアス (別名)」のことです。

1つのフィールドに対して複数エイリアスを設定することができ、設定した分だけ新規フィールドとしてサイドバーに表示されます。エイリアスをいくつ設定しても元ののフィールドへの影響はありません。

エイリアスはフィールド抽出後 Lookups 前に処理されるため、新規で設定したエイリアスをキーに lookup コマンドで外部ファイルと紐づけることが可能です。(←重要)

エイリアスの作成方法は [Settings] > [Fields] > [Fields Aliases] > [New Fields Alias] > 必要な情報を入力 (下記参照)。作成後はサイドバーから自分で確認してみましょう。

定義済みフィールド

定義済みフィールドとは「繰り返し記述する冗長的な処理をショートカット (引数なし関数) にしたもの」です。

Exal expression には元データに存在するフィールドのみを記述でき、Lookupで紐付けたフィールドは含めることができません。

タグ, イベントタイプ

タグ

タグとは「field/value ペアに対する関連情報」のことで、いくつでも設定可能です。

インスタの# (ハッシュタグ) と似たような役割ですが、Splunkのタグは case-sensitive (大文字小文字を区別する) という点でインスタの#とは異なります。

タグの作成方法は、[Action](イベント画面右側) > [Edit Tags] > 好きなタグを入力 (カンマ区切り) 。

例えば今回は、agent, Agent, AGENT という3つのタグを “useragent” というフィールドに紐付けたので、各イベントにこれらのタグが表示されていることが分かります。

例えば、複数データソースの複数フィールドに同一のタグを設定することで、コマンドで tag 検索をかけてそれら関連するイベントのみを抽出するといったことが可能です。

index=web tag=Age*   #ワイルドカードを使った検索
index=web tag::useragent=AGENT   #具体的なフィールドで絞り込んだ検索

タグの作成後は [Settings] > [Tags] > [Listed by field value pair] で権限を編集したり、タグを追加/編集することもできます。

イベントタイプ

イベントタイプは「検索に基づいてイベントを分類するもの」で、似たイベントをタグ付けして組織内でイベントを共有できます。

Event type の作成方法は以下の2パターンです。

  • [Save As] > [Event Type] > Names [, Tags, Color, Priority] の決定。
  • イベント内の [Event Actions] > [Build Event Type] > 各フィールド内の具体的な値を選択 。

※Priorityは数が小さい方が優先度が高く、1(高)〜10(低) から選択できます。

イベント内の [Event Avtions] から作成する際のフローがこちらです。

Event type を作成することで、検索で利用できるようになります。

index=web eventtype=<eventtype_name>

Event Types vs. Saved Reports

Event Types Saved Reports
主な用途 検索文字でイベントを分類したいとき
検索の一部として使いたいとき
他ユーザーと結果を共有したいとき
結果をダッシュボードに追加したいとき
分析の軸を変える予定がないとき
時間範囲 含まれない (検索時に設定) 含まれる (変更不可)

マクロ

マクロとは「よく使う検索条件を関数化した機能」です。

定義済みフィールドと異なり「引数・エラー条件・エラー文」など詳細な設定が行えるのが特徴です。

マクロの作成方法は [Settings] > [Advanced search] > [Search Macros] > 必要な情報を入力(下記参照) で、今回はこちらの計算式をマクロとして定義しました。

stats sum(price) as JPY by product_name
| eval $currency$ = "$symbol$" + tostring(JPY*$rate$, "commas")
| JPY = "¥".round(JPY)

上の画像からわかる通り、マクロを作成時には「引数を$$で囲む」「名前の末尾に引数の数を指定する」といったルールがあります。後はマクロ名をバックティック(`)で囲み引数を指定すれば、実際の検索で利用することができます。

※バックティック(`)は Shift + @ で入力可能。

index=main
| `sales(USD, $, 0.0083)`

ワークフローアクション

ワークフローアクションとは「Splunk での検索結果と外部リソースを連携させる機能」です、GET, POST, Search の3種類があります。

  • GET:外部リソースからデータを取得。
  • POST:フィールド値を外部リソースへ送信。
  • Search:フィールド値を次の検索に再利用。

ワークフローアクションの作成方法は、[Setting] > [Fields] > [Workflow] > [actions] > [New Workflow Action] > 必要な情報を入力 (下記参照)。

GET

下の画像のように必要な情報を入力すると [Event Actions] から利用できるようになります。

イベントを展開して [Event Actions] を押すと、先ほど作成したGETワークフローが表示されていることが確認できます。

POST

先ほどは Show action in (画像中央) で Events menus を選択しましたが、今回のように Fields menus を選択することも可能です。結果は割愛します。

Search

Search も前述した GET, POST と同じ要領で作成できます。

データモデル

データモデル

データモデルとは「階層化されたデータセット」のようなもので Pivot で使われます。
※Pivot はフィールドから直感的にテーブルを作成してレポート/ダッシュボードに利用できる機能です。

▼データモデルのサンプル (階層が深くなるにつれ検索条件が増える)

上の画像では左のサイドバーに EVENTS タイプしかありませんが、他にも SEACHER (e.g. user), TRANSACTION (e.g. visit duration) があります。計3種類です。

データモデル自体は [Settings] > [Data Models] > [New Data Model] から作成でき、既存のデータモデルの階層を追加する場合は、データモデルの画面左サイドバー [Add Dataset] から Root Event や Child などを選択してください。

フィールドの追加方法は以下の5種類です。

  • Auto-Extructed:デフォルト or 手動で追加したフィールドを追加。
  • Eval Expression:指定した計算式に基づいてフィールドを追加。
  • Lookup;既存のテーブルと紐付けてフィールドを追加。
  • Regular Expression:指定した正規表現にマッチする/しないフィールドを追加。
  • Geo IP:IPアドレスに基づいて地理フィールド (e.g. 緯度軽度, 国) を追加。

フィールドタイプには String, Number, Boolean, IPV4 が選択でき、Geo IP でフィールドを追加する場合のみ IPV4 が必須となります。

フィールドフラグには Optional (任意), Required (pivotで必ず表示), Hidden (pivotで非表示), Hidden&Required (このフィールドを指定した場合のみ表示) の4種類を選ぶことができます。

なお作成したデータモデルはWEB上でバックアップ、他ユーザーへの共有、本番環境への移動することも可能です。

Search/Transaction データセット

Transaction Datasetを追加する場合、少なくとも1つのイベントか Search Dataset が必要になります。

また Search/Transaction Dataset は高速化の設定はできません。

データモデル高速化

Data Model Acceleration は「全てのフィールドをインデックス化することで Pivot の完了時間を短縮させる」という機能です。なお、一度インデックス化されたデータは編集できません。

高速化の設定はデータモデルの画面から [Edit] > [Edit Acceleration] > [Acceleration] にチェックを入れるだけです。ただし高速化の設定は、管理者・accelerate_datamodel を付与された者 に限ります。

高速化には time-series index files (tsidx) が使用されており、一度インデックス化されたデータを再び編集することはできません。

Pivot へ取り込み

Pivot でデータモデルを取り込むと初めに全てのイベント数が集計されます。なお pivot を用いてデータモデルを取り込むことも可能です。

| pivot <data_model_name> <object_name> <split_row_field> <split_column_field> <filter>

データの全体像

データモデルに格納されているデータを一覧化することもできます。

Common Information Model (CIM) Add-on

CIM Add-on

CIM アドオンとは「Splunkのデータを正規化する手法」のことです。

例えば、データソースAで [username=Tom], データソースBで [customer_name=Tom] というデータになっていた場合、これらのフィールドを同一のもの [name=Tom] として扱うことができます。
※ フィールド名は利用する CIM によって事前に決まっており、各ログのフィールド名が固有のフィールド名に正規化されるといったイメージです。

CIM アドオンを活用するメリットは大きく以下の3つです。

  • 単一の Splunk deployment で複数アプリを利用可能。
  • 複数アプリを使うためのオブジェクト権限をグローバルで一括設定できる。
  • 異なる source, sourcetype を効率的に紐付けられる。

さらに、定義済みモデルが22個用意されており対応可能なドメインも豊富にあるのも CIM アドオンの特徴です。

CIM アドオンの利用方法は、[Settings] > [Data Models] > CIM に対応させたいデータモデルを選択 > データセットに紐づいているタグを確認 > 不足分のデータをフィールド抽出, lookups を用いて作成 > datamodel コマンドや WEB 上での Pivot を使い CIM へマッピングです。

datamodel

datamodel は特定のデータモデルを検索するコマンドです。

| datamodel <datamodel_name> <datamodel_dataset_name>

from

from はデータモデル, データセット, 保存した検索, CSV lookup ファイル, テーブルからデータを抽出するコマンドです。

datamodel は定義された全てのフィールドを返すのに対して、from は指定したフィールドのみを返します。なお、from は必ずコマンドの先頭で利用しなければなりません。

| from datamodel:"yonaha_info_server.access"