AWS LambdaでRESTでデータ取得してSplunkに送信してみる (Python編)

はじめに

• AWS LambdaでRESTでデータ取得してSplunkに送信してみるではBlueprintを使用してLambdaからSplunkへのHEC送信に成功しました。
• しかしBlueprintはNode.jsでした。Splunk管理者はPythonの方が得意だと思いますので（偏見）、Pythonでできないか調べました。
• PythonでSplunk HECに送信するClass（Splunk-Class-httpevent）がありましたので、今回はこちらを使って実現します。
• Splunk-Class-httpeventはレイヤーにしてLambda上のPythonから呼び出せるようにします。
• Splunkで毎日運勢を占えるようになりました。

環境

• Lambdaからアクセス可能するためEC2などでグローバルIPを持つSplunk Enterprise、もしくはSplunk Cloud
• AWS Lambda

設定手順の概要

1. SplunkでHEC設定
2. AWS Lambdaのレイヤー作成（モジュールアップロード）
3. AWS Lambdaで関数作成
4. CloudWatchでLambdaを定期実行化

設定手順

1. SplunkでHEC設定

まずはいつも通り、AWS Lambdaからデータを受け取るためのHECを設定します。

1. 設定 > データ入力
2. HTTPイベントコレクターの [+ 新規追加] をクリック
3. 名前を適当に入力して [次へ] をクリック
4. 取り込み先のインデックスを選択　※今回はaws_lambdaを作成しました。
5. 生成されたトークンをメモ

以上です。
念のため、SSLの有効/無効状況やHTTPポート番号をグローバル設定から確認しておきましょう。

2. AWS Lambdaのレイヤー作成（モジュールアップロード）

PythonからSplunk HECでデータ送信する神ClassをSplunk Trustの方が作っていらっしゃいました。
Splunk-Class-httpevent
今回はこちらをありがたく使わせていただきます。

とは言ってもLambdaではpipでインストールはできません。
その代わりにLambdaのレイヤー機能を使用します。
モジュールなどをzipで固めてレイヤーにアップロードすることで、Lambda関数から呼び出すことができます。
AWS Lambda レイヤー

2-1. zipの準備

ローカルで以下のようにモジュールをpythonディレクトリにダウンロードしzip化します。
※ディレクトリ名は「python」である必要があります。レイヤーとしてアップロードしたときに/opt/配下に展開されるためです。zip名は任意です。

mkdir ./python
pip install git+git://github.com/georgestarcher/Splunk-Class-httpevent.git -d ./python
zip -r python-hec.zip ./python

2-2. レイヤーへのアップロード

Lambdaページのレイヤーメニューから [レイヤーの作成] をクリックします。

適当に名前を付け、作成したzipファイルをアップロードし、互換性のあるランタイムにPythonを選択し、[作成] をクリックします。

これでレイヤー作成は完了です。
次は関数を作成します。

3. AWS Lambdaで関数作成

3-1. 関数作成

1. Lambdaの関数メニューから [関数を作成] をクリック
2. [一から作成] を選択し諸々入力
   • 関数名：任意
   • ランタイム：Python （本記事の作成時点で最新版の3.8）
   • 実行ロール：「基本的なLambdaアクセス権限で新しいロールを作成」のまま
3. [関数の作成] をクリック

3-2. レイヤーの追加

1. 関数ページの下の方にあるレイヤー枠の [レイヤーの追加] をクリック
   
2. 「カスタムレイヤー」を選択し、カスタムレイヤーから先ほどアップロードしたレイヤーとバージョンを選択。
   
3. [追加] をクリックして完了。

これで関数から通常のimportで呼び出せるようになりました。

3-3. Pythonスクリプト作成

それではPythonスクリプトを作っていきましょう。

Splunk-Class-httpeventの使い方としてサンプルが提供されています。
特に重要な箇所は以下です。

# HEC URLおよびトークン
http_event_collector_key = "4D14F8D9-D788-4E6E-BF2D-D1A46441242E"
http_event_collector_host = "localhost"

# Splunkメタデータ
payload = {}
payload.update({"index":"test"})
payload.update({"sourcetype":"crime"})
payload.update({"source":"witness"})
payload.update({"host":"mansion"})

# データ送信（単発）
for i in range(5):
    event = commitCrime()
    event.update({"action":"success"})
    event.update({"crime_type":"single"})
    event.update({"crime_number":i})
    payload.update({"event":event})
    testevent.sendEvent(payload)

# データ送信（バッチ）
for i in range(50000):
    event = commitCrime()
    event.update({"action":"success"})
    event.update({"crime_type":"batch"})
    event.update({"crime_number":i})
    payload.update({"event":event})
    testevent.batchEvent(payload)
testevent.flushBatch()

それでは、今回も適当なREST APIでデータを取ってみましょう。

占いAPIでSplunkで毎日お手軽に占えるようにしたいと思います。
Web ad Fortune 無料版API
※利用規約あり。商用利用の場合はビジネス版あり。

このような感じで年月日を渡せば良いようです。
http://api.jugemkey.jp/api/horoscope/free/year/month/day

これを使ってデータ取得と送信のスクリプトを作ってみます。
できたものがこちらです。

lambda_function.py

from splunk_http_event_collector import http_event_collector
import requests
import os
import json
import logging
import sys
import datetime

def lambda_handler(event, context):
    # init logging config, this would be job of your main code using this class.
    logging.basicConfig(format='%(asctime)s %(name)s %(levelname)s %(message)s', datefmt='%Y-%m-%d %H:%M:%S %z')
    
    # Create event collector object, default SSL and HTTP Event Collector Port
    # HEC URL、トークンは環境変数化。[設定]タブ > [環境変数]から環境変数を作成できる。
    http_event_collector_key = os.environ["SPLUNK_HEC_TOKEN"]
    http_event_collector_host = os.environ["SPLUNK_HEC_URL"]
    
    testevent = http_event_collector(http_event_collector_key, http_event_collector_host)
    
    # perform a HEC reachable check
    hec_reachable = testevent.check_connectivity()
    if not hec_reachable:
        sys.exit(1)
    
    # Set to pop null fields.  Always a good idea
    testevent.popNullFields = True
    # set logging to DEBUG for example
    testevent.log.setLevel(logging.DEBUG)
    
    # Start event payload and add the metadata information
    payload = {}
    payload.update({"index":"aws_lambda"})
    payload.update({"sourcetype":"horoscope"})
    payload.update({"source":context.function_name})
    payload.update({"host":"lambda"})
    
    # 本日の占い取得。LambdaはUTCで動作するためJSTに変換する。
    dt = datetime.datetime.utcnow() + datetime.timedelta(hours=9)
    url = "http://api.jugemkey.jp/api/horoscope/free/{}/{}/{}".format(dt.year, dt.month, dt.day)
    event = requests.get(url).json()

    # Splunk HECへ送信
    payload.update({"event":event})
    testevent.sendEvent(payload)

さて、これを実行してみます。[Test] ボタンからテストできます。 [Deploy] で更新内容を反映してからテストしてみましょう。

Splunkで見てみましょう。無事取り込めました。すごい！

いい感じに動いたので、次はこのスクリプトを定期実行したいと思います。

3. CloudWatchでLambdaを定期実行化

前回記事と同様に、CloudWatchで定期実行化します。
早速設定してみましょう。

1. 設定 > トリガーから [トリガーを追加] をクリック
   
2. EventBridge (CloudWatch Events) を選択
   
3. ルールタイプで「スケジュール式」を選択。cron(0 15 ? * * *)で毎日日本時間の0時0分に取得するようにします。
4. [追加] をクリックすれば定期実行化が完了

これで毎日、その日の運勢を取得してSplunkに送ってくれるようになりました。

4. 今日の運勢をダッシュボード化

せっかくなのでダッシュボードで本日の運勢を見れるようにしてみましょう。

SPLは以下です。

SPL

index=aws_lambda sourcetype=horoscope earliest=@d
| table horoscope*
| foreach horoscope.*.* [rename <<FIELD>> as <<MATCHSEG2>>]
| mvexpand color
| streamstats count
| eval content=mvindex(content,count - 1)
| eval item=mvindex(item,count - 1)
| eval job=mvindex(job,count - 1)
| eval love=mvindex(love,count - 1)
| eval money=mvindex(money,count - 1)
| eval rank=mvindex(rank,count - 1)
| eval sign=mvindex(sign,count - 1)
| eval total=mvindex(total,count - 1)
| fields - count day
| table rank sign content item color job love money total
| sort rank
| rename rank as "ランキング", sign as "星座", content as "占い", "item" as "ラッキーアイテム", "color" as "ラッキーカラー", "job" as "仕事運", love as "恋愛運", money as "金運", total as "総合運"

ダッシュボードにします。

なんならトップページに載せてしまいましょう。

まとめ

• AWS LambdaでPythonを使用し、RESTでデータ取得、SplunkにHEC送信できるようになりました。
• レイヤーを使えば任意のモジュールをアップロードし呼び出せますのでお好みのモジュールで色々なことができそうです。