ai_qback/tools/ali_log.py

# -*- coding: utf-8 -*-
from aliyun.log import LogClient, PutLogsRequest, LogItem, GetLogsRequest, IndexConfig
import time
import os

accessKeyId = os.getenv("OSS_ACCESS_KEY_ID")
accessKey = os.getenv("OSS_ACCESS_KEY_SECRET")
endpoint = "cn-hangzhou.log.aliyuncs.com"

client = LogClient(endpoint, accessKeyId, accessKey)

project_name = "ai-test-7230"
logstore_name = "test2-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 = int(time.time()) - 3600
to_time = time.time() + 3600


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)


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)


def put_logs(msg: str):
    log_group = []

    log_item = LogItem()
    contents = [
        ('info', msg),
    ]
    log_item.set_contents(contents)
    log_group.append(log_item)
    request = PutLogsRequest(project_name, logstore_name, "", "", log_group, compress=False)
    client.put_logs(request)


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__':
    put_logs("测试")