Gunakan Python SDK untuk mengonsumsi log dari kelompok konsumen - Simple Log Service

Menggunakan kelompok konsumen (ConsumerGroup) untuk mengonsumsi log menawarkan keunggulan signifikan. Anda dapat fokus pada logika bisnis tanpa perlu mengelola detail implementasi seperti load balancing dan failover antar konsumen. Topik ini menjelaskan cara menggunakan kelompok konsumen untuk mengonsumsi log dengan kode Python.

Ikhtisar

Logstore berisi beberapa shard. Kelompok konsumen mengonsumsi data dengan menetapkan shard-shard tersebut ke konsumennya berdasarkan aturan berikut:

Satu shard hanya ditetapkan ke satu konsumen dalam satu waktu.
Satu konsumen dapat ditetapkan beberapa shard.

Ketika konsumen baru bergabung ke dalam kelompok konsumen, shard akan diseimbangkan ulang di antara semua konsumen untuk memastikan load balancing. Aturan yang sama tetap berlaku.

Prasyarat

Layanan Log Sederhana telah diaktifkan.
SDK Layanan Log Sederhana untuk Python telah diinisialisasi.

SDK Layanan Log Sederhana untuk Python telah diinstal. Untuk informasi selengkapnya, lihat Instal SDK Layanan Log Sederhana untuk Python.
Layanan Log Sederhana telah diaktifkan. Untuk informasi selengkapnya, lihat Aktifkan Layanan Log Sederhana.

Perhatian

Pada contoh ini, endpoint publik Layanan Log Sederhana untuk wilayah China (Hangzhou) digunakan. Endpoint: https://cn-hangzhou.log.aliyuncs.com.

Jika Anda ingin mengakses Layanan Log Sederhana dari layanan Alibaba Cloud lain yang berada di wilayah yang sama dengan proyek Anda, Anda dapat menggunakan endpoint internal Layanan Log Sederhana, yaitu https://cn-hangzhou-intranet.log.aliyuncs.com.

Untuk informasi selengkapnya mengenai wilayah dan endpoint yang didukung oleh Layanan Log Sederhana, lihat Endpoint.

Contoh kode

Kode berikut menunjukkan cara membuat Logstore, menulis log ke Logstore, membuat kelompok konsumen, dan mengonsumsi log.

import os
import time

from aliyun.log.consumer import *
from aliyun.log import *
from threading import RLock


class SampleConsumer(ConsumerProcessorBase):
    shard_id = -1
    last_check_time = 0
    log_results = []
    lock = RLock()

    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

            with SampleConsumer.lock:
                SampleConsumer.log_results.append(log_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
        # Untuk kembali 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()


test_item_count = 20

# Menulis log ke Logstore.
def _prepare_data(client, project, logstore):
    topic = 'python-ide-test'
    source = ''

    for i in range(0, test_item_count):
        logitemList = []  # daftar LogItem

        contents = [
            ('user', 'magic_user_' + str(i)),
            ('avg', 'magic_age_' + str(i))
        ]
        logItem = LogItem()
        logItem.set_time(int(time.time()))
        logItem.set_contents(contents)

        logitemList.append(logItem)
        # Menulis log ke Logstore.
        request = PutLogsRequest(project, logstore, topic, source, logitemList)

        response = client.put_logs(request)
        print("successfully put logs in logstore")


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 sample_consumer_group():
    # Endpoint Layanan Log Sederhana. Contoh ini menggunakan endpoint China (Hangzhou). Ganti dengan endpoint aktual Anda.
    endpoint = os.environ.get('ALIYUN_LOG_SAMPLE_ENDPOINT', 'cn-hangzhou.log.aliyuncs.com')

    # Contoh ini mengambil ID AccessKey dan Rahasia AccessKey dari variabel lingkungan.
    accessKeyId = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')
    accessKey = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')

    # Nama proyek. SDK tidak membuat proyek dalam contoh ini. Anda harus membuat proyek sebelum menjalankan kode.
    project = os.environ.get('ALIYUN_LOG_SAMPLE_PROJECT', 'ali-test-project-python')

    # Nama Logstore. SDK secara otomatis membuat Logstore dalam contoh ini. Anda tidak perlu membuatnya terlebih dahulu.
    logstore = 'ali-test-logstore'

    # Nama kelompok konsumen. Anda tidak perlu membuatnya terlebih dahulu. SDK secara otomatis membuatnya.
    consumer_group = 'consumer-group-1'
    consumer_name1 = "consumer-group-1-A"
    consumer_name2 = "consumer-group-1-B"
    token = ""

    if not logstore:
        logstore = 'consumer_group_test_' + str(time.time()).replace('.', '_')

    assert endpoint and accessKeyId and accessKey and project, ValueError("endpoint/access_id/key dan "
                                                                          "project tidak boleh kosong")

    # Membuat Logstore.
    client = LogClient(endpoint, accessKeyId, accessKey, token)
    ret = client.create_logstore(project, logstore, 2, 4)
    print("successfully create logstore")

    time.sleep(60)
    SampleConsumer.log_results = []

    try:
        # Menulis log ke Logstore.
        _prepare_data(client, project, logstore)

        # Membuat dua konsumen dalam kelompok konsumen untuk mengonsumsi data.
        option1 = LogHubConfig(endpoint, accessKeyId, accessKey, project, logstore, consumer_group,
                               consumer_name1, cursor_position=CursorPosition.BEGIN_CURSOR, heartbeat_interval=6,
                               data_fetch_interval=1)
        option2 = LogHubConfig(endpoint, accessKeyId, accessKey, project, logstore, consumer_group,
                               consumer_name2, cursor_position=CursorPosition.BEGIN_CURSOR, heartbeat_interval=6,
                               data_fetch_interval=1)

        print("*** start to consume data...")
        client_worker1 = ConsumerWorker(SampleConsumer, consumer_option=option1)
        client_worker1.start()
        client_worker2 = ConsumerWorker(SampleConsumer, consumer_option=option2)
        client_worker2.start()

        sleep_until(300, lambda: len(SampleConsumer.log_results) >= test_item_count)

        print("*** consumer group status ***")
        ret = client.list_consumer_group(project, logstore)
        print("successfully list consumergroup")

        for c in ret.get_consumer_groups():
            ret = client.get_check_point_fixed(project, logstore, c.get_consumer_group_name())
            print("successfully get checkpoint fixed")

        print("*** stopping workers")
        client_worker1.shutdown()
        client_worker2.shutdown()

    finally:
        # pembersihan
        # ret = client.delete_logstore(project, logstore)
        ret = client.list_logstore(project, logstore)
        print("successfully list logstore")

    # validasi
    ret = str(SampleConsumer.log_results)
    print("*** get content:")
    print(ret)

    assert 'magic_user_0' in ret and 'magic_age_0' in ret \
           and 'magic_user_' + str(test_item_count-1) in ret \
           and 'magic_age_' + str(test_item_count-1) in ret


if __name__ == '__main__':
    sample_consumer_group()

Hasil yang diharapkan:

successfully create logstore
successfully put logs in logstore
successfully put logs in logstore
successfully put logs in logstore
successfully put logs in logstore
successfully put logs in logstore
......
*** start to consume data...
{'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123452, 'user': 'magic_user_0', 'avg': 'magic_age_0'}]}
{'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123453, 'user': 'magic_user_1', 'avg': 'magic_age_1'}]}
{'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123453, 'user': 'magic_user_3', 'avg': 'magic_age_3'}]}
{'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123453, 'user': 'magic_user_5', 'avg': 'magic_age_5'}]}
{'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123453, 'user': 'magic_user_6', 'avg': 'magic_age_6'}]}
......
*** consumer group status ***
successfully list consumergroup
successfully get checkpoint fixed
*** stopping workers
successfully list logstore
*** get content:
[{'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123452, 'user': 'magic_user_0', 'avg': 'magic_age_0'}]}, {'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123453, 'user': 'magic_user_1', 'avg': 'magic_age_1'}]}, {'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123453, 'user': 'magic_user_3', 'avg': 'magic_age_3'}]}, {'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123453, 'user': 'magic_user_5', 'avg': 'magic_age_5'}]}, {'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123453, 'user': 'magic_user_6', 'avg': 'magic_age_6'}]}, {'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123453, 'user': 'magic_user_8', 'avg': 'magic_age_8'}]}, {'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123453, 'user': 'magic_user_9', 'avg': 'magic_age_9'}]}, {'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123453, 'user': 'magic_user_2', 'avg': 'magic_age_2'}]}, {'topic': 'python-ide-test', 'source': '47.100.XX.XX', 'logs': [{'time': 1672123453, 'user': 'magic_user_11', 'avg': 'magic_age_11'}]}, ......}]

Referensi

Selain SDK aslinya, SLS juga mendukung SDK Alibaba Cloud umum. Untuk informasi selengkapnya, lihat Layanan Log Sederhana_Pusat SDK_Alibaba Cloud OpenAPI Explorer.
CLI Layanan Log Sederhana.
Alibaba Cloud Simple Log Service SDK for Python di GitHub.