gunakan sdk untuk mengonsumsi log berdasarkan prosesor konsumsi (spl) - Simple Log Service

Anda dapat menggunakan SDK untuk mengonsumsi log berdasarkan Consume Processor (SPL). Contoh kode disediakan dalam Java, Python, dan Go.

Prasyarat

Pengguna Resource Access Management (RAM) telah dibuat dan diberikan izin yang diperlukan. Untuk informasi selengkapnya, lihat Buat pengguna RAM dan berikan izin.
Variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET telah dikonfigurasi. Untuk informasi selengkapnya, lihat Konfigurasikan variabel lingkungan di Linux, macOS, dan Windows.
Penting
- Pasangan AccessKey akun Alibaba Cloud memiliki izin atas semua operasi API. Kami menyarankan Anda menggunakan pasangan AccessKey pengguna RAM untuk memanggil operasi API atau melakukan O&M rutin.
- Jangan sertakan ID AccessKey atau rahasia AccessKey Anda dalam kode proyek. Jika salah satunya bocor, keamanan semua resource dalam akun Anda mungkin akan terganggu.
Membuat Consumer Processor

Contoh kode

Java

Instal SDK Simple Log Service. Di direktori root proyek Java Anda, buka file pom.xml dan tambahkan dependensi Maven berikut. Untuk informasi selengkapnya, lihat Instal SDK Java.

SDK Simple Log Service untuk Java harus versi 0.6.126 atau lebih baru.

<dependency>
  <groupId>com.google.protobuf</groupId>
  <artifactId>protobuf-java</artifactId>
  <version>2.5.0</version>
</dependency>
<!-- Impor SDK Simple Log Service -->
<dependency>
  <groupId>com.aliyun.openservices</groupId>
  <artifactId>aliyun-log</artifactId>
  <version>0.6.126</version>
</dependency>

Buat file PullLogsWithSPLDemo.java. Contoh ini memanggil operasi PullLog untuk mengonsumsi data log berdasarkan SPL.

import com.aliyun.openservices.log.Client;
import com.aliyun.openservices.log.common.*;
import com.aliyun.openservices.log.request.PullLogsRequest;
import com.aliyun.openservices.log.response.ListShardResponse;
import com.aliyun.openservices.log.response.PullLogsResponse;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class PullLogsWithSPLDemo {
    // Titik akhir Simple Log Service. Contoh ini menggunakan wilayah Tiongkok (Hangzhou). Ganti dengan titik akhir wilayah Anda.
    private static final String endpoint = "cn-hangzhou.log.aliyuncs.com";
    // Contoh ini mengambil ID AccessKey dan rahasia AccessKey dari variabel lingkungan.
    private static final String accessKeyId = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID");
    private static final String accessKeySecret = System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET");
    // Nama proyek. Ganti dengan nama proyek aktual Anda.
    private static final String project = "ali-project-test";
    // Nama penyimpanan log. Ganti dengan nama penyimpanan log aktual Anda.
    private static final String logStore = "test-logstore";

    public static void main(String[] args) throws Exception {
        // ID Consume Processor.
        String processorName = "processor-test";
        // Buat klien Simple Log Service.
        Client client = new Client(endpoint, accessKeyId, accessKeySecret);
        // Kueri shard dari penyimpanan log.
        ListShardResponse resp = client.ListShard(project, logStore);        System.out.printf("%s memiliki %d shard\n", logStore, resp.GetShards().size());
        Map<Integer, String> cursorMap = new HashMap<>();
        for (Shard shard : resp.GetShards()) {
            int shardId = shard.getShardId();
            // Mulai konsumsi dari awal dan dapatkan kursor. Untuk memulai dari akhir, gunakan Consts.CursorMode.END.
            cursorMap.put(shardId, client.GetCursor(project, logStore, shardId, Consts.CursorMode.BEGIN).GetCursor());
        }
        try {
            while (true) {
                // Dapatkan log dari setiap shard.
                for (Shard shard : resp.GetShards()) {
                    int shardId = shard.getShardId();
                    PullLogsRequest request = new PullLogsRequest(project, logStore, shardId, 1000, cursorMap.get(shardId));
                    // ID Consume Processor.
                    request.setProcessor(processorName);
                    PullLogsResponse response = client.pullLogs(request);
                    // Log berada dalam kelompok log. Pisahkan sesuai kebutuhan.
                    List<LogGroupData> logGroups = response.getLogGroups();
                    System.out.printf("Dapatkan %d kelompok log dari penyimpanan log:%s:\tShard:%d\n", logGroups.size(), logStore, shardId);

                    // Pindahkan kursor setelah Anda memproses log yang ditarik.
                    cursorMap.put(shardId, response.getNextCursor());
                }
            }
        } catch (LogException e) {
            System.out.println("kode kesalahan :" + e.GetErrorCode());
            System.out.println("pesan kesalahan :" + e.GetErrorMessage());
            throw e;
        }
    }
}

Jalankan fungsi utama dan lihat output.

Dapatkan 41 kelompok log dari penyimpanan log:test-logstore:	Shard:0
Dapatkan 49 kelompok log dari penyimpanan log:test-logstore:	Shard:1
Dapatkan 43 kelompok log dari penyimpanan log:test-logstore:	Shard:0
Dapatkan 39 kelompok log dari penyimpanan log:test-logstore:	Shard:1
... ...

Python

Instal SDK Simple Log Service. Buat folder proyek bernama spl_demo dan jalankan perintah berikut di dalam folder tersebut. Untuk informasi selengkapnya, lihat Instal SDK Simple Log Service untuk Python.

SDK Alibaba Cloud Simple Log Service harus versi 0.9.28 atau lebih baru.
```
pip install -U aliyun-log-python-sdk
```

Di folder spl_demo, buat file main.py. Kode ini membuat kelompok konsumen dan memulai thread konsumen untuk mengonsumsi data dari penyimpanan log yang ditentukan.

# encoding: utf-8

import time
import os
from aliyun.log import *

def main():
    # Titik akhir Simple Log Service. Contoh ini menggunakan wilayah Tiongkok (Hangzhou). Ganti dengan titik akhir wilayah Anda.
    endpoint = 'cn-hangzhou.log.aliyuncs.com'
    # Contoh ini mengambil ID AccessKey dan rahasia AccessKey dari variabel lingkungan.
    access_key_id = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')
    access_key = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')
    # Nama proyek. Ganti dengan nama proyek aktual Anda.
    project_name = 'ali-project-test'
    # Nama penyimpanan log. Ganti dengan nama penyimpanan log aktual Anda.
    logstore_name = 'test-logstore'
    # ID Consume Processor.
    processor = "processor-test"
    init_cursor = 'end'
    log_group_count = 10

    # Buat klien Simple Log Service.
    client = LogClient(endpoint, access_key_id, access_key)

    cursor_map = {}
    # Daftar shard dari penyimpanan log.
    res = client.list_shards(project_name, logstore_name)
    res.log_print()
    shards = res.shards

    # Dapatkan kursor awal.
    for shard in shards:
        shard_id = shard.get('shardID')
        res = client.get_cursor(project_name, logstore_name, shard_id, init_cursor)
        cursor_map[shard_id] = res.get_cursor()

    # Baca data dari setiap shard dalam loop.
    while True:
        for shard in shards:
            shard_id = shard.get('shardID')
            res = client.pull_logs(project_name, logstore_name, shard_id, cursor_map.get(shard_id), log_group_count,
                                   processor=processor)
            res.log_print()
            if cursor_map[shard_id] == res.next_cursor:
                # hanya untuk debug
                time.sleep(3)
            else:
                cursor_map[shard_id] = res.next_cursor


if __name__ == '__main__':
    main()

Jalankan fungsi utama dan lihat output.

ListShardResponse:
headers: {'Server': 'AliyunSLS', 'Content-Type': 'application/json', 'Content-Length': '335', 'Connection': 'keep-alive', 'Access-Control-Allow-Origin': '*', 'Date': 'Wed, 26 Feb 2025 09:46:17 GMT', 'x-log-time': '1740563177', 'x-log-requestid': '67BEE2E9132069E22A1F967D'}
res: [{'shardID': 0, 'status': 'readwrite', 'inclusiveBeginKey': '00000000000000000000000000000000', 'exclusiveEndKey': '80000000000000000000000000000000', 'createTime': 1737010019}, {'shardID': 1, 'status': 'readwrite', 'inclusiveBeginKey': '80000000000000000000000000000000', 'exclusiveEndKey': 'ffffffffffffffffffffffffffffffff', 'createTime': 1737010019}]
PullLogResponse
next_cursor MTczNz********c3ODgyMjQ0MQ==
log_count 0
headers: {'Server': 'AliyunSLS', 'Content-Type': 'application/x-protobuf', 'Content-Length': '1', 'Connection': 'keep-alive', 'Access-Control-Allow-Origin': '*', 'Date': 'Wed, 26 Feb 2025 09:46:17 GMT', 'x-log-cursor-time': '0', 'x-log-end-of-cursor': '1', 'x-log-failedlines': '0', 'x-log-rawdatacount': '0', 'x-log-rawdatalines': '0', 'x-log-rawdatasize': '0', 'x-log-read-last-cursor': '0', 'x-log-resultlines': '0', 'x-log-time': '1740563177', 'x-log-bodyrawsize': '0', 'x-log-compresstype': 'gzip', 'x-log-count': '0', 'x-log-cursor': 'MTczNzAx********ODgyMjQ0MQ==', 'x-log-requestid': '67BEE2E974CA9ABCE7DDC7D6'}
detail: []
PullLogResponse
next_cursor MTczNz********c3OTg5NzE3NA==
log_count 0
headers: {'Server': 'AliyunSLS', 'Content-Type': 'application/x-protobuf', 'Content-Length': '1', 'Connection': 'keep-alive', 'Access-Control-Allow-Origin': '*', 'Date': 'Wed, 26 Feb 2025 09:46:21 GMT', 'x-log-cursor-time': '0', 'x-log-end-of-cursor': '1', 'x-log-failedlines': '0', 'x-log-rawdatacount': '0', 'x-log-rawdatalines': '0', 'x-log-rawdatasize': '0', 'x-log-read-last-cursor': '0', 'x-log-resultlines': '0', 'x-log-time': '1740563181', 'x-log-bodyrawsize': '0', 'x-log-compresstype': 'gzip', 'x-log-count': '0', 'x-log-cursor': 'MTczNzAx********OTg5NzE3NA==', 'x-log-requestid': '67BEE2EDF2B58CF1756526EF'}
detail: []
PullLogResponse
... ...

Go

Instal SDK Simple Log Service. Buat folder proyek bernama spl_demo dan jalankan perintah berikut di dalam folder tersebut. Untuk informasi selengkapnya, lihat Instal SDK Go.

SDK Alibaba Cloud Simple Log Service harus versi 0.1.107 atau lebih baru.
```
go get -u github.com/aliyun/aliyun-log-go-sdk
```

Di folder spl_demo, buat file main.go. Kode ini membuat kelompok konsumen dan memulai thread konsumen untuk mengonsumsi data dari penyimpanan log yang ditentukan.

package main

import (
	"fmt"
	"os"
	"time"

	sls "github.com/aliyun/aliyun-log-go-sdk"
)

func main() {
	client := &sls.Client{
		AccessKeyID:     os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
		AccessKeySecret: os.Getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
		Endpoint:        "cn-chengdu.log.aliyuncs.com",
	}

	project := "ali-project-test"
	logstore := "test-logstore"
	initCursor := "end"
    // ID Consume Processor.
	consumeProcessor := "ali-test-consume-processor"
	shards, err := client.ListShards(project, logstore)
	if err != nil {
		fmt.Println("kesalahan ListShards", err)
		return
	}

	shardCursorMap := map[int]string{}
	for _, shard := range shards {
		cursor, err := client.GetCursor(project, logstore, shard.ShardID, initCursor)
		if err != nil {
			fmt.Println("kesalahan GetCursor", shard.ShardID, err)
			return
		}
		shardCursorMap[shard.ShardID] = cursor
	}

	for {
		for _, shard := range shards {
			pullLogRequest := &sls.PullLogRequest{
				Project:          project,
				Logstore:         logstore,
				ShardID:          shard.ShardID,
				LogGroupMaxCount: 10,
				Processor:        consumeProcessor,
				Cursor:           shardCursorMap[shard.ShardID],
			  }
			lg, nextCursor, err := client.PullLogsV2(pullLogRequest)
			fmt.Println("shard: ", shard.ShardID, "kelompok log: ", len(lg.LogGroups), "nextCursor: ", nextCursor)
			if err != nil {
				fmt.Println("kesalahan PullLogsV2", shard.ShardID, err)
				return
			}
			shardCursorMap[shard.ShardID] = nextCursor
			if len(lg.LogGroups) == 0 {
				// hanya untuk debug
				time.Sleep(time.Duration(3) * time.Second)
			}
		}
	}
}

Jalankan fungsi utama dan lihat output.

shard:  0 kelompok log:  41 nextCursor:  MTY5Mz*******TIxNjcxMDcwMQ==
shard:  1 kelompok log:  49 nextCursor:  MTY5Mz*******DYwNDIyNDQ2Mw==
shard:  0 kelompok log:  43 nextCursor:  MTY5Mz*******TIxNjcxMDcwMQ==
shard:  1 kelompok log:  39 nextCursor:  MTY5Mz*******DYwNDIyNDQ2Mw==
... ...

Prasyarat​

Contoh kode

Java

Python

Go

Prasyarat