全部产品
Search
文档中心

日志服务:Python SDK快速入门

更新时间:Jun 12, 2024

本文介绍如何快速使用日志服务Python SDK完成常见操作,包括创建项目(Project)、创建日志库(Logstore)、写入日志和查询日志等。

前提条件

  • 已创建RAM用户并完成授权。具体操作,请参见创建RAM用户并完成授权

  • 已配置环境变量ALIBABA_CLOUD_ACCESS_KEY_IDALIBABA_CLOUD_ACCESS_KEY_SECRET。具体操作,请参见在Linux、macOS和Windows系统配置环境变量

    重要
    • 阿里云账号的AccessKey拥有所有API的访问权限,建议您使用RAM用户的AccessKey进行API访问或日常运维。

    • 强烈建议不要把AccessKey ID和AccessKey Secret保存到工程代码里,否则可能导致AccessKey泄露,威胁您账号下所有资源的安全。

  • 已完成日志服务Python SDK安装。更多信息,请参见安装Python SDK

示例

  • 直接编写Python代码采集日志

    本示例中,创建一个SLSQuickStart.py文件,并调用接口分别完成创建Project、创建Logstore、创建索引、写入日志数据和查询日志数据。示例如下:

    from aliyun.log import LogClient, PutLogsRequest, LogItem, GetLogsRequest, IndexConfig
    import time
    import os
    
    # 本示例从环境变量中获取AccessKey ID和AccessKey Secret。
    accessKeyId = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_ID', '')
    accessKey = os.environ.get('ALIBABA_CLOUD_ACCESS_KEY_SECRET', '')
    # 日志服务的服务接入点。此处以杭州为例,其它地域请根据实际情况填写。
    endpoint = "cn-hangzhou.log.aliyuncs.com" 
    
    # 创建日志服务Client。 
    client = LogClient(endpoint, accessKeyId, accessKey)
    
    # Project名称。
    project_name = "aliyun-test-project"
    # Logstore名称
    logstore_name = "aliyun-test-logstore"
    # 查询语句。
    query = "*| select dev,id from " + logstore_name
    # 索引。
    logstore_index = {'line': {
        'token': [',', ' ', "'", '"', ';', '=', '(', ')', '[', ']', '{', '}', '?', '@', '&', '<', '>', '/', ':', '\n', '\t',
                  '\r'], 'caseSensitive': False, 'chn': False}, 'keys': {'dev': {'type': 'text',
                                                                                 'token': [',', ' ', "'", '"', ';', '=',
                                                                                           '(', ')', '[', ']', '{', '}',
                                                                                           '?', '@', '&', '<', '>', '/',
                                                                                           ':', '\n', '\t', '\r'],
                                                                                 'caseSensitive': False, 'alias': '',
                                                                                 'doc_value': True, 'chn': False},
                                                                         'id': {'type': 'long', 'alias': '',
                                                                                'doc_value': True}}, 'log_reduce': False,
        'max_text_len': 2048}
    
    # from_time和to_time表示查询日志的时间范围,Unix时间戳格式。
    from_time = int(time.time()) - 3600
    to_time = time.time() + 3600
    
    # 创建Project。
    def create_project():
        print("ready to create project %s" % project_name)
        client.create_project(project_name, project_des="")
        print("create project %s success " % project_name)
        time.sleep(60)
    
    # 创建Logstore。
    def create_logstore():
        print("ready to create logstore %s" % logstore_name)
        client.create_logstore(project_name, logstore_name, ttl=3, shard_count=2)
        print("create logstore %s success " % project_name)
        time.sleep(30)
    
    # 创建索引。
    def create_index():
        print("ready to create index for %s" % logstore_name)
        index_config = IndexConfig()
        index_config.from_json(logstore_index)
        client.create_index(project_name, logstore_name, index_config)
        print("create index for %s success " % logstore_name)
        time.sleep(60 * 2)
    
    # 向Logstore写入数据。
    def put_logs():
        print("ready to put logs for %s" % logstore_name)
        log_group = []
        for i in range(0, 100):
            log_item = LogItem()
            contents = [
                ('dev', 'test_put'),
                ('id', str(i))
            ]
            log_item.set_contents(contents)
            log_group.append(log_item)
        request = PutLogsRequest(project_name, logstore_name, "", "", log_group, compress=False)
        client.put_logs(request)
        print("put logs for %s success " % logstore_name)
        time.sleep(60)
    
    
    # 通过SQL查询日志。
    def get_logs():
        print("ready to query logs from logstore %s" % logstore_name)
        request = GetLogsRequest(project_name, logstore_name, from_time, to_time, query=query)
        response = client.get_logs(request)
        for log in response.get_logs():
            for k, v in log.contents.items():
                print("%s : %s" % (k, v))
            print("*********************")
    
    
    if __name__ == '__main__':
        # 创建Project。
        create_project()
        # 创建Logstore。
        create_logstore()
        # 创建索引。
        create_index()
        # 向Logstore写入数据。
        put_logs()
        # 通过SQL查询日志。
        get_logs()

    返回结果示例如下:

    ready to create project aliyun-test-project
    create project aliyun-test-project success
    ready to create logstore aliyun-test-logstore
    create logstore aliyun-test-project success
    ready to create index for aliyun-test-logstore
    create index for aliyun-test-logstore success
    ready to put logs for aliyun-test-logstore
    put logs for aliyun-test-logstore success
    ready to query logs from logstore aliyun-test-logstore
    dev : test_put
    id : 0
    *********************
    dev : test_put
    id : 1
    *********************
    dev : test_put
    id : 2
    *********************
    dev : test_put
    id : 3
    *********************
    ........

    更多示例代码,请参见Aliyun Log Python SDK

  • 通过Logtail采集Python日志

    通过Logtail方式,以采集Python的logging模块日志为例,采集Python日志。更多信息,请参见采集Python日志

相关文档

  • 在调用API接口过程中,若服务端返回结果中包含错误信息,则表示调用API接口失败。您可以参考API错误码对照表查找对应的解决方法。更多信息,请参见API错误处理对照表

  • 阿里云OpenAPI开发者门户提供调试、SDK、示例和配套文档。通过OpenAPI,您无需手动封装请求和签名操作,就可以快速对日志服务API进行调试。更多信息,请参见OpenAPI开发者门户

  • 为满足越来越多的自动化日志服务配置需求,日志服务提供命令行工具CLI(Command Line Interface)。更多信息,请参见日志服务命令行工具CLI

  • 更多示例代码,请参见Aliyun Log Python On GitHub