云日志服务Python SDK

云日志服务SDK提供了Python语言上报日志的一系列方法，方便用户直接使用编码方式上报日志到云日志服务后台。

传输协议

HTTPS

使用前提

参考注册华为账号并开通华为云中操作，完成注册。

确认云日志服务的区域，请用户根据所在区域，选择RegionName。
获取华为账号的AK/SK。
获取华为云账号的项目ID（project id），详细步骤请参见API凭证。
获取需要上报到LTS的日志组ID和日志流ID。
Python SDK适用于python 3.10.1及以上版本。

使用说明

当用户修改权限后，权限信息在一天后生效。通过SDK上报日志到LTS的时间相距当前时间不超过2天，否则上报日志会被LTS删除。

注意事项

认证用的ak和sk硬编码到代码中或者明文存储都有很大的安全风险，建议在配置文件或者环境变量中密文存放，使用时解密，确保安全。

安装Python SDK

获取LTS Python SDK包源码。

git clone https://gitee.com/lordstar-habile/huaweicloud-lts-python-sdk.git

安装相关依赖。

pip3 install requests
pip3 install loguru
pip3 install six

编写上报日志代码。

可以自定义发送成功与失败时的回调方法，实现方式请参考如下命令：创建CallBack对象，实现invoke_on_success和invoke_on_failed方法，并且在发送日志时直接使用。

class CallBack(object):
    def invoke_on_success(self, result):
        pass

    def invoke_on_failed(self, result):
        pass

示例如下：

import random
import time
from concurrent.futures import ThreadPoolExecutor
from loguru import logger
from producer.core.producer import Producer
from producer.callback.call_back import CallBack
from producer.model import lts_store
from producer.model.config import Config
from producer.utils import common
def generate_random_str(str_len=16):
    """
    生成一个指定长度的随机字符串
    """
    random_str = ''
    base_str = 'ABCDEFGHIGKLMNOPQRSTUVWXYZabcdefghigklmnopqrstuvwxyz0123456789'
    length = len(base_str) - 1
    for i in range(str_len):
        random_str += base_str[random.randint(0, length)]
    return random_str
def send_log_single_producer():
    thread_num = 100
    threads = []
    begin_ms = common.current_time_ms()
    submit_send_log_thread_pool = ThreadPoolExecutor(max_workers=thread_num, thread_name_prefix="send_log_thread")
    config = Config()
    config.endpoint = ""
    config.access_key = ""
    config.access_secret = ""
    config.region_id = ""
    config.project_id = ""
    producer = Producer.init_producer(config)
    producer.start_producer()
    group_id = ""
    stream_id = ""

    log_content = generate_random_str(1024)
    log = [log_content]
    labels = {"keyA": "valueA"}
    log_p = lts_store.generate_log(log, labels)
    for i in range(thread_num):
        new_thread = submit_send_log_thread_pool.submit(send_log_function, producer, group_id, stream_id, log_p)
        threads.append(new_thread)
    for t in threads:
        t.result()
    end_ms = common.current_time_ms()
    logger.info("end send log, cost [{}]ms", end_ms - begin_ms)
def send_log_function(one_producer, group_id, stream_id, log_p):
    for i in range(2000):
        one_producer.send_log_with_callback(group_id, stream_id, log_p)
class MyCallBack(CallBack):
    def invoke_on_success(self, result):
        logger.info("send log success")
    def invoke_on_failed(self, result):
        logger.error("send log success")
if __name__ == '__main__':
    send_log_single_producer()
    time.sleep(1000)

配置参数说明

producer config参数说明。

参数名称	描述	类型	是否需要填写	默认值
projectId	华为云账号的项目ID（project id）。	String	必填	-
ak	华为云账号的AK。	String	必填	-
sk	华为云账号的SK。	String	必填	-
region	云日志服务的区域。	String	必填	-
endpoint	上报目的。	String	必填	-
logGroup	LTS的日志组ID。	string	必填	-
logStream	LTS的日志流ID。	string	必填	-

日志生成GenerateLog方法类参数说明。单次上报条数小于4096条，且小于512K。

参数名称	描述	类型	是否需要填写
contents	批量日志内容。	[]string	是
labels	日志标签，map格式。	map[string]string	是

日志上报SendLog方法类参数说明。

表1 SendLog方法类参数说明
参数名称	描述	类型	是否需要填写
groupId	日志组id	string	是
streamId	日志流id	string	是
log	日志结构	class Log	是

参数获取方式

上线区域表，使用华为云主机时参考以下信息：

表2 区域表
区域名称	区域	终端节点
华北-北京四	cn-north-4	https://lts-access.cn-north-4.myhuaweicloud.com:8102
华东-上海一	cn-east-3	https://lts-access.cn-east-3.myhuaweicloud.com:8102
华南-广州	cn-south-1	https://lts-access.cn-south-1.myhuaweicloud.com:8102
亚太-新加坡	ap-southeast-3	https://lts-access.ap-southeast-3.myhuaweicloud.com:8102