ビジネス要件に基づいてログデータを消費し、ログデータを変換するようにFunction Computeトリガーを設定する - Simple Log Service

Simple Log Serviceでは、Function Computeを使用してストリーミングデータを変換できます。 Function Computeトリガーを設定して、データの更新を検出し、関数を呼び出すことができます。これにより、Logstoreの増分データを使用してデータを変換できます。テンプレート関数またはユーザー定義関数を使用して、データを変換できます。

前提条件

Simple Log Serviceは関数を呼び出す権限があります。 [クラウドリソースアクセス権限付与] ページで権限付与を実行できます。
Function Compute
- イベント関数の作成
シンプルなLog Service
- プロジェクトとログストアの作成
  1つのSimple Log Serviceプロジェクトと2つのログストアを作成する必要があります。一方のログストアは収集されたログを保存するために使用され (Function Computeは増分ログに基づいてトリガーされるため、継続的なログ収集が必要です) 、もう一方のログストアはSimple log Serviceトリガーによって生成されたログを保存するために使用されます。
重要
Simple Log ServiceプロジェクトとFunction Computeサービスは同じリージョンに属している必要があります。

制限事項

1つのログプロジェクトに関連付けられているSimple Log Serviceトリガーの数は、プロジェクト内のログストアの数の5倍を超えてはなりません。

説明

各ログストアに5つ以下のSimple Log Serviceトリガーを設定することを推奨します。そうしないと、データがFunction Computeに効率的に出荷されない可能性があります。

シナリオ

データのクレンジングと変換
Simple Log Serviceを使用すると、効率的にログを収集、変換、クエリ、および分析できます。
データ配信
Simple Log Serviceを使用して、指定した宛先にデータを送信できます。このシナリオでは、Simple Log Serviceはクラウド内のビッグデータサービスのデータチャネルとして機能します。

データ変換関数

関数タイプ
- テンプレート関数
  詳細については、「aliyun-log-fc-functions」をご参照ください。
- ユーザー定義関数
  関数形式は、関数の実装に関連しています。詳細については、「カスタム関数の作成」をご参照ください。
関数呼び出しメカニズム
Function Computeトリガーは、関数を呼び出すために使用されます。 Simple Log ServiceでLogstoreのFunction Computeトリガーを作成すると、トリガー設定に基づいてLogstoreのシャードを監視するタイマーが起動されます。新しいデータがLogstoreに書き込まれると、<shard_id,begin_cursor,end_cursor > 形式の3組のデータレコードが関数イベントとして生成されます。次に、関数が呼び出される。
説明
新しいデータがLogstoreに書き込まれず、ストレージシステムが更新された場合、カーソル情報が変更される可能性があります。その結果、各シャードに対して関数が呼び出されますが、データは変換されません。この場合、カーソル情報を使用してシャードからデータを取得できます。データが取得されない場合、関数が呼び出されますが、データは変換されません。関数呼び出しは無視できます。詳細については、「カスタム関数の作成」をご参照ください。
Function Computeトリガーは、時間に基づいて関数を呼び出します。たとえば、LogstoreのFunction Computeトリガーの呼び出し間隔は60秒に設定されています。 Shard 0にデータが連続して書き込まれている場合、60秒間隔で関数が呼び出され、前の60秒のカーソル範囲に基づいてデータが変換されます。

手順1: シンプルなLog Serviceトリガーの作成

Function Computeコンソールにログインします。左側のナビゲーションウィンドウで、[関数] をクリックします。
上部のナビゲーションバーで、リージョンを選択します。 [関数] ページで、管理する関数をクリックします。
機能の詳細ページで、[設定] タブをクリックします。左側のナビゲーションウィンドウで、[トリガー] をクリックします。次に、[トリガーの作成] をクリックします。

トリガーを作成するには、パネルに必要な情報を入力し、確認.

設定アイテム	操作	例
トリガータイプ	[Simple Log Service SLS] を選択します。	シンプルなLog Service SLS
名前	カスタムトリガー名を入力します。名前を入力しない場合、Function Computeは自動的にトリガー名を生成します。	log_trigger
バージョンまたはエイリアス	デフォルト値はLATESTです。別のバージョンまたはエイリアスのトリガーを作成する場合は、関数の詳細ページの右上隅にあるバージョンまたはエイリアスに切り替える必要があります。バージョンとエイリアスの詳細については、「バージョン管理」および「エイリアス管理」をご参照ください。	LATEST
ログプロジェクト	使用するログプロジェクトを選択します。	aliyun-fc-cn-hangzhou-2238f0df-a742-524f-9f90-976ba457 ****
ログストア	使用するLogstoreを選択します。現在のトリガーは、このLogstoreから関数サービスへのデータを定期的にサブスクライブしてカスタム処理を行います。	関数ログ
トリガー間隔	Simple Log Serviceが関数をトリガーする時間間隔を入力します。有効な値: [3,600] 。単位は秒です。デフォルト値：60。	60
再試行	1回のトリガー操作で許可される再試行の最大数を入力します。有効な値: [0,100] 。デフォルト値：3 説明 status=200で、ヘッダーの`X-Fc-Error-Type`パラメーターの値が`UnhandledInvocationError`または`HandledInvocationError`でない場合、実行は成功します。それ以外の場合は、実行が失敗し、再試行がトリガーされます。 `X-Fc-Error-Type`パラメーターの詳細については、「データの戻り」をご参照ください。関数の実行に失敗した場合、関数が正常に実行されるまで、現在の要求が再試行されます。再試行の回数は、このパラメーターの値に従います。このパラメーターの値に達しても関数が失敗した場合、システムは間隔を広げて指数バックオフモードでリクエストを再試行します。	3
トリガーログ	Simple Log Serviceが関数を呼び出したときに生成されるログを保存するLogstoreを選択します。	function-log2
呼び出しパラメーター	カスタムパラメータを渡す場合は、ここで設定できます。パラメーターは、イベントのパラメーターパラメーターとして関数に渡されます。このパラメーターの値は、JSON形式の文字列である必要があります。このパラメーターはデフォルトでは何も指定されていません。	なし
ロール名	[AliyunLogETLRole] を選択します。説明このタイプのトリガーを初めて作成する場合は、[確認] をクリックした後に表示されるダイアログボックスで [今すぐ許可] を選択する必要があります。	AliyunLogETLRole

トリガーが作成されると、[トリガー] タブに表示されます。トリガーを変更または削除するには、「トリガー管理」をご参照ください。

手順2: 権限の設定

[関数の詳細] ページで、[設定] > [権限] を選択し、[編集] をクリックします。表示される権限ページで、[関数ロール] を選択します。
- デフォルトのロールAliyunFCServerlessDevsRoleを使用できます。このロールには、デフォルトでSimple Log Serviceに対する読み取り専用権限があります。
- カスタムRAMロールを作成することもできます。カスタムRAMロールは、次の2つの要件を満たす必要があります。
  1. RAMロールを作成するには、Alibaba Cloud Serviceを選択し、信頼できるサービスとしてFunction Computeを選択します。詳細については、「」をご参照ください。信頼できるAlibaba CloudサービスのRAMロールの作成.
  2. 関数の特定の要件に基づいて、RAMロールにSimple Log Serviceで必要な権限を付与する必要があります。詳細については、「カスタムポリシーを使用してRAMユーザーに権限を付与する例」をご参照ください。
設定が完了したら、[デプロイ] をクリックします。

ステップ3: 関数をデプロイし、印刷されたログを表示する

に移動します。Navigate to theコード関数の詳細ページのタブで、エディターにコードを入力し、コードのデプロイ変更を適用します。

この例では、Python関数をデプロイして次の機能を実装します。

eventから、endpoint、projectName、logstoreName、beginCursorなどのSimple Log Serviceイベントトリガー関連情報を取得します。
accessKeyId、accessKey、securityTokenなどの権限情報をcontextから取得します。
取得した情報に基づいてSLSクライアントを初期化します。
ソースログストアの指定されたカーソル位置からログデータを取得します。

説明

次のサンプルコードは、ほとんどの論理ログを抽出する方法に関するテンプレートを提供します。

"""
This sample code is mainly doing the following things:
* Get Simple Log Service processing related information from event.
* Initiate the Simple Log Service client.
* Pull logs from the source logstore.

"""
#!/usr/bin/env python
# -*- coding: utf-8 -*-

import logging
import json
import os
from aliyun.log import LogClient


logger = logging.getLogger()


def handler(event, context):

    # Access keys can be fetched through context.credentials.
    print("The content in context entity is: ", context)
    creds = context.credentials
    access_key_id = creds.access_key_id
    access_key_secret = creds.access_key_secret
    security_token = creds.security_token

    # Parse event in object.
    event_obj = json.loads(event.decode())
    print("The content in event entity is: ", event_obj)

    # Get the project name, logstore name, Simple Log Service endpoint, begin cursor, end cursor, and shard Id from event.source.
    source = event_obj['source']
    log_project = source['projectName']
    log_store = source['logstoreName']
    endpoint = source['endpoint']
    begin_cursor = source['beginCursor']
    end_cursor = source['endCursor']
    shard_id = source['shardId']

    # Initialize the client of Simple Log Service.
    client = LogClient(endpoint=endpoint, accessKeyId=access_key_id, accessKey=access_key_secret, securityToken=security_token)

    # Read data from the source logstore within cursor: [begin_cursor, end_cursor) in the example, which contains all the logs trigger the invocation
    while True:
      response = client.pull_logs(project_name=log_project, logstore_name=log_store,
                                shard_id=shard_id, cursor=begin_cursor, count=100,
                                end_cursor=end_cursor, compress=False)
      log_group_cnt = response.get_loggroup_count()
      if log_group_cnt == 0:
        break
      logger.info("get %d log group from %s" % (log_group_cnt, log_store))
      logger.info(response.get_loggroup_list())

      begin_cursor = response.get_next_cursor()

    return 'success'

関数の詳細ページで、[ログ] > [関数ログ] を選択して、関数の実行中に取得した最新のデータを表示します。現在の機能でログ機能が有効になっていない場合は、[有効にする] をクリックします。

(オプション) ステップ4: シミュレートされたイベントを使用して関数をテストする

関数の詳細ページの [コード] タブで、[テスト関数] の隣のアイコンをクリックし、ドロップダウンリストから [テストパラメーターの設定] を選択します。
[テストパラメーターの設定] パネルで、[新しいテストイベントの作成] または [既存のテストイベントの変更] タブをクリックし、イベント名とイベント内容を入力し、[OK] をクリックします。新しいテストイベントを作成するときは、イベントテンプレートとしてSimple Log Serviceを使用します。テストデータの構成について詳しくは、「event」をご参照ください。
仮想イベントを設定した後、クリックテスト機能.
関数の実行後、[テスト] タブで結果を表示できます。

入力パラメーターの説明

`イベント`

Simple Log Serviceトリガーがアクティブになると、イベントデータがランタイムに送信されます。次に、ランタイムはこのデータをJSONオブジェクトに変換し、関数の入力パラメーターイベントに渡します。 JSON形式は次のとおりです。

{
    "parameter": {},
    "source": {
        "endpoint": "http://cn-hangzhou-intranet.log.aliyuncs.com",
        "projectName": "fc-test-project",
        "logstoreName": "fc-test-logstore",
        "shardId": 0,
        "beginCursor": "MTUyOTQ4MDIwOTY1NTk3ODQ2Mw==",
        "endCursor": "MTUyOTQ4MDIwOTY1NTk3ODQ2NA=="
    },
    "jobName": "1f7043ced683de1a4e3d8d70b5a412843d81****",
    "taskId": "c2691505-38da-4d1b-998a-f1d4bb8c****",
    "cursorTime": 1529486425
}

以下はパラメータの説明です。

パラメーター	説明
パラメーター	トリガーの設定時に入力した呼び出しパラメーターの値。
ソース	関数が読み取るログブロック情報。 endpoint: ログプロジェクトが配置されているAlibaba Cloudリージョン。 projectName: ログプロジェクトの名前。 logstoreName: Function Computeが使用するログストアの名前。現在のトリガーは、このLogstoreから関数サービスへのデータを定期的にサブスクライブしてカスタム処理を行います。 shardId: ログストア内の特定のシャード。 beginCursor: データ消費が開始される位置。 endCursor: データ消費が停止する位置。説明関数をデバッグするときに、GetCursor - Queryカーソルby time APIを呼び出してbeginCursorとendCursorを取得し、上記の例に基づいてテスト用の関数イベントを構築できます。
jobName	Simple Log Service ETLジョブの名前。関数用に設定されたSimple Log Serviceトリガーは、Simple Log Service ETLジョブに対応しています。このパラメーターは、Function Computeによって自動的に生成されます。設定する必要はありません。
taskId	ETLジョブの場合、taskIdは決定論的関数呼び出し識別子です。このパラメーターは、Function Computeによって自動的に生成されます。設定する必要はありません。
cursorTime	Simple log Serviceに最後のログが到着した時刻のUnixタイムスタンプ。単位は秒です。

コンテキスト
Function Computeが関数を実行すると、関数の入力パラメーターcontextにcontextオブジェクトを提供します。このオブジェクトには、呼び出し、サービス、関数、、および実行環境に関する詳細が含まれます。
このトピックでは、context.credentialsを使用してキー情報を取得する方法について説明します。その他のフィールドの詳細については、「コンテキスト」をご参照ください。

次のステップ

クエリトリガログ
トリガーログが保存されているログストアのインデックスを作成し、トリガーの実行結果を表示できます。詳細については、「インデックスの作成」をご参照ください。
関数の操作ログを表示する
CLIを使用して、関数呼び出しに関する詳細情報を表示できます。詳細については、「関数呼び出しログの表示」をご参照ください。

よくある質問

新しいログが生成されたときにSimple Log Serviceトリガーが関数の実行をトリガーしない場合はどうすればよいですか。

次の方法を使用して問題をトラブルシューティングできます。

Function Computeトリガーが設定されているLogstoreに新しいデータが書き込まれているかどうかを確認します。新しいデータがLogstoreに書き込まれると、関数が呼び出されます。
関数のトリガーログと操作ログに例外があるかどうかを確認します。

Simple Log Serviceトリガーによってトリガーされる関数の実行頻度が予想よりも高くなることがあるのはなぜですか。

関数はシャードごとに個別に呼び出されます。 Logstore内のシャードに対して関数が呼び出される回数が多い場合でも、各シャードに対して関数が呼び出される間隔は、指定された呼び出し間隔と一致する可能性があります。

関数がシャードに対して呼び出される呼び出し間隔は、データ変換に指定された時間間隔と同じです。次のリストでは、指定された呼び出し間隔が60秒の2つのシナリオについて説明します。

レイテンシが存在しない: この関数は、指定された60秒間隔で呼び出され、[now -60 s, now) の時間範囲内に生成されたデータを変換します。
説明
各シャードは別々の関数呼び出しをトリガーします。 10個のシャードを持つLogstoreでは、レイテンシの問題がないと仮定すると、関数は60秒間隔で10回呼び出され、リアルタイムでデータを処理します。
レイテンシが存在します。 Simple Log Serviceシャード内のデータが変換された時点と、最新のデータがSimple Log Serviceに書き込まれた時点との時間差が10秒を超えています。この場合、トリガーは通話間隔を短くします。たとえば、関数を2秒間隔で呼び出して、60秒以内に生成されたデータを変換できます。

stsまたはramによって拒否されました。action: log:GetCursorOrData、resource: ****

関数ログで上記のエラーが発生した場合、関数が権限で設定されていないか、権限ポリシーが誤って設定されているためにエラーが発生する可能性があります。詳細については、「手順2: 権限の設定」をご参照ください。

Simple Log Service:Function Computeを使用してログデータを消費する