Simple Log Service:Use consumer groups to consume data

Terms

Allocation rules

After a new consumer is added to a consumer group, shards that are allocated to the consumers in the consumer group are reallocated to each consumer for load balancing. The shards are reallocated based on the preceding rules.

Step 1: Consume data

You can use Simple Log Service SDK for Java, C++, Python, or Go to create consumer groups and consume data. In this example, Simple Log Service SDK for Java is used.

1. Add Maven dependencies.

   Open the pom.xml file in the root directory of the Java project 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>loghub-client-lib</artifactId>
     <version>0.6.33</version>
   </dependency>

2. Create a Main.java file. Create a consumer group and start a consumer thread to allow consumers in the consumer group to consume data in the specified Logstore. 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.

   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. Configure the parameter based on your business scenario. 
       private static String Endpoint = "cn-hangzhou.log.aliyuncs.com";
       // The name of the Simple Log Service project. Configure the parameter based on your business scenario. You must enter the name of an existing project. 
       private static String Project = "ali-cn-hangzhou-sls-admin";
       // The name of the Logstore. Configure the parameter based on your business scenario. You must enter the name of an existing Logstore. 
       private static String Logstore = "sls_operation_log";
       // The name of the consumer group. Configure the parameter based on your business scenario. 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 = "consumerGroupX";
       // Configure environment variables. 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 range: (0,1000]. 
           LogHubConfig config = new LogHubConfig(ConsumerGroup, "consumer_1", Endpoint, Project, Logstore, AccessKeyId, AccessKeySecret, LogHubConfig.ConsumePosition.BEGIN_CURSOR,1000);
           ClientWorker worker = new ClientWorker(new SampleLogHubProcessorFactory(), config);
           Thread thread = new Thread(worker);
           // After you execute the thread, 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 consumption instance. The associated thread 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);
       }
   }

3. Create a SampleLogHubProcessor.java file.

   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 SampleLogHubProcessor implements ILogHubProcessor {
       private int shardId;
       // The time when the last 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 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 checkpoint. A small amount of data may be repeatedly consumed. 
           try {
               if (curTime - mLastSaveTime > 30 * 1000) {
                   // The value true indicates that the checkpoints are immediately updated to Simple Log Service. By default, the 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 checkpoints are locally cached and can be updated to Simple Log Service by using the automatic 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 checkpoints. 
       public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
           // Immediately store checkpoints to Simple Log Service. 
           try {
               checkPointTracker.saveCheckPoint(true);
           } catch (LogHubCheckPointException e) {
               e.printStackTrace();
           }
       }
   }

   For more information, see Java, C++, Python, and Go.

4. Create a SampleLogHubProcessorFactory.java file.

   class SampleLogHubProcessorFactory implements ILogHubProcessorFactory {
       public ILogHubProcessor generatorProcessor() {
           // Generate a consumer. Each time the generatorProcessor method is called, it is expected to return a new SampleLogHubProcessor object. 
           return new SampleLogHubProcessor();
       }
   }

5. Run Main.java.

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

Step 2: View the status of a consumer group

Authorize a RAM user to perform operations on consumer groups

Before you can use a Resource Access Management (RAM) user to manage consumer groups, you must grant the required permissions to the RAM user. For more information, see Step 2: Grant permissions to the RAM user.

The following list provides resource information about a consumer group. To allow a RAM user to perform operations on the consumer group, you can refer to the following code to grant the required permissions to the RAM user:

• ID of the Alibaba Cloud account to which the project belongs: 174649****602745

• ID of the region where the project resides: cn-hangzhou

• Name of the project: project-test

• Name of the Logstore: logstore-test

• Name of the consumer group: consumergroup-test

{
  "Version": "1",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "log:GetCursorOrData"
      ],
      "Resource": "acs:log:cn-hangzhou:174649****602745:project/project-test/logstore/logstore-test"
    },
    {
      "Effect": "Allow",
      "Action": [
        "log:CreateConsumerGroup",
        "log:ListConsumerGroup"
      ],
      "Resource": "acs:log:cn-hangzhou:174649****602745:project/project-test/logstore/logstore-test/consumergroup/*"
    },
    {
      "Effect": "Allow",
      "Action": [
        "log:ConsumerGroupUpdateCheckPoint",
        "log:ConsumerGroupHeartBeat",
        "log:UpdateConsumerGroup",
        "log:GetConsumerGroupCheckPoint"
      ],
      "Resource": "acs:log:cn-hangzhou:174649****602745:project/project-test/logstore/logstore-test/consumergroup/consumergroup-test"
    }
  ]
}

What to do next

• Configure Log4j for troubleshooting.

  We recommend that you configure Log4j for your consumer program to display error messages when exceptions occur in consumer groups. This helps you troubleshoot the errors. The following code shows a log4j.properties configuration file that is commonly used:

  log4j.rootLogger = info,stdout
  log4j.appender.stdout = org.apache.log4j.ConsoleAppender
  log4j.appender.stdout.Target = System.out
  log4j.appender.stdout.layout = org.apache.log4j.PatternLayout
  log4j.appender.stdout.layout.ConversionPattern = [%-5p] %d{yyyy-MM-dd HH:mm:ss,SSS} method:%l%n%m%n

  After you configure Log4j, you can receive error messages when you run the consumer program. The following example shows an error message:

  [WARN ] 2018-03-14 12:01:52,747 method:com.aliyun.openservices.loghub.client.LogHubConsumer.sampleLogError(LogHubConsumer.java:159)
  com.aliyun.openservices.log.exception.LogException: Invalid loggroup count, (0,1000]

• Use a consumer group to consume data that is generated after a point in time

  // consumerStartTimeInSeconds specifies a point in time. The data that is generated after the point in time is consumed. 
  public LogHubConfig(String consumerGroupName, 
                        String consumerName, 
                        String loghubEndPoint,
                        String project, String logStore,
                        String accessId, String accessKey,
                        int consumerStartTimeInSeconds);
  
  // position is an enumeration variable. LogHubConfig.ConsumePosition.BEGIN_CURSOR specifies that the consumption starts from the earliest data. LogHubConfig.ConsumePosition.END_CURSOR specifies that the consumption starts from the latest data. 
  public LogHubConfig(String consumerGroupName, 
                        String consumerName, 
                        String loghubEndPoint,
                        String project, String logStore,
                        String accessId, String accessKey,
                        ConsumePosition position);

  Note

  • You can use different constructors based on your business requirements.

  • If a checkpoint is stored on Simple Log Service, data consumption starts from the checkpoint.

  • When Simple Log Service consumes data, a checkpoint is preferentially used to start data consumption. If you want to specify a point in time for Simple Log Service to start data consumption, make sure that the value of consumerStartTimeInSeconds falls within the time-to-live (TTL) period. Otherwise, Simple Log Service cannot consume data based on your configurations.

• Reset a checkpoint.

  public static void updateCheckpoint() throws Exception {
          Client client = new Client(host, accessId, accessKey);
          // Specify a UNIX timestamp that is accurate to seconds. If you specify a timestamp in milliseconds, divide the timestamp by 1000. Example:
          long timestamp = Timestamp.valueOf("2017-11-15 00:00:00").getTime() / 1000;
          ListShardResponse response = client.ListShard(new ListShardRequest(project, logStore));
          for (Shard shard : response.GetShards()) {
              int shardId = shard.GetShardId();
              String cursor = client.GetCursor(project, logStore, shardId, timestamp).GetCursor();
              client.UpdateCheckPoint(project, logStore, consumerGroup, shardId, cursor);
          }
      }

References

• API

  Operation	API operation
  Create a consumer group	CreateConsumerGroup
  Query a consumer group	ListConsumerGroup
  Delete a consumer group	DeleteConsumerGroup
  Update a consumer group	UpdateConsumerGroup
  Send a heartbeat message for a consumer	ConsumerGroupHeartBeat
  Query the checkpoints of a consumer group	GetCheckPoint
  Update the checkpoints of a consumer group	ConsumerGroupUpdateCheckPoint

• SDK

  Programming language	References
  Java	Use Simple Log Service SDK for Java to manage consumer groups Use consumer groups to consume data
  Python	Use Simple Log Service SDK for Python to manage consumer groups Use consumer groups to consume data

• CLI

  Operation	CLI command
  Create a consumer group	create_consumer_group
  Query a consumer group	list_consumer_group
  Update a consumer group	update_consumer_group
  Delete a consumer group	delete_consumer_group
  Query the checkpoints of a consumer group	get_check_point
  Update the checkpoints of a consumer group	update_check_point