Simple Log Service:Consuma logs com grupos de consumidores e SPL

Adicione as dependências do Maven.

Requer o com.aliyun.openservices versão 0.6.51 ou posterior.

<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.51</version>
</dependency>

Crie a lógica de consumo. O exemplo a seguir mostra 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 when the checkpoint was last persisted.
    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 for data consumption. Handle all exceptions that occur during consumption. Do not throw exceptions directly.
    public String process(List<LogGroupData> logGroups, ILogHubCheckPointTracker checkPointTracker) {
        // Print the retrieved 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();
        // Write a checkpoint to the server every 30 seconds. If a worker terminates unexpectedly within 30 seconds, the new worker starts to consume data from the last checkpoint. A small amount of data may be repeatedly consumed.
        try {
            if (curTime - mLastSaveTime > 30 * 1000) {
                // The true parameter indicates that the checkpoint is immediately updated to the server. By default, the checkpoint cached in memory is automatically updated to the server every 60 seconds.
                checkPointTracker.saveCheckPoint(true);
                mLastSaveTime = curTime;
            } else {
                // The false parameter indicates that the checkpoint is cached locally. The checkpoint can be updated to the server by the automatic update mechanism.
                checkPointTracker.saveCheckPoint(false);
            }
        } catch (LogHubCheckPointException e) {
            e.printStackTrace();
        }
        return null;
    }

    // This function is called when the worker exits. You can perform cleanup tasks here.
    public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
        // Immediately save the checkpoint to the server.
        try {
            checkPointTracker.saveCheckPoint(true);
        } catch (LogHubCheckPointException e) {
            e.printStackTrace();
        }
    }
}

Crie uma fábrica de consumidores. O exemplo a seguir mostra 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 time the generatorProcessor method is called, a new SPLLogHubProcessor object must be returned.
        return new SPLLogHubProcessor();
    }
}

Crie um consumidor e inicie o consumo de dados do Logstore.

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. Replace the value with the actual endpoint.
    private static String Endpoint = "cn-hangzhou.log.aliyuncs.com";
    // The name of the Simple Log Service project. Replace the value with the actual project name. Obtain the project name from an existing project.
    private static String Project = "ali-test-project";
    // The name of the Logstore. Replace the value with the actual Logstore name. Obtain the Logstore name from an existing Logstore.
    private static String Logstore = "ali-test-logstore";
    // The name of the consumer group. Replace the value with the actual consumer group name. You do not need to create the consumer group in advance. The program automatically creates it at runtime.
    private static String ConsumerGroup = "ali-test-consumergroup2";

    // New content
    // The name of the consumer processor. Create the processor in the console or by calling the consumer processor API.
    private static String ConsumeProcessor = "test-consumer-processor";
    // End of new content
    // This example obtains 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. The names of consumers in the same consumer group must be unique. If different consumers start multiple processes on different machines to consume data from a Logstore in a balanced manner, you can use the machine IP addresses as consumer names.
        // maxFetchLogGroupSize specifies the maximum number of log groups that can be retrieved from the server at a time. You can use the default value. To change the value, use config.setMaxFetchLogGroupSize(100);. The value must be in the range of (0, 1000].
        LogHubConfig config = new LogHubConfig(ConsumerGroup, "consumer_1", Endpoint, Project, Logstore, AccessKeyId, AccessKeySecret, LogHubConfig.ConsumePosition.BEGIN_CURSOR, 1000);
        // setProcessor specifies the consumer processor for data consumption.
        config.setProcessor(ConsumeProcessor);
        ClientWorker worker = new ClientWorker(new SPLLogHubProcessorFactory(), config);
        Thread thread = new Thread(worker);
        // After the thread runs, the ClientWorker automatically runs. The ClientWorker extends the Runnable interface.
        thread.start();
        Thread.sleep(60 * 60 * 1000);
        // Call the shutdown function of the worker to exit the consumer instance. The associated thread also stops automatically.
        worker.shutdown();
        // Multiple asynchronous tasks are generated when the ClientWorker is running. After the shutdown is complete, wait for the running tasks to securely exit. We recommend that you set the sleep time to 30 seconds.
        Thread.sleep(30 * 1000);
    }
}

Execute Main.java. A saída de exemplo a seguir mostra os logs do NGINX consumidos.

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

Instale o SLS SDK para Python. Crie uma pasta chamada spl_consumer_demo e execute o comando a seguir. Install the Simple Log Service SDK for Python.

Requer o SLS SDK versão 0.9.28 ou posterior.

pip install -U aliyun-log-python-sdk

Na pasta spl_consumer_demo, crie um arquivo main.py para configurar um grupo de consumidores e consumir dados do Logstore.

import os
import time

from aliyun.log.consumer import *
from aliyun.log import *

class SPLConsumer(ConsumerProcessorBase):
    shard_id = -1
    last_check_time = 0

    def initialize(self, shard):
        self.shard_id = shard

    def process(self, log_groups, check_point_tracker):
        for log_group in log_groups.LogGroups:
            items = []
            for log in log_group.Logs:
                item = dict()
                item['time'] = log.Time
                for content in log.Contents:
                    item[content.Key] = content.Value
                items.append(item)
            log_items = dict()
            log_items['topic'] = log_group.Topic
            log_items['source'] = log_group.Source
            log_items['logs'] = items

            print(log_items)

        current_time = time.time()
        if current_time - self.last_check_time > 3:
            try:
                self.last_check_time = current_time
                check_point_tracker.save_check_point(True)
            except Exception:
                import traceback
                traceback.print_exc()
        else:
            try:
                check_point_tracker.save_check_point(False)
            except Exception:
                import traceback
                traceback.print_exc()

        # None means succesful process
        # if need to roll-back to previous checkpoint，return check_point_tracker.get_check_point()
        return None

    def shutdown(self, check_point_tracker):
        try:
            check_point_tracker.save_check_point(True)
        except Exception:
            import traceback
            traceback.print_exc()

def sleep_until(seconds, exit_condition=None, expect_error=False):
    if not exit_condition:
        time.sleep(seconds)
        return

    s = time.time()
    while time.time() - s < seconds:
        try:
            if exit_condition():
                break
        except Exception:
            if expect_error:
                continue
        time.sleep(1)

def spl_consumer_group():
    # The Simple Log Service endpoint. This example uses the endpoint of the China (Hangzhou) region. Replace the value with the actual endpoint.
    endpoint = os.environ.get('ALIYUN_LOG_SAMPLE_ENDPOINT', 'cn-hangzhou.log.aliyuncs.com')

    # This example obtains the AccessKey ID and AccessKey secret from environment variables.
    access_key_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')
    access_key = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')

    project = 'your_project'
    logstore = 'your_logstore'

    # The name of the consumer group. You do not need to create the consumer group in advance. The SDK automatically creates it.
    consumer_group = 'consumer-group'
    consumer_name = "consumer-group-name"

    # Create two consumers in the consumer group to consume data.
    option = LogHubConfig(endpoint,
                          access_key_id,
                          access_key,
                          project,
                          logstore,
                          consumer_group,
                          consumer_name,
                          processor="test-consume-processor",
                          cursor_position=CursorPosition.BEGIN_CURSOR,
                          heartbeat_interval=6,
                          data_fetch_interval=1)

    print("*** start to consume data...")
    client_worker = ConsumerWorker(SPLConsumer, consumer_option=option)
    client_worker.start()

    time.sleep(10000)

if __name__ == '__main__':
    spl_consumer_group()

Execute o arquivo main.py na pasta spl_consumer_demo e visualize o resultado.

python main.py

Instale o SDK do Go. Crie uma pasta chamada spl_demo e execute o comando a seguir. Install the Go SDK.

Requer o SLS SDK versão v0.1.107 ou posterior.

go get -u github.com/aliyun/aliyun-log-go-sdk

Na pasta spl_demo, crie um arquivo main.go para configurar um grupo de consumidores e consumir dados do Logstore.

package main

import (
  "fmt"
  "os"
  "os/signal"
  "syscall"

  sls "github.com/aliyun/aliyun-log-go-sdk"
  consumerLibrary "github.com/aliyun/aliyun-log-go-sdk/consumer"
  "github.com/go-kit/kit/log/level"
)

// README :
// This is a very simple example of pulling data from your logstore and printing it for consumption, including pre-handling for logs.

func main() {
  // The Simple Log Service endpoint. This example uses the endpoint of the China (Hangzhou) region. Replace the value with the actual endpoint.
option := consumerLibrary.LogHubConfig{
    Endpoint:          "cn-hangzhou.log.aliyuncs.com",
    AccessKeyID:       os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
    AccessKeySecret:   os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
    Project:           "your_project",
    Logstore:          "your_logstore",
    ConsumerGroupName: "test-spl-cg",
    ConsumerName:      "test-spl-consumer",
    // This options is used for initialization, will be ignored once consumer group is created and each shard has been started to be consumed.
    // Could be "begin", "end", "specific time format in time stamp", it's log receiving time.
    CursorPosition: consumerLibrary.END_CURSOR,
    // Processor is for log pre-handling before return to client, more info refer to https://www.alibabacloud.com/help/zh/sls/user-guide/rule-based-consumption
    Processor: "test-consume-processor",
  }

  consumerWorker := consumerLibrary.InitConsumerWorkerWithCheckpointTracker(option, process)
  ch := make(chan os.Signal, 1)
  signal.Notify(ch, syscall.SIGINT, syscall.SIGTERM)
  consumerWorker.Start()
  if _, ok := <-ch; ok {
    level.Info(consumerWorker.Logger).Log("msg", "get stop signal, start to stop consumer worker", "consumer worker name", option.ConsumerName)
    consumerWorker.StopAndWait()
  }
}

// Fill in your consumption logic here, and be careful not to change the parameters of the function and the return value,
// otherwise you will report errors.
func process(shardId int, logGroupList *sls.LogGroupList, checkpointTracker consumerLibrary.CheckPointTracker) (string, error) {
  fmt.Println(shardId, "loggroup", len(logGroupList.LogGroups))
  checkpointTracker.SaveCheckPoint(false)
  return "", nil
}

Execute os comandos a seguir na pasta spl_demo para instalar as dependências.

go mod tidy
go mod vendor

Execute a função main e visualize a saída.

go run main.go