Function Compute:Simple Log Service トリガー

ユースケース

• データクレンジングと処理

  SLS を使用して、ログを迅速に収集、処理、クエリ、分析します。

  

• データ転送

  ログデータをクラウド上のビッグデータ製品やサードパーティサービスなどの宛先に転送し、それらの間にデータパイプラインを構築します。

  

仕組み

SLS トリガーは Function Compute の 1 つの SLS ETL ジョブに対応します。SLS ETL ジョブを作成すると、SLS は ETL ジョブ設定に基づいてタイマーを開始します。タイマーは Logstore のシャード情報をポーリングします。新しいデータが書き込まれると、SLS は関数イベントとして <shard_id, begin_cursor, end_cursor> のトリプレットを生成し、関数を呼び出します。

ストレージシステムがアップグレードされると、新しいデータが書き込まれていなくてもカーソルの変更が発生する場合があります。この場合、各シャードはデータなしでさらに 1 回トリガーされます。関数内のカーソルを使用してシャードデータの取得を試みることができます。データが返されない場合、その呼び出しは空のトリガーのため、関数内で無視できます。詳細については、「カスタム関数開発ガイド」をご参照ください。

トリガーメカニズムは時間ベースです。たとえば、ETL ジョブのトリガー間隔を 60 秒に設定し、データが Logstore の Shard0 に継続的に書き込まれる場合、シャードは 60 秒ごとに関数呼び出しをトリガーします。シャードに新しいデータが書き込まれない場合、呼び出しはトリガーされません。各呼び出しの入力は、過去 60 秒間のカーソル範囲です。関数では、カーソルに基づいて Shard0 データを読み取り、さらに処理できます。

トリガーの頻度には、次の特徴があります。

• 各シャードは個別にトリガーされます。Logstore で確認できる呼び出しの総数は多く見えるかもしれませんが、各シャードの実際のトリガー時間は設定された間隔と一致します。たとえば、Logstore に 10 個のシャードがある場合、トリガーの遅延なしでリアルタイムデータ処理を行うと、60 秒ごとに 10 回の関数呼び出しが発生します。

• 単一シャードのトリガー間隔は、毎回処理されるデータの時間範囲と同じです。トリガー間隔が 60 秒であると仮定します。関数が実行されると、トリガー間隔は次の 2 つのケースに分類されます。

  • トリガー遅延なし: 関数はスケジュールどおりに 60 秒ごとにトリガーされ、処理されるデータ範囲は [now -60s, now) です。

  • トリガーの遅延：SLS シャードの現在の処理位置が、最新の書き込みデータより 10 秒以上遅れている場合に発生します。トリガーは追いつこうとし、2 秒に 1 回起動することがあります。各呼び出しは、引き続き 60 秒のウィンドウを処理します。

  

データ処理関数

SLS トリガーが呼び出す関数は、次のいずれかのタイプになります。

• テンプレート関数

  詳細については、「aliyun-log-fc-functions」をご参照ください。

• カスタム関数

  設定形式は、関数の実装方法によって異なります。詳細については、「ETL 関数開発ガイド」をご参照ください。

前提条件

• Function Compute

  • イベントトリガー関数の作成

• Simple Log Service (SLS)

  • プロジェクトと Logstore の作成

  • 1 つのプロジェクトと 2 つの Logstore を作成します。1 つの Logstore は収集されたログを保存します。Function Compute は増分ログをトリガーとして使用するため、ログがこの Logstore に継続的に収集されるようにしてください。もう 1 つの Logstore は、SLS トリガーによって生成されたログを保存します。

  プロジェクトは、Function Compute サービスと同じリージョンにある必要があります。

ステップ1：SLS トリガーの作成

SLS トリガーを設定して、更新されたデータを定期的に取得し、関数を呼び出して SLS Logstore 内のデータを増分消費できます。関数では、データクレンジングや処理などのカスタム処理タスクを実行し、サードパーティサービスにデータを配信します。この例では、ログデータを取得して出力する方法のみを示します。データ処理に使用される関数は、SLS が提供するテンプレートまたはカスタム関数のいずれかです。次の手順では、カスタム関数を使用します。

1. Function Compute コンソールにログインします。左側のナビゲーションペインで、[Functions] > [Functions] を選択します。

2. 上部のナビゲーションバーで、リージョンを選択します。[Functions] ページで、対象の関数をクリックします。

3. [Function Details] ページで、[Triggers] タブをクリックし、[Create Trigger] をクリックします。[Create Trigger] パネルで、[Trigger Type] を [Log Service] に設定し、その他のパラメーターを設定して [OK] をクリックします。

トリガーが作成されると、[Triggers] タブに表示されます。トリガーを変更または削除するには、『Function Compute User Guide』の「Manage Triggers」をご参照ください。

ステップ2：権限の設定

関数ロールは、関数が呼び出されるときに必要な SLS 権限を提供します。

1. [Function Details] ページで、[Configurations] タブをクリックします。[Advanced Settings] セクションで、[Modify] をクリックします。[Advanced Settings] パネルで、[Function Role] を選択します。

   • 関数がログデータの読み取りのみを行う場合は、デフォルトで SLS の読み取り専用権限を持つデフォルトロール [AliyunFCServerlessDevsRole] を使用できます。

   • 関数が SLS への読み取り専用以上の権限を必要とする場合は、次の 2 つの要件を満たすカスタム RAM ロールを作成します。

     a. RAM ロールを作成するときは、[Select Trusted Entity] を [Alibaba Cloud Service] に設定し、[Trusted Service] を [Function Compute] に設定します。詳細については、「信頼できる Alibaba Cloud サービス用の RAM ロールを作成」をご参照ください。

     b. RAM ロールに、関数が必要とする SLS 権限を付与します。詳細については、「カスタム RAM ポリシーの例」をご参照ください。

2. [Deploy] をクリックします。

ステップ3：関数のデプロイとログの表示

1. [Function Details] ページの [Code] タブで、コードエディターにコードを記述し、[Deploy] をクリックします。

   この例では、次の処理を行う Python 関数をデプロイします。

   • event から、endpoint、projectName、logstoreName、beginCursor などの SLS イベント情報を取得します。

   • context から認証情報 accessKeyId、accessKeySecret、および securityToken を取得します。

   • 取得した情報に基づいて SLS クライアントを初期化します。

   • ソース Logstore から指定されたカーソル位置のログデータを取得します。

     次のサンプルコードは、ほとんどのログ処理シナリオの開始テンプレートとして使用できます。

   """
   このサンプルコードは、主に次の処理を実行します。
   * イベントから SLS 処理関連情報を取得
   * SLS クライアントを初期化
   * ソース Logstore からログをプル
   """
   import logging
   import json
   from aliyun.log import LogClient
   
   logger = logging.getLogger()
   
   
   def handler(event, context):
       # アクセスキーは 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
   
       # イベントをオブジェクトにパースします
       event_obj = json.loads(event.decode())
       print("The content in event entity is: ", event_obj)
   
       # event.source からログプロジェクト名、Logstore 名、SLS のエンドポイント、開始カーソル、終了カーソル、shardId を取得します
       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']
   
       # SLS のクライアントを初期化します
       client = LogClient(endpoint=endpoint, accessKeyId=access_key_id, accessKey=access_key_secret, securityToken=security_token)
   
       # この例では、カーソル [begin_cursor, end_cursor) 内のソース Logstore からデータを読み取ります。これには、呼び出しをトリガーしたすべてのログが含まれます
       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'

2. [Function Details] ページで、[Logs] > [Function Logs] を選択して、関数の実行時に取得された最新のデータを表示します。「The logging feature is not enabled for the current function.」と表示された場合は、[Enable] をクリックします。

   これで、SLS トリガーの設定は完了です。トリガー間隔が経過し、新しいデータがソース Logstore に書き込まれると、トリガーが関数を呼び出し、関数ログが [Function Logs] ページに表示されます。ステップ1で [トリガーログ] パラメーターに指定した Logstore を確認して、トリガーが起動したかを確認することもできます。コンソールでコードをデバッグするには、次の手順を実行します。

(オプション) ステップ4：シミュレートされたイベントでの関数のテスト

1. [Function Details] ページの [Code] タブで、[Test Function] の右側にある アイコンをクリックし、ドロップダウンリストから [Configure Test Parameters] を選択します。

2. [Configure Test Parameters] パネルで、[Create New Test Event] または [Modify Existing Test Event] を選択し、イベント名とイベント内容を入力してから [OK] をクリックします。新しいテストイベントを作成する場合は、[Log Service] イベントテンプレートを選択することを推奨します。テストデータの詳細については、「event パラメーター」をご参照ください。

3. シミュレートされたイベントを設定した後、[Test Function] をクリックします。

   呼び出しが完了すると、[Code] タブの上に結果が表示されます。

制限事項

単一のプロジェクトに設定できる SLS トリガーの数は、そのプロジェクトに存在する Logstore の数の 5 倍を超えることはできません。

各 Logstore につき、5 つ以下の SLS トリガーを設定してください。そうしないと、Function Compute へのデータ転送の効率に影響を及ぼす可能性があります。

入力パラメーター

{
    "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
}

よくある質問