以前、「Sumo Logicを利用したハイブリッドクラウド環境に最適なサーバ＆ネットワーク監視ソリューション」と題して、モニタリングソリューションの紹介をいたしました。
これの追加機能としてNetFlow v9のフローデータを取り込むためにFlowコレクターを開発しました。
今回はそのFlowコレクターに関する記事になります。
本記事は開発したFlowコレクターをメインにしているので、NetFlowの詳細に関しては外部サイトをご参照ください。
参考までにCiscoさんのサイトLinkを記載しておきます。
NetFlow Version 9 Flow-Record Format [IP Application Services] - Cisco Systems
開発環境
以下は実際に開発で利用した環境です。
（特に拘りがなかったので、余っていたLinux上で開発しました。）
OS ：CentOS 7
開発言語 ：Python 3.7.3
Pythonパッケージ ：socket、struct
それでは本題に入りたいと思います。
Flowコレクターについて
NetFlow v9のパケットフォーマット
Flowコレクターの解説の前にNetFlow v9のパケットの構造に関しておさらいです。
NetFlow v9はPacket header、１つ以上のTemplate FlowSet、１つ以上のData FlowSetから構成されています。
つまり以下のような構成になっています。
また、それぞれのフォーマットは以下となっています。
①Packet Header
②Template FlowSet
③Data FlowSet
Flowコレクターでの処理
Flowコレクターでは、それぞれのフォーマットを基にNetFlowのパケットを解析し、必要なデータをSumo Logicへアップしていきます。
Flowコレクターの大まかな処理の流れは以下のようになります。
- バージョンの確認
- Template FlowSetかData FlowSetかを判定
- Template FlowSetの解析
- Data FlowSetの解析
- Sumo Logic用にデータを整形
- Sumo Logicへアップロード
なお、パケットから値を取り出す際はPythonのstructを使用し、ビックエンディアンとして処理しています。
１．バージョンの確認
まずはバージョンを確認します。
バージョンはPacket Headerの先頭2バイトに格納されているので以下の処理で取得し、取得した値が9である事を確認します。
-------------------------
version = struct.unpack("!H", Header[0:2])
-------------------------
２．Template FlowSetかData FlowSetかを判定
次に受信したデータがTemplate FlowSetなのかData FlowSetなのかを判定します。
判定にはデータ部の先頭2バイトに格納されている値で判定します。
以下の処理で取得します。
-------------------------
FlowSetID = struct.unpack("!H", Data[0:2])
-------------------------
取得した値が0～255の場合はTemplate FlowSet、255より大きい場合はData FlowSetを表しています。
３．Template FlowSetの解析
2で取得した値が0～255だった場合、受信したパケットはTemplate FlowSetということになるので、以下の処理でテンプレートIDとフィールドの数を取得します。
-------------------------
TemplateID = struct.unpack("!H", Data[4:2])
FieldCount = struct.unpack("!H", Data[6:2])
-------------------------
これでテンプレートIDとそのテンプレートが格納するフィールドの個数が確認できました。
続いてテンプレートのフィールド情報を解析していきます。
フィールド情報はField IDとLengthが2バイトずつ、フィールドの数分繰り返されているのでループで処理していきます。
-------------------------
for field_idx in FieldCount:
n = フィールドの情報のスタート位置 + (field_idx – 1) * 4
FieldID = struct.unpack(“!HH”, rawdata[n:2])
FieldLength = struct.unpack(“!HH”, rawdata[n+2:2])
-------------------------
この処理を繰り返すことで、テンプレートのフォーマットが解析できます。
解析したデータはData FlowSetの解析時に使用するためテンプレートIDをキーとしてフィールドID、フィールド長の辞書型のデータとして保存しておきます。
-------------------------
{"256": [ ←テンプレートID
{"id": 8, "length": 4}, ←以降、フィールドID、フィールド長が繰り返し
{"id": 11, "length": 2},
{"id": 5, "length": 1},
・・・
]}
-------------------------
４．Data FlowSetの解析
2で取得した値が「0～255以外」の場合、受信したパケットはData FlowSetということになるので、3で作成したテンプレートを基に解析していきます。
Data FlowSetの場合は、先頭の2バイトがテンプレートIDとなるので、以下でテンプレートIDを取得します。
-------------------------
TemplateID = struct.unpack(“!IH”, rawdata[0:2])
-------------------------
先ほど3で生成されたフォーマットの例で記載すると、以下のようにデータを取得します。
-------------------------
value1 = struct.unpack(“!I”, rawdata[4:4]) ←rawdataの4バイト目から4バイト分
value2 = struct.unpack(“!H”, rawdata[8:2]) ←rawdataの8バイト目から2バイト分
value3 = struct.unpack(“!B”, rawdata[10:1]) ←rawdataの10バイト目から1バイト分
・・・
-------------------------
５．Sumo Logic用にデータを整形
NetFlowのデータをSumo Logicで表示させるために以下の値をData FlowSetから取得します。
①パケットの受信時間
パケットの受信時間はData FlowSetの”FIRST_SWITCHED”と”LAST_SWITCHED”から計算しています。
仕様ではFIRST_SWITCHEDは最初のパケットがスイッチされた実行日時、LAST_SWITCHEDは最後のパケットがスイッチされた実行日時となっていたので、
「LAST_SWITCHED = Data FlowSetの受信時間」として、「パケットの受信時間 = Data FlowSetの受信時間 – (“FIRST_SWITCHED”と”LAST_SWITCHED”の差)」で計算しています。
②受信したデータ量
受信したデータの合計は”IN_BYTES”から取得しています。
③IPアドレス
IPアドレスは、送信先アドレスは” IPV4_DST_ADDR”、送信元アドレスは” IPV4_SRC_ADDR”から取得しています。
④ポート番号
ポート番号は、送信先ポート番号は” L4_DST_PORT”、送信元ポート番号は” L4_SRC_PORT”から取得しています。
⑤プロトコル
プロトコルは④で取得した送信先ポート番号からsocket.getservbyportで取得しています。
６．Sumo Logicへアップロード
５で取得したデータをJSONフォーマットに整形しSumo LogicのHosted Collectorに送信しています。
以下に掲載したスクリーンショットは実際にSumo Logicが受信したFlowコレクターから送られてきたデータになります。
そして、これらFlowコレクターから送られてきたデータから、送信元、送信先、送信先ポート番号を基にダッシュボードにすると以下のような通信したデータを可視化させることができます。
このダッシュボードでは、timesliceを10分としてデータ量の合計値を表示させているため少しわかりづらい図になっています。
timesliceを短くするとか、合計値ではなくデータ量をそのまま表示させたりするといった表示上の工夫は必要になりますが、開発したFlowコレクターでNetFlowのデータもSumo Logicに取り込み、可視化することが可能です。
NetFlow以外にもコレクターを間に挟むことでSumoLogicにデータを取り込むことが可能になりますので、取り込んでみたいデータ・可視化したいデータがございましたらお気軽にお問い合わせください。
