Synchronize index configuration, Synchronize index configuration

Configuration file ​

Save the following JSON as ./sls_tools/sync_logstore_index_config.json and update the field values for your environment.

{
  "target": {
    "endpoint": "cn-chengdu.log.aliyuncs.com",
    "project": "sls-ml-demo",
    "logstore": "cdn_access_log"
  },
  "destination": [
    {
      "endpoint": "cn-chengdu.log.aliyuncs.com",
      "project": "sls-ml-demo",
      "logstore": "test_temp"
    }
  ]
}

• target: The source Logstore whose index configuration will be read. Specify the endpoint, project, and logstore for the source.

  Sub-field	Type	Description
  endpoint	String	Public endpoint for the source Logstore's region — for example, cn-chengdu.log.aliyuncs.com. See endpoint.
  project	String	Name of the SLS project containing the source Logstore.
  logstore	String	Name of the source Logstore.

• destination: An array of one or more destination Logstore configurations. The index configuration is fully replaced on each destination — existing custom field configurations will be overwritten.

  Sub-field	Type	Description
  endpoint	String	Public endpoint for the destination Logstore's region.
  project	String	Name of the SLS project containing the destination Logstore.
  logstore	String	Name of the destination Logstore.

Sample code ​

The script reads the index configuration from the source Logstore and applies it to each destination. For each destination, it calls update_index if an index exists or create_index if none has been created yet.

# -*- coding: utf-8 -*-
import json
from aliyun.log import *

# The permissions required to access the Logstores in the configuration file.
global_ak_id = ""
global_ak_key = ""

client_map = {}

def get_sls_client(endpoint: str) -> LogClient:
    sls_client = LogClient(endpoint, global_ak_id, global_ak_key)
    if endpoint in client_map.keys():
        return client_map[endpoint]
    client_map[endpoint] = sls_client
    return sls_client

def get_logstore_index(endpoint: str, project: str, logstore: str) -> IndexConfig:
    sls_client = get_sls_client(endpoint)
    get_index_config_resp = sls_client.get_index_config(project, logstore)
    index_config = get_index_config_resp.get_index_config()
    return index_config

def set_logstore_index(endpoint: str, project: str, logstore: str, target_index: IndexConfig):
    """
    Check whether an index configuration is created for the Logstore. If no index configuration is created, create one. Otherwise, update the index configuration.
    """
    is_need_create_index = False
    sls_client = get_sls_client(endpoint)
    try:
        update_index_resp = sls_client.update_index(project, logstore, target_index)
        update_index_resp.log_print()
    except LogException as logE:
        if logE.get_error_code() == "InternalServerError":
            msgE = logE.get_error_message()
            if msgE == "log store index is not created":
                is_need_create_index = True
            else:
                raise logE
        else:
            raise logE
    if is_need_create_index:
        create_index_resp = sls_client.create_index(project, logstore, target_index)
        create_index_resp.log_print()

def check_logstore_item(store_item: dict):
    key_names = ["project", "logstore", "endpoint"]
    for key in key_names:
        if key not in store_item.keys():
            raise ValueError(f"logstore config miss key {key}")
        if len(store_item[key]) == "":
            raise ValueError(f"logstore config miss key [{key}]")

if __name__ == "__main__":
    # Specify the path of the configuration file.
    sync_store_config_path = "./sls_tools/sync_logstore_index_config.json"
    with open(sync_store_config_path, "r") as reader:
        sync_map = json.load(reader)
    target_logstore = sync_map["target"]
    check_logstore_item(target_logstore)

    destinations = sync_map["destination"]
    target_index_config = get_logstore_index(
        target_logstore["endpoint"], target_logstore["project"], target_logstore["logstore"])
    for dest_item in destinations:
        try:
            check_logstore_item(dest_item)
            set_logstore_index(dest_item["endpoint"], dest_item["project"], dest_item["logstore"], target_index_config)
            print(f"dest logstore {dest_item} update index config.")
        except Exception as e:
            print(f"dest logstore {dest_item}, failed {e}")

Each destination Logstore prints a confirmation line on success or an error message on failure.