Konsumsi log dengan kelompok konsumen dan SPL - SDK Java Go Python - Simple Log Service

Kelompok konsumen secara otomatis menangani load balancing dan failover, sehingga Anda dapat fokus pada logika bisnis. Gunakan SDK Java, Go, atau Python untuk mengonsumsi data Logstore dengan pernyataan SPL.

Prasyarat

RAM user telah dibuat dan diberikan izin yang diperlukan. Buat RAM user dan berikan izin.
Variabel lingkungan ALIBABA_CLOUD_ACCESS_KEY_ID dan ALIBABA_CLOUD_ACCESS_KEY_SECRET telah dikonfigurasi. Untuk informasi selengkapnya, lihat Konfigurasi variabel lingkungan di Linux, macOS, dan Windows.
Penting
- Pasangan AccessKey dari Akun Alibaba Cloud memiliki izin atas semua operasi API. Kami menyarankan Anda menggunakan pasangan AccessKey dari RAM user untuk memanggil operasi API atau melakukan tugas O&M rutin.
- Jangan sertakan ID AccessKey atau Rahasia AccessKey Anda dalam kode proyek. Jika salah satunya bocor, keamanan semua sumber daya dalam akun Anda mungkin terganggu.
Buat prosesor konsumen

Contoh kode

Java

Tambahkan dependensi Maven.

Memerlukan com.aliyun.openservices versi 0.6.51 atau lebih baru.

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

Buat logika konsumsi. Contoh berikut menunjukkan 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;
    // Waktu saat checkpoint terakhir dipertahankan.
    private long mLastSaveTime = 0;

    // Metode initialize dipanggil sekali saat objek prosesor diinisialisasi.
    public void initialize(int shardId) {
        this.shardId = shardId;
    }

    // Logika utama untuk konsumsi data. Tangani semua exception yang terjadi selama konsumsi. Jangan lempar exception secara langsung.
    public String process(List<LogGroupData> logGroups, ILogHubCheckPointTracker checkPointTracker) {
        // Cetak data yang diambil.
        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();
        // Tulis checkpoint ke server setiap 30 detik. Jika worker berhenti secara tak terduga dalam rentang 30 detik, worker baru akan mulai mengonsumsi data dari checkpoint terakhir. Sejumlah kecil data mungkin dikonsumsi berulang.
        try {
            if (curTime - mLastSaveTime > 30 * 1000) {
                // Parameter true menunjukkan bahwa checkpoint segera diperbarui ke server. Secara default, checkpoint yang di-cache di memori diperbarui ke server setiap 60 detik.
                checkPointTracker.saveCheckPoint(true);
                mLastSaveTime = curTime;
            } else {
                // Parameter false menunjukkan bahwa checkpoint di-cache secara lokal. Checkpoint dapat diperbarui ke server melalui mekanisme pembaruan otomatis.
                checkPointTracker.saveCheckPoint(false);
            }
        } catch (LogHubCheckPointException e) {
            e.printStackTrace();
        }
        return null;
    }

    // Fungsi ini dipanggil saat worker keluar. Anda dapat melakukan tugas pembersihan di sini.
    public void shutdown(ILogHubCheckPointTracker checkPointTracker) {
        // Segera simpan checkpoint ke server.
        try {
            checkPointTracker.saveCheckPoint(true);
        } catch (LogHubCheckPointException e) {
            e.printStackTrace();
        }
    }
}

Buat factory konsumen. Contoh berikut menunjukkan 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() {
        // Hasilkan instans konsumen. Catatan: Setiap kali metode generatorProcessor dipanggil, objek SPLLogHubProcessor baru harus dikembalikan.
        return new SPLLogHubProcessor();
    }
}

Buat konsumen dan mulai mengonsumsi data dari 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 {
    // Titik akhir Simple Log Service. Ganti nilai ini dengan titik akhir yang sebenarnya.
    private static String Endpoint = "cn-hangzhou.log.aliyuncs.com";
    // Nama proyek Simple Log Service. Ganti nilai ini dengan nama proyek yang sebenarnya. Dapatkan nama proyek dari proyek yang sudah ada.
    private static String Project = "ali-test-project";
    // Nama Logstore. Ganti nilai ini dengan nama Logstore yang sebenarnya. Dapatkan nama Logstore dari Logstore yang sudah ada.
    private static String Logstore = "ali-test-logstore";
    // Nama kelompok konsumen. Ganti nilai ini dengan nama kelompok konsumen yang sebenarnya. Anda tidak perlu membuat kelompok konsumen terlebih dahulu. Program akan membuatnya secara otomatis saat runtime.
    private static String ConsumerGroup = "ali-test-consumergroup2";

    // Konten baru
    // Nama prosesor konsumen. Buat prosesor di Konsol atau dengan memanggil API prosesor konsumen.
    private static String ConsumeProcessor = "test-consumer-processor";
    // Akhir konten baru
    // Contoh ini mengambil ID AccessKey dan Rahasia AccessKey dari variabel lingkungan.
    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 adalah nama konsumen. Nama konsumen dalam kelompok konsumen yang sama harus unik. Jika konsumen berbeda menjalankan beberapa proses di mesin berbeda untuk mengonsumsi data dari Logstore secara seimbang, Anda dapat menggunakan alamat IP mesin sebagai nama konsumen.
        // maxFetchLogGroupSize menentukan jumlah maksimum kelompok log yang dapat diambil dari server sekaligus. Anda dapat menggunakan nilai default. Untuk mengubah nilai, gunakan config.setMaxFetchLogGroupSize(100);. Nilainya harus berada dalam rentang (0, 1000].
        LogHubConfig config = new LogHubConfig(ConsumerGroup, "consumer_1", Endpoint, Project, Logstore, AccessKeyId, AccessKeySecret, LogHubConfig.ConsumePosition.BEGIN_CURSOR, 1000);
        // setProcessor menentukan prosesor konsumen untuk konsumsi data.
        config.setProcessor(ConsumeProcessor);
        ClientWorker worker = new ClientWorker(new SPLLogHubProcessorFactory(), config);
        Thread thread = new Thread(worker);
        // Setelah thread berjalan, ClientWorker akan berjalan secara otomatis. ClientWorker mengimplementasikan antarmuka Runnable.
        thread.start();
        Thread.sleep(60 * 60 * 1000);
        // Panggil fungsi shutdown worker untuk keluar dari instans konsumen. Thread terkait juga akan berhenti secara otomatis.
        worker.shutdown();
        // Beberapa tugas asinkron dihasilkan saat ClientWorker berjalan. Setelah shutdown selesai, tunggu tugas yang sedang berjalan untuk keluar dengan aman. Kami menyarankan Anda mengatur waktu sleep selama 30 detik.
        Thread.sleep(30 * 1000);
    }
}

Jalankan Main.java. Output contoh berikut menunjukkan log NGINX yang telah dikonsumsi.

:    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

Python

Instal SDK SLS untuk Python. Buat folder bernama spl_consumer_demo dan jalankan perintah berikut. Instal SDK Simple Log Service untuk Python.

Memerlukan SDK SLS versi 0.9.28 atau lebih baru.
```
pip install -U aliyun-log-python-sdk
```

Di folder spl_consumer_demo, buat file main.py untuk menyiapkan kelompok konsumen yang mengonsumsi data dari 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 berarti proses berhasil
        # jika perlu rollback ke checkpoint sebelumnya, kembalikan 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():
    # Titik akhir Simple Log Service. Contoh ini menggunakan titik akhir wilayah China (Hangzhou). Ganti nilai ini dengan titik akhir yang sebenarnya.
    endpoint = os.environ.get('ALIYUN_LOG_SAMPLE_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', '')

    project = 'your_project'
    logstore = 'your_logstore'

    # Nama kelompok konsumen. Anda tidak perlu membuat kelompok konsumen terlebih dahulu. SDK akan membuatnya secara otomatis.
    consumer_group = 'consumer-group'
    consumer_name = "consumer-group-name"

    # Buat dua konsumen dalam kelompok konsumen untuk mengonsumsi 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("*** mulai mengonsumsi data...")
    client_worker = ConsumerWorker(SPLConsumer, consumer_option=option)
    client_worker.start()

    time.sleep(10000)


if __name__ == '__main__':
    spl_consumer_group()

Jalankan main.py di folder spl_consumer_demo dan lihat hasilnya.
```
python main.py
```

Go

Instal SDK Go. Buat folder bernama spl_demo dan jalankan perintah berikut. Instal SDK Go.

Memerlukan SDK SLS versi v0.1.107 atau lebih baru.
```
go get -u github.com/aliyun/aliyun-log-go-sdk
```

Di folder spl_demo, buat file main.go untuk menyiapkan kelompok konsumen yang mengonsumsi data dari 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 :
// Ini adalah contoh sangat sederhana untuk menarik data dari logstore Anda dan mencetaknya untuk dikonsumsi, termasuk pra-pemrosesan log.

func main() {
  // Titik akhir Simple Log Service. Contoh ini menggunakan titik akhir wilayah China (Hangzhou). Ganti nilai ini dengan titik akhir yang sebenarnya.
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",
    // Opsi ini digunakan untuk inisialisasi, akan diabaikan setelah kelompok konsumen dibuat dan setiap shard mulai dikonsumsi.
    // Bisa berupa "begin", "end", atau "format waktu spesifik dalam timestamp", yaitu waktu penerimaan log.
    CursorPosition: consumerLibrary.END_CURSOR,
    // Processor digunakan untuk pra-pemrosesan log sebelum dikembalikan ke klien, informasi lebih lanjut lihat 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", "menerima sinyal berhenti, mulai menghentikan consumer worker", "consumer worker name", option.ConsumerName)
    consumerWorker.StopAndWait()
  }
}

// Isi logika konsumsi Anda di sini, dan berhati-hatilah agar tidak mengubah parameter fungsi dan nilai kembali,
// jika tidak, Anda akan mendapat error.
func process(shardId int, logGroupList *sls.LogGroupList, checkpointTracker consumerLibrary.CheckPointTracker) (string, error) {
  fmt.Println(shardId, "loggroup", len(logGroupList.LogGroups))
  checkpointTracker.SaveCheckPoint(false)
  return "", nil
}

Jalankan perintah berikut di folder spl_demo untuk menginstal dependensi.
```
go mod tidy
go mod vendor
```
Jalankan fungsi main dan lihat output-nya.
```
go run main.go
```

Prasyarat​

Contoh kode

Java

Python

Go

Prasyarat