Mengonsumsi log berdasarkan prosesor konsumen (SPL) menggunakan SDK - Simple Log Service

Topik ini memberikan contoh cara mengonsumsi log menggunakan prosesor konsumen dengan Search Processing Language (SPL) dan software development kit (SDK).

Prasyarat

Pengguna Resource Access Management (RAM) telah dibuat dan diberi izin yang diperlukan. Untuk informasi lebih lanjut, lihat Buat pengguna RAM dan berikan izin.
Variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET telah dikonfigurasi. Untuk informasi lebih lanjut, lihat Konfigurasikan variabel lingkungan di Linux, macOS, dan Windows.
Penting
- Pasangan Kunci Akses dari akun Alibaba Cloud memiliki izin untuk semua Operasi API. Kami menyarankan Anda menggunakan Pasangan Kunci Akses dari Pengguna RAM untuk memanggil Operasi API atau melakukan pemeliharaan rutin O&M.
- Jangan sertakan ID AccessKey atau Rahasia AccessKey Anda dalam kode proyek Anda. Jika salah satunya bocor, keamanan semua sumber daya di akun Anda mungkin terganggu.
Buat prosesor konsumen

Contoh kode

Java

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

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

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

Buat file bernama PullLogsWithSPLDemo.java. Contoh ini memanggil operasi PullLog untuk membaca data log. Ini menunjukkan cara menggunakan Java SDK untuk mengonsumsi data log dengan 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 layanan Simple Log Service. Contoh ini menggunakan titik akhir wilayah China (Hangzhou). Ganti titik akhir dengan yang sesuai untuk wilayah Anda.
    private static final String endpoint = "cn-hangzhou.log.aliyuncs.com";
    // Contoh ini mendapatkan 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 nilainya dengan nama proyek sebenarnya Anda.
    private static final String project = "ali-project-test";
    // Nama penyimpanan log. Ganti nilainya dengan nama penyimpanan log sebenarnya Anda.
    private static final String logStore = "test-logstore";

    public static void main(String[] args) throws Exception {
        // Parameter processorName menentukan identitas prosesor konsumen.
        String processorName = "processor-test";
        // Buat klien Simple Log Service.
        Client client = new Client(endpoint, accessKeyId, accessKeySecret);
        // Query 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 mulai 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));
                    // Parameter processorName menentukan identitas prosesor konsumen.
                    request.setProcessor(processorName);
                    PullLogsResponse response = client.pullLogs(request);
                    // Log disimpan dalam kelompok log. Anda dapat memisahkan log berdasarkan logika bisnis Anda.
                    List<LogGroupData> logGroups = response.getLogGroups();
                    System.out.printf("Dapatkan %d logGroup dari logstore:%s:\tShard:%d\n", logGroups.size(), logStore, shardId);

                    // Setelah Anda memproses log yang ditarik, pindahkan kursor.
                    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 hasilnya.

Dapatkan 41 logGroup dari logstore:test-logstore:	Shard:0
Dapatkan 49 logGroup dari logstore:test-logstore:	Shard:1
Dapatkan 43 logGroup dari logstore:test-logstore:	Shard:0
Dapatkan 39 logGroup dari logstore:test-logstore:	Shard:1
... ...

Python

Instal Simple Log Service SDK. Buat folder proyek bernama spl_demo dan jalankan perintah berikut di folder tersebut. Untuk informasi lebih lanjut, lihat Instal Simple Log Service SDK untuk Python.
Versi Simple Log Service SDK untuk Python harus 0.9.28 atau lebih baru.
```
pip install -U aliyun-log-python-sdk
```

Di folder spl_demo, buat file bernama main.py. File 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 layanan Simple Log Service. Contoh ini menggunakan titik akhir wilayah China (Hangzhou). Ganti titik akhir dengan yang sesuai untuk wilayah Anda.
    endpoint = 'cn-hangzhou.log.aliyuncs.com'
    # Contoh ini mendapatkan 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 nilainya dengan nama proyek sebenarnya Anda.
    project_name = 'ali-project-test'
    # Nama penyimpanan log. Ganti nilainya dengan nama penyimpanan log sebenarnya Anda.
    logstore_name = 'test-logstore'
    # Parameter processor menentukan identitas prosesor konsumen.
    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 hasilnya.

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 Simple Log Service SDK. Buat folder proyek bernama spl_demo dan jalankan perintah berikut di folder tersebut. Untuk informasi lebih lanjut, lihat Instal Go SDK.
Versi Simple Log Service SDK untuk Go harus v0.1.107 atau lebih baru.
```
go get -u github.com/aliyun/aliyun-log-go-sdk
```

Di folder spl_demo, buat file bernama main.go. File 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"
    // Parameter consumeProcessor menentukan identitas prosesor konsumen.
	consumeProcessor := "ali-test-consume-processor"
	shards, err := client.ListShards(project, logstore)
	if err != nil {
		fmt.Println("ListShards error", err)
		return
	}

	shardCursorMap := map[int]string{}
	for _, shard := range shards {
		cursor, err := client.GetCursor(project, logstore, shard.ShardID, initCursor)
		if err != nil {
			fmt.Println("GetCursor error", 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, "loggroups: ", len(lg.LogGroups), "nextCursor: ", nextCursor)
			if err != nil {
				fmt.Println("PullLogsV2 error", 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 hasilnya.

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

Prasyarat​

Contoh kode

Java

Python

Go

Prasyarat