Simple Log Service:コンシューマーグループによるログの消費

SDK

以下のコードは、コンシューマーグループの作成方法を示しています：

import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.common.ConsumerGroup;
import com.aliyun.openservices.log.exception.LogException;

public class CreateConsumerGroup {
    public static void main(String[] args) throws LogException {
         // この例では、AccessKey ID および AccessKey Secret を環境変数から取得します。
        String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
        String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
        // プロジェクト名を入力します。
        String projectName = "ali-test-project";
        // Logstore 名を入力します。
        String logstoreName = "ali-test-logstore";
        // Simple Log Service のエンドポイントを設定します。この例では中国 (杭州) リージョンのエンドポイントを使用しています。実際のエンドポイントに置き換えてください。
        String host = "https://cn-hangzhou.log.aliyuncs.com";

        // Simple Log Service クライアントを作成します。
        Client client = new Client(host, accessId, accessKey);

        try {
            // コンシューマーグループ名を設定します。
            String consumerGroupName = "ali-test-consumergroup2";
            System.out.println("コンシューマーグループの作成準備完了");

            ConsumerGroup consumerGroup = new ConsumerGroup(consumerGroupName, 300, true);

            client.CreateConsumerGroup(projectName, logstoreName, consumerGroup);

            System.out.println(String.format("コンシューマーグループ %s の作成に成功しました", consumerGroupName));

        } catch (LogException e) {
            System.out.println("LogException e :" + e.toString());
            System.out.println("エラーコード :" + e.GetErrorCode());
            System.out.println("エラーメッセージ :" + e.GetErrorMessage());
            throw e;
        }
    }
}

その他のコード例については、「Java SDK を使用したコンシューマーグループの管理」および「Python 向け Log Service SDK を使用したコンシューマーグループの管理」をご参照ください。

API

コンシューマーグループを作成するには、CreateConsumerGroup 操作を呼び出します。

コンシューマーグループが正常に作成されたかを確認するには、ListConsumerGroup 操作を呼び出します。

CLI

コンシューマーグループを作成するには、create_consumer_group コマンドを実行します。

コンシューマーグループが正常に作成されたかを確認するには、list_consumer_group コマンドを実行します。

SDK

SDK および SPL

1. Maven 依存関係を追加します。

   pom.xml ファイルに、以下の依存関係を追加します：

   <dependency>
     <groupId>com.google.protobuf</groupId>
     <artifactId>protobuf-java</artifactId>
     <version>2.5.0</version>
   </dependency>
   <dependency>
     <groupId>com.aliyun.openservices</groupId>
     <artifactId>loghub-client-lib</artifactId>
     <version>0.6.50</version>
   </dependency>

2. 消費ロジックを作成します。以下のコードは一例です：

   SPLLogHubProcessor.java

   import com.aliyun.openservices.log.common.FastLog;
   import com.aliyun.openservices.log.common.FastLogContent;
   import com.aliyun.openservices.log.common.FastLogGroup;
   import com.aliyun.openservices.log.common.FastLogTag;
   import com.aliyun.openservices.log.common.LogGroupData;
   import com.aliyun.openservices.loghub.client.ILogHubCheckPointTracker;
   import com.aliyun.openservices.loghub.client.exceptions.LogHubCheckPointException;
   import com.aliyun.openservices.loghub.client.interfaces.ILogHubProcessor;
   
   import java.util.List;
   
   public class SPLLogHubProcessor implements ILogHubProcessor {
       private int shardId;
       // 最終チェックポイント保存時刻を追跡します。
       private long mLastSaveTime = 0;
   
       // プロセッサの初期化時に呼び出されます。
       public void initialize(int shardId) {
           this.shardId = shardId;
       }
   
       // 主要なデータ消費ロジック。このメソッド内で発生するすべての例外を処理し、再スローしないでください。
       public String process(List<LogGroupData> logGroups, ILogHubCheckPointTracker checkPointTracker) {
           // 取得したデータを出力します。
           for (LogGroupData logGroup : logGroups) {
               FastLogGroup fastLogGroup = logGroup.GetFastLogGroup();
               System.out.println("Tags");
               for (int i = 0; i < fastLogGroup.getLogTagsCount(); ++i) {
                   FastLogTag logTag = fastLogGroup.getLogTags(i);
                   System.out.printf("%s : %s\n", logTag.getKey(), logTag.getValue());
               }
               for (int i = 0; i < fastLogGroup.getLogsCount(); ++i) {
                   FastLog log = fastLogGroup.getLogs(i);
                   System.out.println("--------\nLog: " + i + ", time: " + log.getTime() + ", GetContentCount: " + log.getContentsCount());
                   for (int j = 0; j < log.getContentsCount(); ++j) {
                       FastLogContent content = log.getContents(j);
                       System.out.println(content.getKey() + "\t:\t" + content.getValue());
                   }
               }
           }
           long curTime = System.currentTimeMillis();
           // 30 秒ごとにチェックポイントをサーバーに保存します。ワーカーが予期せず終了した場合、新しいワーカーは最後のチェックポイントから消費を再開しますが、少量のデータが重複して消費される可能性があります。
           try {
               if (curTime - mLastSaveTime > 30 * 1000) {
                   // <code>true

   false

3. コンシューマーインスタンスファクトリを作成します。以下のコードは一例です：

   SPLLogHubProcessorFactory.java

   import com.aliyun.openservices.loghub.client.interfaces.ILogHubProcessor;
   import com.aliyun.openservices.loghub.client.interfaces.ILogHubProcessorFactory;
   
   class SPLLogHubProcessorFactory implements ILogHubProcessorFactory {
       public ILogHubProcessor generatorProcessor() {
           // コンシューマーインスタンスを生成します。注意：このメソッドは毎回新しい SPLLogHubProcessor オブジェクトを返す必要があります。
           return new SPLLogHubProcessor();
       }
   }

4. コンシューマーを作成し、指定された Logstore からデータを消費するコンシューマースレッドを起動します。以下のコードは一例です：

   Main.java

   import com.aliyun.openservices.loghub.client.ClientWorker;
   import com.aliyun.openservices.loghub.client.config.LogHubConfig;
   import com.aliyun.openservices.loghub.client.exceptions.LogHubClientWorkerException;
   
   public class Main {
       // Simple Log Service のエンドポイント。 実際のエンドポイントに置き換えてください。 この例では、中国 (杭州) リージョンのエンドポイントを使用します。
       private static String Endpoint = "cn-hangzhou.log.aliyuncs.com";
       // プロジェクト名。
       private static String Project = "ali-test-project";
       // Logstore 名。
       private static String Logstore = "ali-test-logstore";
       // コンシューマーグループ名。 コンシューマーグループを事前に作成する必要はありません。 プログラムが実行時に自動的に作成します。
       private static String ConsumerGroup = "ali-test-consumergroup2";
       // 環境変数から AccessKey ID と AccessKey Secret を取得します。
       private static String AccessKeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
       private static String AccessKeySecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
   
   
       public static void main(String[] args) throws LogHubClientWorkerException, InterruptedException {
           // 'consumer_1' はコンシューマー名であり、コンシューマーグループ内で一意である必要があります。 複数のマシン間で Logstore からの消費のバランスを取るには、各コンシューマーにマシンの IP アドレスなどの一意の識別子を使用します。
           // maxFetchLogGroupSize: 1 回のリクエストでフェッチするロググループの最大数。 デフォルト値の使用を推奨します。 変更するには、config.setMaxFetchLogGroupSize(100) を使用します。 値は (0, 1000] の範囲である必要があります。
           LogHubConfig config = new LogHubConfig(ConsumerGroup, "consumer_1", Endpoint, Project, Logstore, AccessKeyId, AccessKeySecret, LogHubConfig.ConsumePosition.BEGIN_CURSOR, 1000);
           // 消費のための Simple Log Service Processing Language (SPL) クエリを指定します。
           config.setQuery("* | where cast(body_bytes_sent as bigint) > 14000");
           ClientWorker worker = new ClientWorker(new SPLLogHubProcessorFactory(), config);
           Thread thread = new Thread(worker);
           // スレッドは ClientWorker を開始します。ClientWorker は Runnable インターフェイスを実装し、自動的に実行されます。
           thread.start();
           Thread.sleep(60 * 60 * 1000);
           // worker.shutdown() を呼び出してコンシューマーインスタンスを停止します。 関連付けられたスレッドは自動的に停止します。
           worker.shutdown();
           // ClientWorker は実行時に非同期タスクを作成します。 シャットダウン後、実行中のタスクが安全に終了するのを待ちます。 30 秒間のスリープが推奨されます。
           Thread.sleep(30 * 1000);
       }
   }

5. Main.java を実行します。

   たとえば、NGINX ログを消費する場合、出力は以下のようになります：

   :    GET
   request_uri    :    /request/path-3/file-7
   status    :    200
   body_bytes_sent    :    3820
   host    :    www.example.com
   request_time    :    43
   request_length    :    1987
   http_user_agent    :    Mozilla/5.0 (Windows NT 6.1) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/41.0.2228.0 Safari/537.36
   http_referer    :    www.example.com
   http_x_forwarded_for    :    192.168.10.196
   upstream_response_time    :    0.02
   --------
   Log: 158, time: 1635629778, GetContentCount: 14
   ......
       category    :    null
       source    :    127.0.0.1
       topic    :    nginx_access_log
       machineUUID    :    null
   Tags
       __receive_time__    :    1635629815
   --------
   Log: 0, time: 1635629788, GetContentCount: 14
   ......
       category    :    null
       source    :    127.0.0.1
       topic    :    nginx_access_log
       machineUUID    :    null
   Tags
       __receive_time__    :    1635629877
   --------
   ......