Function Compute:Simple Log Service トリガー

シナリオ

• データクレンジングと変換

  Simple Log Service を使用すると、ログを迅速に収集、処理、クエリ、分析できます。

• データシッピング

  Simple Log Service を使用すると、データを宛先にシッピングし、クラウド上のビッグデータサービス間にデータパイプラインを構築できます。

データ処理のための関数

関数の種類

• テンプレート関数

  詳細については、GitHub の aliyun-log-fc-functions ページをご参照ください。

• カスタム関数

  関数のフォーマットは、関数の実装に関連しています。詳細については、「カスタム関数の作成」をご参照ください。

トリガーメカニズム

制限事項

Simple Log Service プロジェクト内の Logstore の数の最大 5 倍をプロジェクトに関連付けることができます。

サンプルシナリオ

Simple Log Service トリガーを設定して、定期的に更新されたデータを取得し、関数を呼び出すことができます。Simple Log Service トリガーは、Logstore から増分的にデータを使用したいシナリオに適しています。関数を呼び出して、データクレンジングタスクやデータ処理タスクなどのカスタム処理タスクを実行し、データをサードパーティサービスにシッピングできます。この例では、ログデータを取得して表示する方法のみを示します。

前提条件

• Function Compute

  • 関数が作成されていること。詳細については、「イベント関数の作成」をご参照ください。

• Simple Log Service

  • Simple Log Service プロジェクトと 2 つの Logstore が作成されていること。詳細については、「リソース管理の概要」をご参照ください。

    1 つの Logstore は、増分ログに基づいてトリガーされる Function Compute によって収集されたログを保存するために使用されます。したがって、ログが継続的に収集されることを確認する必要があります。もう 1 つの Logstore は、Simple Log Service トリガーによって生成されたログを保存するために使用されます。

入力パラメーター

• event

  Simple Log Service トリガーが実行された後、イベントデータがランタイムに渡されます。ランタイムはイベントを JSON オブジェクトに変換し、そのオブジェクトを関数の入力パラメーター event に渡します。サンプルコード:

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

  次の表にパラメーターを示します。

  パラメーター	説明
  parameter	トリガーの作成時に設定する 呼び出しパラメーター パラメーターの値。
  source	関数が Simple Log Service から読み取るログブロック情報。 endpoint: Simple Log Service プロジェクトが存在する Alibaba Cloud リージョンのエンドポイント。 projectName: Simple Log Service プロジェクトの名前。 logstoreName: Function Compute が使用する Logstore の名前。現在のトリガーは Logstore のデータをサブスクライブし、カスタム処理のために定期的に Function Compute にデータを送信します。 shardId: Logstore 内の特定のシャードの ID。 beginCursor: データ使用が開始されるオフセット。 endCursor: データ使用が終了するオフセット。 説明 GetCursor 操作を呼び出して 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 タイムスタンプ。単位: 秒。

• context

  Function Compute が関数を実行すると、システムはコンテキストオブジェクトを関数の入力パラメーター context に渡します。このオブジェクトには、呼び出し、サービス、関数、および実行環境に関する情報が含まれています。

  この例では、context.credentials オブジェクトを使用してキー情報を取得します。オブジェクトでサポートされているパラメーターの詳細については、「コンテキスト」をご参照ください。

ステップ 1: Simple Log Service トリガーの作成

1. Function Compute コンソールにログインします。左側のナビゲーションウィンドウで、[関数管理] > [関数] を選択します。

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

3. 関数詳細ページで、[トリガー] タブをクリックし、[トリガーの作成] をクリックします。

4. [トリガーの作成] パネルでパラメーターを設定し、[OK] をクリックします。

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

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

ステップ 2: 権限の設定

1. 関数の[関数詳細]ページで、[設定]タブをクリックします。 左側のナビゲーションウィンドウで[権限]をクリックし、次に[変更]をクリックします。 [権限] パネルで、[関数ロール]を指定します。

   • デフォルトのロール AliyunFCServerlessDevsRole を使用できます。デフォルトでは、このロールには Simple Log Service に対する読み取り専用権限があります。

   • カスタム RAM ロールを使用することもできます。カスタム RAM ロールは、次の 2 つの要件を満たす必要があります。

     1. Resource Access Management (RAM) コンソールでロールを作成するときは、Alibaba Cloud サービスを選択し、[信頼できるサービスの選択] ドロップダウンリストから Function Compute を選択する必要があります。詳細については、「信頼できる Alibaba Cloud サービスの RAM ロールを作成する」をご参照ください。

     2. 関数の特定の要件に基づいて、RAM ロールに Simple Log Service に対する必要な権限を付与する必要があります。詳細については、「カスタムポリシーを使用して RAM ユーザーに権限を付与する例」をご参照ください。

2. [デプロイ] をクリックします。

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

1. 関数詳細ページの [コード] タブで、コードエディターにコードを記述し、[デプロイ] をクリックします。

   この例では、次の機能を実装するために Python 関数がデプロイされます。

   • event パラメーターから、endpoint、projectName、logstoreName、beginCursor などの Simple Log Service イベントトリガーに関する情報を取得します。

   • context パラメーターから、accessKeyId、accessKey、securityToken などの権限付与情報を取得します。

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

   • 指定されたカーソルに基づいてソース Logstore からログデータを取得します。

   説明

   次のサンプルコードは、ほとんどの論理ログを抽出する方法の例を示しています。

   """
   このサンプルコードは、次の機能を実装するために使用されます:
   * イベントパラメーターを解析して、Simple Log Service イベントのトリガー情報を取得します。
   * 上記の情報に基づいて Simple Log Service クライアントを初期化します。
   * ソース Logstore からリアルタイムログを取得します。
   
   
   このサンプルコードは主に次のことを行います:
   * イベントから SLS 処理関連情報を取得します
   * SLS クライアントを初期化します
   * ソースログストアからログをプルします
   
   """
   #!/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):
   
       # context.credentials を使用してキーに関する情報を取得します。
       # アクセスキーは 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 名、Simple Log Service プロジェクトにアクセスするためのエンドポイント、開始カーソル、終了カーソル、およびシャード ID。
       # event.source からログプロジェクト名、ログストア名、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']
   
       # Simple Log Service クライアントを初期化します。
       # sls のクライアントを初期化します
       client = LogClient(endpoint=endpoint, accessKeyId=access_key_id, accessKey=access_key_secret, securityToken=security_token)
   
       # ソース Logstore で開始カーソルと終了カーソルから始まるログを読み取ります。この例では、指定されたカーソルには関数呼び出しのすべてのログが含まれます。
       # この例では、カーソル内のソースログストアからデータを読み取ります: [begin_cursor, end_cursor)。これには、呼び出しをトリガーするすべてのログが含まれます
       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. [関数詳細] ページで、[ログ] タブをクリックします。左側のナビゲーションウィンドウで [関数ログ] をクリックし、関数が実行されたときに取得された最新のデータを表示します。「現在の関数のロギング機能が有効になっていません。」 というメッセージが表示された場合は、[有効にする] をクリックします。

上記の手順を実行すると、Simple Log Service トリガーが設定されます。Function Compute コンソールでコードをデバッグする場合は、次の手順に進みます。

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

1. 関数詳細ページの[コード]タブで、[関数のテスト] の横にある アイコンをクリックし、ドロップダウンリストから[テストパラメーターの設定]を選択します。

2. [テストパラメーターの設定] パネルで、[新しいテストイベントの作成] または [既存のテストイベントの変更] タブをクリックし、イベント名とイベント内容を入力して、[OK] をクリックします。[新しいテストイベントの作成] を選択した場合は、[イベントテンプレート] に Log Service を選択することをお勧めします。テストデータの設定方法の詳細については、このトピックの「event」セクションをご参照ください。

3. シミュレートされたイベントが設定された後、[関数のテスト] をクリックします。

   関数テストが完了したら、[コード] タブで結果を表示できます。

よくある質問

• 新しいログが生成されても Simple Log Service トリガーが関数の実行をトリガーしない場合はどうすればよいですか？

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

  • トリガーに関連付けられている Logstore で増分データの変更が発生したかどうかを確認します。シャードデータが変更されると、関連する関数がトリガーされます。

  • トリガーログと関数の操作ログに例外が見つかるかどうかを確認します。

• Simple Log Service トリガーの実行頻度が予想よりも高いのはなぜですか？

  関数は各シャードに対して個別にトリガーされます。Logstore 内のシャードに対して関数がトリガーされる回数が多くても、各シャードに対して関数がトリガーされる間隔は、指定されたトリガー間隔と一致する可能性があります。

  シャードに対して関数がトリガーされるトリガー間隔は、データ変換に指定された時間間隔と同じです。関数がトリガーされると、遅延が発生する可能性があります。これにより、トリガー間隔が予想よりも大きくなる可能性があります。次のリストは、指定されたトリガー間隔が 60 秒の 2 つのシナリオを示しています。

  • シナリオ 1: 関数がトリガーされ、遅延は存在しません。関数は 60 秒間隔でトリガーされ、次の時間範囲で生成されたデータを変換します: [now -60s, now)。

    説明

    関数は各シャードに対して個別にトリガーされます。Logstore に 10 個のシャードが含まれており、関数がトリガーされたときに遅延が存在しない場合、関数は 60 秒間隔で 10 回トリガーされ、リアルタイムでデータを変換します。

  • シナリオ 2: 関数がトリガーされ、遅延が存在します。Simple Log Service シャードのデータが変換された時点と、最新のデータが Simple Log Service に書き込まれた時点との時間差が 10 秒を超えています。この場合、トリガーは間隔を短縮します。たとえば、関数は 2 秒間隔でトリガーされ、60 秒以内に生成されたデータを変換できます。

• 「denied by sts or ram, action: log:GetCursorOrData, resource: ****」というエラーメッセージが返された場合はどうすればよいですか？

  このエラーメッセージが関数ログに表示された場合、関連する権限が関数に設定されていないか、権限が誤って設定されている可能性があります。詳細については、このトピックの「ステップ 2: 権限の設定」セクションをご参照ください。