Simple Log Service:Consumo de logs com um consumer group

SDK

O código abaixo cria um consumer group:

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 {
         // This example obtains the AccessKey ID and AccessKey secret from environment variables.
        String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
        String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
        // Enter the project name.
        String projectName = "ali-test-project";
        // Enter the Logstore name.
        String logstoreName = "ali-test-logstore";
        // Set the endpoint for Simple Log Service. This example uses the endpoint of the China (Hangzhou) region. Replace it with the actual endpoint.
        String host = "https://cn-hangzhou.log.aliyuncs.com";

        // Create a Simple Log Service client.
        Client client = new Client(host, accessId, accessKey);

        try {
            // Set the consumer group name.
            String consumerGroupName = "ali-test-consumergroup2";
            System.out.println("ready to create consumergroup");

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

            client.CreateConsumerGroup(projectName, logstoreName, consumerGroup);

            System.out.println(String.format("create consumergroup %s success", consumerGroupName));

        } catch (LogException e) {
            System.out.println("LogException e :" + e.toString());
            System.out.println("error code :" + e.GetErrorCode());
            System.out.println("error message :" + e.GetErrorMessage());
            throw e;
        }
    }
}

Para exemplos de código sobre como gerenciar consumer groups, consulte Usar o Java SDK para gerenciar consumer groups e Usar o Simple Log Service SDK for Python para gerenciar consumer groups.

API

Para criar um consumer group usando a API, consulte CreateConsumerGroup.

Para verificar se o consumer group foi criado, consulte ListConsumerGroup.

CLI

Para criar um consumer group usando a CLI, consulte create_consumer_group.

Para verificar se o consumer group foi criado, consulte list_consumer_group.

Exemplo 1: Consumo via SDK

Exemplo 2: Consumo via SDK com SPL

1. Adicione as dependências do Maven.

   No arquivo pom.xml, adicione as seguintes dependências:

   <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. Crie uma classe para definir a lógica de processamento dos logs.

   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;
       // Tracks the last time a checkpoint was saved.
       private long mLastSaveTime = 0;
       // Called once upon processor initialization.
       public void initialize(int shardId) {
           this.shardId = shardId;
       }
       // The main logic for consuming data. All exceptions must be handled within this method. Do not throw exceptions directly.
       public String process(List<LogGroupData> logGroups, ILogHubCheckPointTracker checkPointTracker) {
           // Print the fetched data.
           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();
           // Save a checkpoint to the server every 30 seconds. If the worker terminates unexpectedly, the new worker resumes from the last checkpoint, potentially reprocessing a small amount of data.
           try {
               if (curTime - mLastSaveTime > 30 * 1000) {
                   // A 'true' parameter flushes the checkpoint to the server immediately. By default, the in-memory checkpoint is automatically flushed to the server every 60 seconds.
                   checkPointTracker.saveCheckPoint(true);
                   mLastSaveTime = curTime;
               } else {
                   // A 'false' parameter caches the checkpoint locally. It will be flushed to the server by the auto-update mechanism.
                   checkPointTracker.saveCheckPoint(false);
               }
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
           return null;
       }
       // This method is called when the worker is shutting down. You can perform cleanup tasks here.
       public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
           // Save the checkpoint to the server immediately.
           try {
               checkPointTracker.saveCheckPoint(true);
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
       }
   }

3. Crie uma factory para gerar instâncias do seu processador de logs.

   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() {
           // Generate a consumer instance. Note: Each call to the generatorProcessor method should return a new SPLLogHubProcessor object.
           return new SPLLogHubProcessor();
       }
   }

4. Crie uma classe principal para configurar e iniciar a thread worker.

   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 {
       // The endpoint of Simple Log Service. Replace with your actual endpoint.
       private static String Endpoint = "cn-hangzhou.log.aliyuncs.com";
       // The project name. Replace with the name of an existing project.
       private static String Project = "ali-test-project";
       // The Logstore name. Replace with the name of an existing Logstore.
       private static String Logstore = "ali-test-logstore";
       // You do not need to create the consumer group beforehand; the program automatically creates it on the first run.
       private static String ConsumerGroup = "ali-test-consumergroup2";
       // This example retrieves the AccessKey ID and AccessKey Secret from environment variables.
       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" is the consumer name, which must be unique within a consumer group. To balance consumption across multiple machines, you can use unique identifiers like the machine's IP address as the consumer name.
           // maxFetchLogGroupSize sets the maximum number of LogGroups to fetch per request. The default value is usually sufficient. You can adjust it by using config.setMaxFetchLogGroupSize(100). The valid range is (0, 1000].
           LogHubConfig config = new LogHubConfig(ConsumerGroup, "consumer_1", Endpoint, Project, Logstore, AccessKeyId, AccessKeySecret, LogHubConfig.ConsumePosition.BEGIN_CURSOR, 1000);
           // Use setQuery to specify an SPL statement to filter logs during consumption.
           config.setQuery("* | where cast(body_bytes_sent as bigint) > 14000");
           ClientWorker worker = new ClientWorker(new SPLLogHubProcessorFactory(), config);
           Thread thread = new Thread(worker);
           // After the thread starts, the ClientWorker runs automatically. ClientWorker implements the Runnable interface.
           thread.start();
           Thread.sleep(60 * 60 * 1000);
           // Call the worker's shutdown() method to stop the consumer instance and its associated thread.
           worker.shutdown();
           // The ClientWorker creates asynchronous tasks. After shutdown(), wait for these tasks to complete gracefully. A 30-second sleep is recommended.
           Thread.sleep(30 * 1000);
       }
   }

5. Execute o arquivo Main.java.

   Abaixo está um exemplo da saída ao consumir logs do 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
   --------
   ......

Java SDK

1. Visualize o checkpoint de consumo de cada shard. O código abaixo serve como exemplo:

   ConsumerGroupTest.java

   import java.util.List;
   import com.aliyun.openservices.log.Client;
   import com.aliyun.openservices.log.common.Consts.CursorMode;
   import com.aliyun.openservices.log.common.ConsumerGroup;
   import com.aliyun.openservices.log.common.ConsumerGroupShardCheckPoint;
   import com.aliyun.openservices.log.exception.LogException;
   public class ConsumerGroupTest {
       static String endpoint = "cn-hangzhou.log.aliyuncs.com";
       static String project = "ali-test-project";
       static String logstore = "ali-test-logstore";
       static String accesskeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
       static String accesskey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
       public static void main(String[] args) throws LogException {
           Client client = new Client(endpoint, accesskeyId, accesskey);
           // Get all consumer groups in the Logstore. The list is empty if no consumer groups exist.
           List<ConsumerGroup> consumerGroups = client.ListConsumerGroup(project, logstore).GetConsumerGroups();
           for(ConsumerGroup c: consumerGroups){
               // Print the consumer group properties: name, heartbeat timeout, and ordered consumption status.
               System.out.println("Name: " + c.getConsumerGroupName());
               System.out.println("heartbeat timeout: " + c.getTimeout());
               System.out.println("ordered consumption: " + c.isInOrder());
               for(ConsumerGroupShardCheckPoint cp: client.GetCheckPoint(project, logstore, c.getConsumerGroupName()).GetCheckPoints()){
                   System.out.println("shard: " + cp.getShard());
                   // The time is a long integer, accurate to the microsecond.
                   System.out.println("Last checkpoint update time: " + cp.getUpdateTime());
                   System.out.println("Consumer name: " + cp.getConsumer());
                   String consumerPrg = "";
                   if(cp.getCheckPoint().isEmpty())
                       consumerPrg = "Consumption has not started";
                   else{
                       // A UNIX timestamp in seconds. Format the output as needed.
                       try{
                           int prg = client.GetPrevCursorTime(project, logstore, cp.getShard(), cp.getCheckPoint()).GetCursorTime();
                           consumerPrg = "" + prg;
                       }
                       catch(LogException e){
                           if(e.GetErrorCode() == "InvalidCursor")
                               consumerPrg = "Invalid. The consumption checkpoint is older than the data retention period.";
                           else{
                               // internal server error
                               throw e;
                           }
                       }
                   }
                   System.out.println("consumption checkpoint: " + consumerPrg);
                   String endCursor = client.GetCursor(project, logstore, cp.getShard(), CursorMode.END).GetCursor();
                   int endPrg = 0;
                   try{
                       endPrg = client.GetPrevCursorTime(project, logstore, cp.getShard(), endCursor).GetCursorTime();
                   }
                   catch(LogException e){
                       // do nothing
                   }
                   // A UNIX timestamp in seconds. Format the output as needed.
                   System.out.println("Arrival time of the last record: " + endPrg);
               }
           }
       }
   }

2. Confira abaixo um exemplo de saída:

   Name: ali-test-consumergroup2
   heartbeat timeout: 60
   ordered consumption: false
   shard: 0
   Last checkpoint update time: 0
   Consumer name: consumer_1
   consumption checkpoint: Consumption has not started
   Arrival time of the last record: 1729583617
   shard: 1
   Last checkpoint update time: 0
   Consumer name: consumer_1
   consumption checkpoint: Consumption has not started
   Arrival time of the last record: 1729583738
   Process finished with exit code 0

Console

1. Faça login no console do Simple Log Service.

2. Na seção Projects, clique em no projeto desejado.

   

3. Na aba Log Storage > Logstores, clique em no ícone à esquerda do Logstore alvo e, em seguida, clique em no ícone à esquerda de Data Consumption.

4. Na lista de consumer groups, clique em no grupo desejado.

5. Na página Consumer Group Status, visualize o checkpoint de consumo de cada shard. Esta página exibe detalhes individuais de cada shard, incluindo seu ID (shard), Last Consumed Time e o Client responsável pelo consumo. Também estão disponíveis os botões Refresh e Reset Checkpoint.