Simple Log Service:Use consumer groups to consume logs

Use Simple Log Service SDK

Sample code:

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 {
         // In this example, the AccessKey ID and AccessKey secret are obtained from environment variables. 
        String accessId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
        String accessKey = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
        // The name of the project. 
        String projectName = "ali-test-project";
        // The name of the Logstore. 
        String logstoreName = "ali-test-logstore";
        // The Simple Log Service endpoint. In this example, the Simple Log Service endpoint for the China (Hangzhou) region is used. Replace the parameter value 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 {
            // The name of the consumer group. 
            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;
        }
    }
}

For more information about the sample code that is used to manage consumer groups, see Use Simple Log Service SDK for Java to manage consumer groups and Use Simple Log Service SDK for Python to manage consumer groups.

Use Simple Log Service API

For more information about how to use Simple Log Service API to create a consumer group, see CreateConsumerGroup.

For more information about how to check whether a consumer group is created, see ListConsumerGroup.

Use Simple Log Service CLI

For more information about how to use Simple Log Service CLI to create a consumer group, see create_consumer_group.

For more information about how to check whether a consumer group is created, see list_consumer_group.

Example 1: Use the SDK

Example 2: Use the SDK and SPL

1. Add Maven dependencies.

   Open the pom.xml file and add the following code:

       <dependency>
         <groupId>com.google.protobuf</groupId>
         <artifactId>protobuf-java</artifactId>
         <version>2.5.0</version>
       </dependency>
       <dependency>
         <groupId>com.aliyun.openservices</groupId>
         <artifactId>aliyun-log</artifactId>
         <version>0.6.114</version>
       </dependency>

2. Write the implementation logic of data consumption. Sample code:

   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;
       // The time at which the last consumption checkpoint was stored. 
       private long mLastSaveTime = 0;
   
       // The initialize method is called once when a processor object is initialized.
       public void initialize(int shardId) {
           this.shardId = shardId;
       }
   
       // The main logic of data consumption. You must include the code to handle all exceptions that may occur during data consumption. 
       public String process(List<LogGroupData> logGroups, ILogHubCheckPointTracker checkPointTracker) {
           // Display the obtained 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();
           // A consumption checkpoint is written to Simple Log Service at an interval of 30 seconds. If the ClientWorker instance unexpectedly stops within 30 seconds, a newly started ClientWorker instance consumes data from the last consumption checkpoint. A small amount of data may be repeatedly consumed. 
           try {
               if (curTime - mLastSaveTime > 30 * 1000) {
                   // The value true indicates that the consumption checkpoints are immediately updated to Simple Log Service. By default, the consumption checkpoints that are cached in memory are automatically updated to Simple Log Service at an interval of 60 seconds. 
                   checkPointTracker.saveCheckPoint(true);
                   mLastSaveTime = curTime;
               } else {
                   // The value false indicates that the consumption checkpoints are locally cached and can be updated to Simple Log Service by using the automatic consumption checkpoint update mechanism. 
                   checkPointTracker.saveCheckPoint(false);
               }
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
           return null;
       }
   
       // The shutdown function of the ClientWorker instance is called. You can manage the consumption checkpoints. 
       public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
           // Immediately store consumption checkpoints to Simple Log Service. 
           try {
               checkPointTracker.saveCheckPoint(true);
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
       }
   }

3. Define a consumer entity. Sample code:

   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. Each time the generatorProcessor method is called, a new SPLLogHubProcessor object is returned as expected. 
           return new SPLLogHubProcessor();
       }
   }

4. Create a consumer group and start a consumer thread to allow consumers in the consumer group to consume data in the specified Logstore. Sample code:

   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 Simple Log Service endpoint. Enter an endpoint based on your business requirements. 
       private static String Endpoint = "cn-hangzhou.log.aliyuncs.com";
       // The name of the Simple Log Service project. Enter a name based on your business requirements. You must enter the name of an existing project. 
       private static String Project = "ali-test-project";
       // The name of the Logstore. Enter a name based on your business requirements. You must enter the name of an existing Logstore. 
       private static String Logstore = "ali-test-logstore";
       // The name of the consumer group. Enter a name based on your business requirements. You do not need to create a consumer group in advance. A consumer group is automatically created when a program runs. 
       private static String ConsumerGroup = "ali-test-consumergroup2";
       // In this example, the AccessKey ID and AccessKey secret are obtained 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 specifies the name of a consumer. The name of each consumer in a consumer group must be unique. If different consumers start processes on different machines to consume data in a Logstore, you can use the machine IP addresses to identify each consumer. 
           // maxFetchLogGroupSize specifies the maximum number of log groups that can be obtained from Simple Log Service at a time. Retain the default value. You can use config.setMaxFetchLogGroupSize(100); to change the maximum number. Valid values: (0,1000]. 
           LogHubConfig config = new LogHubConfig(ConsumerGroup, "consumer_1", Endpoint, Project, Logstore, AccessKeyId, AccessKeySecret, LogHubConfig.ConsumePosition.BEGIN_CURSOR, 1000);
           // You can use setQuery to specify a Simple Log Service Processing Language (SPL) statement for data 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 instance runs, the ClientWorker instance automatically runs and extends the Runnable interface. 
           thread.start();
           Thread.sleep(60 * 60 * 1000);
           // The shutdown function of the ClientWorker instance is called to exit the consumer. The Thread instance is automatically stopped. 
           worker.shutdown();
           // Multiple asynchronous tasks are generated when the ClientWorker instance is running. To ensure that all running tasks securely stop after the shutdown, we recommend that you set Thread.sleep to 30 seconds. 
           Thread.sleep(30 * 1000);
       }
   }

5. Run the Main.java file.

   In this example, NGINX logs are consumed and the consumption result is displayed.

   :    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
   --------
   ......