Help Center/ Workspace/ Best Practices/ Interconnecting with LTS to Collect Workspace Monitoring Metrics

Updated on 2026-05-18 GMT+08:00

Interconnecting with LTS to Collect Workspace Monitoring Metrics

Scenarios

Workspace provides multiple types of desktop statistics. For instance, you can learn desktop statuses by users, user groups, terminal types, and time.

Prerequisites

You have purchased VPC endpoints by referring to Buying a VPC Endpoint.

To enable a function to access resources in a VPC, you need to purchase VPC endpoints and configure an agency with VPC management permissions for the function. You need to purchase the following endpoints:
- com.myhuaweicloud.cn-south-1.ces
- com.myhuaweicloud.cn-south-1.lts-access
- com.myhuaweicloud.cn-south-1.workspace
When purchasing VPC endpoints, make sure to select the same VPC for all of them.
If you configure public network access when creating a function, you do not need to create VPC endpoints.

Step 1: Creating and Configuring a Function

Create an event function by referring to Creating an Event Function.

When creating a function, set Runtime to Python 3.9.

Configuring Basic Settings

Dynamically adjust the timeout interval based on the service logic, data volume, and execution frequency. You are advised to set the timeout interval to 60 seconds in most service scenarios.

Click the name of the function you created. The basic information page of the function is displayed.
Click the Configuration tab and set Execution Timeout (s) to 60.

Figure 1 Setting the executing timeout interval
Click Save.

Creating the Function Trigger

Create a trigger based on the service. You are advised to use a cron expression to define trigger rules. For example, you can schedule the screen recording reporting to run every hour. As desktop recording relies on data reported by Cloud Eye Agent, it is recommended that the previous hour's data be reported at the 40th minute of each hour. User recording can be reported once in the early morning every day.

On the function details page, choose Configuration > Triggers > Create Trigger.

Figure 2 Creating a timer trigger

**Table 1** Trigger parameters
Parameter	Description	Example
Trigger Type	Select Timer.	Timer
Timer Name	Specify a custom trigger name. The name can contain letters, digits, underscores (_), and hyphens (-). It must start with a letter and cannot exceed 64 characters.	Timer-Workspace
Trigger Rule	Define the trigger rule. You can select Fixed rate or Cron expression. Fixed rate: The function is triggered at a fixed interval of minutes, hours, or days, with maximum values of 60, 24, and 30, respectively. Cron expression: The function is triggered based on a complex rule. For example, you can set a function to be executed at 08:30:00 every Monday to Friday. For more information, see Cron Expression Rules.	Cron expression 40 * * * * ?
Enable Trigger	Determine whether to enable the timer trigger. If this parameter is disabled, the function will not be triggered at the specified frequency.	Enabled
Additional Information	This is an optional parameter. The additional information you configure will be put into the user_event field of the timer event source. For details, see Supported Trigger Events.	-

Click OK.

Configuring Function Permissions

On the basic function information page, choose Configuration > Permissions.
Click Create Agency on the right of Agency.

Figure 3 Creating an agency
On the IAM console, choose Agencies from the navigation pane, and click Create Agency in the upper right corner.

Figure 4 Creating an agency
Set agency parameters
- Agency Name: Enter a custom agency name.
- Agency Type: Select Cloud service.
- Cloud Service: Select FunctionGraph.
- Validity Period: Retain the default value Unlimited.
  Figure 5 Configuring an agency
Click Finish.
In the displayed dialog box, click Authorize.
On the displayed page, click Create Policy.

Figure 6 Creating a policy
Enter a policy name and select JSON for Policy View.

Figure 7 Configuring a custom policy

Copy the following content into the text box.

{
    "Version": "1.1",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "workspace:users:list",
                "workspace:desktops:listDetail",
                "workspace:connections:securityList",
                "workspace:screenRecord:listOperations",
                "workspace:statistics:listUserMetrics",
                "workspace:screenRecord:list"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "ces:metricData:list"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "lts:logs:upload"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "vpc:ports:delete",
                "vpc:ports:get",
                "vpc:ports:create",
                "vpc:vpcs:get",
                "vpc:subnets:get",
                "vpc:publicIps:list",
                "vpc:securityGroups:get"
            ]
        },
        {
            "Effect": "Allow",
            "Action": [
                "nat:natGateways:list",
                "nat:snatRules:list"
            ]
        }
    ]
}

Click Create.
Select the created agency and click Next.
Retain the default value under Scope and click OK.
On the basic function information page, choose Configuration > Permissions.
Select the created agency from the Agency drop-down list.

Figure 8 Permission settings
Click Save.

Configuring Function Network Settings

You can configure the network to allow a function to access resources in a VPC or from a public network. The function accesses the public network with the default NIC and shared bandwidth. For the best possible bandwidth, performance, and reliability, enable VPC access.

On the basic function information page, choose Configuration > Network.
On the Network page, enable VPC access and configure required parameters.

Figure 9 Enabling VPC access
- VPC: Select the VPC EP created in Prerequisites from the drop-down list.
- Subnet: The subnet in the VPC is used by default.
Click Save.

Configuring Environment Variables

On the basic function information page, choose Configuration > Environment Variables.
Click Edit Environment Variables. On the displayed page, choose JSON.

Figure 10 Configuring environment variables
Configure environment variables based on the following template, edit the variables based on the description, and copy them into the JSON text box.

Replace the values of the following parameters:
```
{
    "region": "cn-south-1",
    "ip": "100.xxx.xxx.xxx",
    "project_id": "xxx",
    "log_group_id": "xxx",
    "log_stream_id": "xxx"
}
```
- region: Set this parameter based on the current site. For example, set it to cn-south-1 for CN South-Guangzhou.
- ip: Obtain the value by parsing the domain name for accessing Workspace.
  1. Press Win+R. In the Run dialog box, enter cmd and press Enter.
  2. Run the following commands to obtain the IP address:
    ping workspace.{region_id}.myhuaweicloud.com
    
    Replace region_id with the actual region ID.
- project_id: Obtain the value by referring to Obtaining a Project ID.
- log_group_id: Obtain the value by referring to 2.
- log_stream_id: Obtain the value by referring to 2.

Executing Code

On the basic function information page, click the Code tab.
Choose File > New File.

Figure 11 Creating a file
Rename the new file index.py.

Figure 12 Renaming a file
Test the required code and copy the sample code in Step 3: Deploying Code to index.py.

Modify the code based on site requirements.
Click Test.

Figure 13 Testing code
Configure LTS logs by referring to 3.

Step 3: Deploying Code

The following sample code illustrates how to use the FunctionGraph to periodically query Workspace API results, including statistics about user connections, desktop performance, and screen recording operations, and to report the data to LTS. You can view the query results on the LTS console.

Sample Code 1: Reporting User Connection Statistics

FunctionGraph invokes handler on a schedule to query user connection statistics from the Workspace APIs and report them to LTS. Queried statistics include the username, user description, user group, user expiration time, terminal type, enterprise project, the number of total connections, successful connections, and failed connections, usage duration, and average latency. The statistics are collected by calendar day.

Workspace API reference: Querying Users, Querying User Usage Statistics, Querying User Details, and Querying Login Information
LTS API reference: API Access Overview

The sample code is as follows: (Modify the code based on the site requirements.)

# -*- coding:utf-8 -*-
import datetime
import json
import time
import requests
from datetime import datetime, timedelta
from collections import defaultdict
from urllib.parse import urlencode

def get_workspace_headers(region_name, token):
    headers = {"Host": "workspace." + region_name + ".myhuaweicloud.com",
               "X-Auth-Token": token,
               "Content-Type": "application/json"}
    return headers


def list_user_usage_metric_for_region(region_name, ip, token, project_id, start_time, end_time):
    base_url = "https://" + ip + "/v2/" + project_id + "/statistics/metrics/users"

    headers = get_workspace_headers(region_name, token)

    params = {}
    if start_time:
        params['start_time'] = start_time.strip()
    if end_time:
        params['end_time'] = end_time.strip()

    limit = 100
    params['limit'] = limit
    offset = 0
    params['offset'] = offset

    all_users = []
    total_count = 0

    while True:
        params['offset'] = offset

        query_string = urlencode(params)
        full_url = f"{base_url}?{query_string}" if query_string else base_url

        response = requests.request("GET", full_url, headers=headers, verify=False)
        response_data = json.loads(response.text)
        print ("list_user_usage_metric response.text content preview:", response.text)
        items = response_data.get("items", [])
        current_total = response_data.get("count", len(items)) # fallback: usage length

        if total_count == 0:
            total_count = current_total
        all_users.extend(items)
        if len(all_users) >= total_count:
            break
        offset += limit

    return {
        "count": total_count,
        "items": all_users
    }


def list_login_records(region_name, ip, token, project_id, start_time, end_time):
    base_url = "https://" + ip + "/v2/" + project_id + "/connections/desktops"

    headers = get_workspace_headers(region_name, token)

    params = {}

    if start_time:
        params['start_time'] = start_time.strip()
    if end_time:
        params['end_time'] = end_time.strip()

    # Construct a complete URL.
    query_string = urlencode(params)
    if query_string:
        full_url = base_url + "?" + query_string
    else:
        full_url = base_url

    print(full_url)

    response = requests.request("GET", full_url, headers=headers, verify=False)

    return json.loads(response.text)


def list_users(region_name, ip, token, project_id, user_name_list):
    base_url = "https://" + ip + "/v2/" + project_id + "/users"

    headers = get_workspace_headers(region_name, token)

    params = {}

    if user_name_list:
        params['user_names'] = ",".join(user_name_list)

    query_string = urlencode(params)
    if query_string:
        full_url = base_url + "?" + query_string
    else:
        full_url = base_url

    response = requests.request("GET", full_url, headers=headers, verify=False)

    return json.loads(response.text)


def fetch_users_in_batches(region_name, ip, token, project_id, user_name_list, batch_size=100):
    all_responses = []
    merged_responses = {
        "count": 0,
        "users": []
    }
    for i in range(0, len(user_name_list), batch_size):
        batch = user_name_list[i:i + batch_size]
        response = list_users(region_name, ip, token, project_id, batch)
        all_responses.append(response)

    for resp in all_responses:
        merged_responses["count"] += resp["total_count"]
        merged_responses["users"].extend(resp["users"])

    return merged_responses


def report_lts(region_name, token, project_id, log_group_id, log_stream_id, report_items):
    url = ("https://lts-access." + region_name + ".myhuaweicloud.com:8102/v2/" + project_id + "/lts/groups/"
           + log_group_id + "/streams/" + log_stream_id + "/tenant/contents")
    headers = {"X-Auth-Token": token,
               "Content-Type": "application/json"}
    print("url:", url)
    str_report_items = list(map(lambda x: json.dumps(x), report_items))
    payload = {
        "log_time_ns": time.time_ns(),
        "contents": str_report_items,
        "labels": {
            "tag": "workspace"
        }
    }
    response = requests.post(
        url=url,
        headers=headers,
        json=payload,
        verify=False
    )

    print ("report_lts_response content preview:," response.text)
    return {
        "body": ""
    }


def generate_user_status_dict(user_detail_list):
    result = {}
    users = user_detail_list.get("users", [])

    for user in users:
        username = user.get("user_name")
        if not username:
            continue

        is_locked = user.get("locked", False)
        is_disabled = user.get("disabled", False)

        if not is_locked and not is_disabled:
            status = "normal"
        else:
            status = ",".join(
                part for part in ["disabled" if is_disabled else None,
                                  "locked" if is_locked else None]
                if part
            )
        result[username] = status

    return result

def handler(event, context):
    print("START START START")

    token = context.getToken()

    region = context.getUserData('region')
    ip = context.getUserData('ip')
    project_id = context.getUserData('project_id')
    log_group_id = context.getUserData('log_group_id')
    log_stream_id = context.getUserData('log_stream_id')

    now = datetime.utcnow()

    yesterday = now.date() - timedelta(days=1)
    start_of_yesterday = datetime.combine(yesterday, datetime.min.time())  # 00:00:00
    end_of_yesterday = datetime.combine(yesterday, datetime.max.time())  # 23:59:59.999999
    end_of_yesterday = end_of_yesterday.replace(microsecond=0) # The time format is standardized to omit microseconds, expressed as 23:59:59.

    start_time_str = start_of_yesterday.strftime("%Y-%m-%dT%H:%M:%SZ")
    end_time_str = end_of_yesterday.strftime("%Y-%m-%dT%H:%M:%SZ")

    login_records = list_login_records(region, ip, token, project_id, start_time_str, end_time_str)

    user_rtt_dict = defaultdict(list)
    terminal_type_dict = defaultdict(set)

    for record in login_records["records"]:
        username = record["user_name"]
        network_rtt = record.get("network_rtt", 0)
        terminal_type = record.get("terminal_type", "")
        user_rtt_dict[username].append(network_rtt)
        terminal_type_dict[username].add(terminal_type)
    user_rtt_avg_dict = {username: sum(rtts) / len(rtts) for username, rtts in user_rtt_dict.items()}

    user_usage_metric_list = list_user_usage_metric_for_region(region, ip, token, project_id, start_time_str,
                                                               end_time_str)

    unique_user_name_set = set()
    for item in user_usage_metric_list["items"]:
        unique_user_name_set.add(item["username"].split("@")[0])

    user_detail_list = fetch_users_in_batches(region, ip, token, project_id, list(unique_user_name_set))

    user_info_dict = {
        user_detail["user_name"]: {
            "group_names": sorted(user_detail.get("group_names", [])),
            "account_expires": user_detail.get("account_expires", ""),
            "description": user_detail.get("description", "")
        }
        for user_detail in user_detail_list["users"]
    }
    user_status_dict = generate_user_status_dict(user_detail_list)

    result_list = []

    for item in user_usage_metric_list["items"]:
        flat_item = {
            "username": item["username"].split("@")[0],
            "enterprise_project_id": item["enterprise_project_id"]
        }

        # Traverse the metric list and decompose each metric into key-value pairs.
        for metric in item["metric"]:
            name = metric["metric_name"]
            value = metric["metric_value"]
            if isinstance(value, float) and value.is_integer():
                flat_item[name] = int(value)
            else:
                flat_item[name] = value

        flat_item["avg_network_rtt"] = user_rtt_avg_dict.get(item["username"])
        flat_item["terminal_type"] = list(terminal_type_dict.get(item["username"], set()))
        flat_item["user_status"] = user_status_dict.get(item["username"].split("@")[0])
        flat_item["user_group_names"] = user_info_dict.get(item["username"].split("@")[0], {}).get("group_names")
        expires = user_info_dict.get(item["username"].split("@")[0], {}).get("account_expires")
        flat_item["account_expires"] = "Never expires" if expires == "0" or expires is None else expires
        flat_item["description"] = user_info_dict.get(item["username"].split("@")[0], {}).get("description")
        result_list.append(flat_item)

    report_lts(region, token, project_id, log_group_id, log_stream_id, result_list)

    return {
        "statusCode": 200,
        "isBase64Encoded": False,
        "body": result_list,
        "headers": {"Content-Type": "application/json"}
    }

Sample Code 2: Reporting Desktop Performance Statistics

FunctionGraph invokes handler on a schedule to query desktop performance statistics from the Workspace APIs and report them to LTS. Queried statistics include the desktop name, desktop ID, desktop IP address, bound user, specifications, system disk size (GB), data disk size (GB), enterprise project, CPU usage, memory usage, disk usage, and monitoring data events. The statistics are collected and reported by the hour.

Workspace API reference: Listing Desktop Details
Cloud Eye API reference: Querying Monitoring Data of Multiple Metrics
LTS API reference: API Access Overview

The sample code is as follows: (Modify the code based on the site requirements.)

# -*- coding:utf-8 -*-
import datetime
import json
import time
from itertools import product
import requests
from datetime import datetime, timezone, timedelta
from urllib.parse import urlencode

def get_workspace_headers(region_name, token):
    headers = {"Host": "workspace." + region_name + ".myhuaweicloud.com",
               "X-Auth-Token": token,
               "Content-Type": "application/json"}
    return headers

def list_desktops_detail(region_name, ip, token, project_id):
    base_url = "https://" + ip + "/v2/" + project_id + "/desktops/detail"
    headers = get_workspace_headers(region_name, token)

    params = {}
    limit = 100
    params['limit'] = limit
    offset = 0
    params['offset'] = offset

    all_desktops = []
    total_count = 0

    while True:
        params['offset'] = offset
        query_string = urlencode(params)
        full_url = f"{base_url}?{query_string}" if query_string else base_url
        response = requests.request("GET", full_url, headers=headers, verify=False)
        response_data = json.loads(response.text)
        desktops = response_data.get("desktops", [])
        current_total = response_data.get("total_count", len(desktops))
        if total_count == 0:
            total_count = current_total
        all_desktops.extend(desktops)
        if len(all_desktops) >= total_count:
            break
        offset += limit

    return {
        "total_count": total_count,
        "desktops": all_desktops
    }


def list_metric_data(region_name, token, project_id, start_time, end_time, instance_ids, metric_names):
    url = "https://ces." + region_name + ".myhuaweicloud.com" + "/V1.0/" + project_id + "/batch-query-metric-data"

    headers = {
        "X-Auth-Token": token,
        "Content-Type": "application/json"
    }

    all_responses = []

    def chunked(lst, n):
        for i in range(0, len(lst), n):
            yield lst[i:i + n]

    for batch_instance_ids in chunked(instance_ids, 100):
        payload = {
            "from": start_time,
            "to": end_time,
            "period": "3600",
            "filter": "average",
            "metrics": []
        }

        for metric_name, instance_id in product(metric_names, batch_instance_ids):
            metric = {
                "namespace": "SYS.Workspace",
                "dimensions": [
                    {
                        "name": "instance_id",
                        "value": instance_id
                    }
                ],
                "metric_name": metric_name
            }
            payload["metrics"].append(metric)

        response = requests.post(url=url, headers=headers, json=payload, verify=False)

        try:
            result = response.json()
        except json.JSONDecodeError:
            result = {"error": "Invalid JSON response", "raw": response.text}
        all_responses.append(result)

    return flatten_and_transform(all_responses)

def report_lts(region_name, token, project_id, log_group_id, log_stream_id, report_items):
    url = ("https://lts-access." + region_name + ".myhuaweicloud.com:8102/v2/" + project_id + "/lts/groups/"
           + log_group_id + "/streams/" + log_stream_id + "/tenant/contents")

    headers = {"X-Auth-Token": token,
               "Content-Type": "application/json"}

    str_report_items = list(map(lambda x: json.dumps(x), report_items))
    payload = {
        "log_time_ns": time.time_ns(),
        "contents": str_report_items,
        "labels": {
            "tag": "workspace"
        }
    }

    requests.post(url=url, headers=headers, json=payload, verify=False)

    return {
        "body": ""
    }


def flatten_and_transform(response_list):
    result = {}

    for item in response_list:
        if not isinstance(item, dict) or "metrics" not in item:
            continue
        for metric in item["metrics"]:
            instance_id = metric["dimensions"][0]["value"]
            metric_name = metric["metric_name"]
            datapoints = metric["datapoints"]

            if instance_id not in result:
                result[instance_id] = {}

            result[instance_id][metric_name] = datapoints

    return result


def timestamp_to_beijing(ms_timestamp):
    try:
        timestamp_s = float(ms_timestamp) / 1000
        return datetime.fromtimestamp(
            timestamp_s,
            tz=timezone(timedelta(hours=8))
        ).strftime("%Y-%m-%d %H:%M:%S")
    except (TypeError, ValueError, OSError):
        return None


def handler(event, context):
    print("START START START")

    token = context.getToken()

    region = context.getUserData('region')
    ip = context.getUserData('ip')
    project_id = context.getUserData('project_id')
    log_group_id = context.getUserData('log_group_id')
    log_stream_id = context.getUserData('log_stream_id')

    # Name of the metric to be queried on the desktop
    metric_names = ["cpu_usage"]

    # Obtain the current UTC time.
    now = datetime.now(timezone.utc)
    current_hour_start = now.replace(minute=0, second=0, microsecond=0)
    previous_hour_start = current_hour_start - timedelta(hours=1)
    now_timestamp_ms = int(now.timestamp() * 1000)
    previous_hour_timestamp_ms = int(previous_hour_start.timestamp() * 1000)

    desktops_detail_list = list_desktops_detail(region, ip, token, project_id)

    instance_ids = [desktop["desktop_id"] for desktop in desktops_detail_list["desktops"]]

    metric_datas = list_metric_data(region, token, project_id, previous_hour_timestamp_ms, now_timestamp_ms,
                                    instance_ids, metric_names)

    report_data = []

    for desktop in desktops_detail_list["desktops"]:
        desktop_id = desktop.get("desktop_id", "")
        computer_name = desktop.get("computer_name", "")
        ipv4 = desktop.get("ipv4", "")
        user_name = desktop.get("user_name", "")
        flavor = desktop.get("product", {}).get("product_id", "") + ": " + desktop.get("product", {}).get(
            "descriptions", "")
        root_volume_size = desktop.get("root_volume", {}).get("size", "")
        data_volumes_size = [data_volume["size"] for data_volume in desktop.get("data_volumes", "")]
        enterprise_project_id = desktop.get("enterprise_project_id", "")
        cpu_usage = next((item.get("average") for item in metric_datas.get(desktop_id, {})
                         .get("cpu_usage", [])[-1:] if isinstance(item, dict)), None)
        mem_used_percent = next((item.get("average") for item in metric_datas.get(desktop_id, {})
                                .get("mem_usedPercent", [])[-1:] if isinstance(item, dict)), None)
        disk_used_percent = next((item.get("average") for item in metric_datas.get(desktop_id, {})
                                 .get("disk_usedPercent", [])[-1:] if isinstance(item, dict)), None)
        timestamp = timestamp_to_beijing(next((item.get("timestamp") for item in metric_datas.get(desktop_id, {})
                                              .get("cpu_usage", [])[-1:] if isinstance(item, dict)), None))
        report_data.append({
            "computer_name": computer_name,
            "desktop_id": desktop_id,
            "ipv4": ipv4,
            "user_name": user_name,
            "flavor": flavor,
            "root_volume_size": root_volume_size,
            "data_volumes_size": data_volumes_size,
            "enterprise_project_id": enterprise_project_id,
            "cpu_usage": cpu_usage,
            "mem_used_percent": mem_used_percent,
            "disk_used_percent": disk_used_percent,
            "timestamp": timestamp
        })

    report_lts(region, token, project_id, log_group_id, log_stream_id, report_data)

    return {
        "statusCode": 200,
        "isBase64Encoded": False,
        "body": report_data,
        "headers": {"Content-Type": "application/json"}
    }

Sample Code 3: Reporting Screen Recording Statistics

FunctionGraph invokes handler on a schedule to query screen recording statistics from the Workspace APIs and report them to LTS. Queried statistics include the username, user group, desktop name, desktop ID, screen recording record ID, event type, event ID, event level, event content, event occurrence time, and time (in seconds) relative to the video start time. The statistics are collected and reported by the hour.

Workspace API reference: Querying Screen Recording Records, Querying Key Desktop Events, and Querying Users
LTS API reference: API Access Overview

The sample code is as follows: (Modify the code based on the site requirements.)

# -*- coding:utf-8 -*-
import json
import time # New module (time) to record timestamps

import requests
from datetime import datetime, timezone, timedelta
from urllib.parse import urlencode

def get_workspace_headers(region_name, token):
    headers = {"Host": "workspace." + region_name + ".myhuaweicloud.com",
               "X-Auth-Token": token,
               "Content-Type": "application/json"}
    return headers

def fetch_users_in_batches(region_name, ip, token, project_id, user_name_list, batch_size=100):
    all_responses = []
    merged_responses = {
        "count": 0,
        "users": []
    }
    for i in range(0, len(user_name_list), batch_size):
        batch = user_name_list[i:i + batch_size]
        response = list_users(region_name, ip, token, project_id, batch)
        all_responses.append(response)

    for resp in all_responses:
        merged_responses["count"] += resp["total_count"]
        merged_responses["users"].extend(resp["users"])

    return merged_responses


def list_users(region_name, ip, token, project_id, user_name_list):
    base_url = "https://" + ip + "/v2/" + project_id + "/users"
    headers = get_workspace_headers(region_name, token)
    params = {}

    if user_name_list:
        params['user_names'] = ",".join(user_name_list)

    query_string = urlencode(params)
    if query_string:
        full_url = base_url + "?" + query_string
    else:
        full_url = base_url

    response = requests.request("GET", full_url, headers=headers, verify=False)

    return json.loads(response.text)


def list_screen_records(region_name, ip, token, project_id, start_time, end_time):
    base_url = "https://" + ip + "/v2/" + project_id + "/screen-records"

    headers = get_workspace_headers(region_name, token)

    params = {}
    limit = 1000
    params['limit'] = limit
    offset = 0
    params['offset'] = offset
    params['start_time'] = start_time
    params['end_time'] = end_time
    all_screen_records = []
    total_count = 0

    while True:
        params['offset'] = offset
        query_string = urlencode(params)
        full_url = f"{base_url}?{query_string}" if query_string else base_url
        response = requests.request("GET", full_url, headers=headers, verify=False)
        print("full_url", full_url)
        response_data = json.loads(response.text)
        print("response_data", response_data)
        screen_records = response_data.get("screen_records", [])
        current_total = response_data.get("total_count", len(screen_records))
        if total_count == 0:
            total_count = current_total
        all_screen_records.extend(screen_records)
        if len(all_screen_records) >= total_count:
            break
        offset += limit

    return {
        "total_count": total_count,
        "screen_records": all_screen_records
    }


def list_desktop_operations(region_name, ip, token, project_id, record_id):
    base_url = "https://" + ip + "/v2/" + project_id + "/screen-records/" + record_id + "/os-operations"

    headers = get_workspace_headers(region_name, token)

    params = {}
    limit = 1000
    params['limit'] = limit
    offset = 0
    params['offset'] = offset

    all_operations = []
    total_count = 0

    while True:
        params['offset'] = offset
        query_string = urlencode(params)
        full_url = f"{base_url}?{query_string}" if query_string else base_url
        response = requests.request("GET", full_url, headers=headers, verify=False)
        response_data = json.loads(response.text)
        operations = response_data.get("operations", [])
        current_total = response_data.get("total_count", len(operations))
        if total_count == 0:
            total_count = current_total
        all_operations.extend(operations)
        if len(all_operations) >= total_count:
            break
        offset += limit

    return {
        "screen_record_id": record_id,
        "total_count": total_count,
        "screen_records": all_operations
    }


def report_lts(region_name, token, project_id, log_group_id, log_stream_id, report_items):
    url = ("https://lts-access." + region_name + ".myhuaweicloud.com:8102/v2/" + project_id + "/lts/groups/"
           + log_group_id + "/streams/" + log_stream_id + "/tenant/contents")
    headers = {"X-Auth-Token": token,
               "Content-Type": "application/json"}
    print("url:", url)
    str_report_items = list(map(lambda x: json.dumps(x), report_items))
    payload = {
        "log_time_ns": time.time_ns(),
        "contents": str_report_items,
        "labels": {
            "tag": "workspace"
        }
    }
    response = requests.post(url=url, headers=headers, json=payload, verify=False)
    print ("report_lts_response content preview:," response.text)
    return {
        "body": ""
    }


def handler(event, context):
    print("START START START")

    token = context.getToken()

    region = context.getUserData('region')
    ip = context.getUserData('ip')
    project_id = context.getUserData('project_id')
    log_group_id = context.getUserData('log_group_id')
    log_stream_id = context.getUserData('log_stream_id')
    batch_size = int(context.getUserData('batch_size'))
    wait_time = int(context.getUserData('wait_time'))

    now = datetime.utcnow()

    start_of_previous_hour = now.replace(minute=0, second=0, microsecond=0) - timedelta(hours=1)
    end_of_previous_hour = start_of_previous_hour + timedelta(hours=1) - timedelta(microseconds=1)

    start_time = start_of_previous_hour.strftime("%Y-%m-%dT%H:%M:%SZ")
    end_time = end_of_previous_hour.strftime("%Y-%m-%dT%H:%M:%SZ")

    screen_records_list = list_screen_records(region, ip, token, project_id, start_time, end_time)

    screen_record_ids = [screen_record["id"] for screen_record in screen_records_list["screen_records"]]

    screen_records_dict = {}

    for screen_record in screen_records_list["screen_records"]:
        screen_records_dict[screen_record.get("id")] = screen_record

    unique_user_name_set = set()
    for item in screen_records_list["screen_records"]:
        unique_user_name_set.add(item["username"])

    user_detail_list = fetch_users_in_batches(region, ip, token, project_id, list(unique_user_name_set))

    user_group_names_dict = {
        user_detail["user_name"]: sorted(user_detail.get("group_names", []))
        for user_detail in user_detail_list["users"]
    }

    all_desktop_operations = []
    for i in range(0, len(screen_record_ids), batch_size):
        batch = screen_record_ids[i:i + batch_size]
        print(f"Processing batch {i // batch_size + 1}, processing {len(batch)} records...")

        for record_id in batch:
            operations = list_desktop_operations(region, ip, token, project_id, record_id)
            all_desktop_operations.extend(operations.get("screen_records", []))
        # If the current round is not the last one and a total of 50 records are processed in the current round, wait for 60 seconds to avoid being throttled.
        if (i + batch_size) < len(screen_record_ids):
            print(f"Reached rate limit. Waiting for {wait_time} seconds...")
            time.sleep(wait_time)
        else:
            print("All batches processed.")

    report_data = []
    print("len(all_desktop_operations)", len(all_desktop_operations))
    for desktop_operation in all_desktop_operations:
        username = desktop_operation.get("username", "")
        user_group_names = user_group_names_dict.get("username", [])
        desktop_id = desktop_operation.get("desktop_id", "")
        screen_record_id = desktop_operation.get("screen_record_id", "")
        event_type = desktop_operation.get("event_type", "")
        event_level = desktop_operation.get("event_level", "")
        event_id = desktop_operation.get("event_id", "")
        event_data = desktop_operation.get("event_data", "")
        operation_time = desktop_operation.get("operation_time", "")
        relative_start_time = desktop_operation.get("relative_start_time", "")
        desktop_name = screen_records_dict.get(screen_record_id, "").get("desktop_name", "")

        report_data.append({
            "username": username,
            "user_group_names": user_group_names,
            "desktop_name": desktop_name,
            "desktop_id": desktop_id,
            "screen_record_id": screen_record_id,
            "event_type": event_type,
            "event_level": event_level,
            "event_id": event_id,
            "event_data": event_data,
            "operation_time": operation_time,
            "relative_start_time": relative_start_time
        })

    report_lts(region, token, project_id, log_group_id, log_stream_id, report_data)

    return {
        "statusCode": 200,
        "isBase64Encoded": False,
        "body": len(report_data),
        "headers": {"Content-Type": "application/json"}
    }

Step 3: Configuring LTS Logs

Creating a Log Group and Stream

Perform the operations in Creating a Log Group and Creating a Log Stream.
Perform the operations in Obtaining Log Group and Log Stream IDs.

Configuring Cloud Structuring Parsing and Setting Indexes

After configuring cloud structuring parsing, set indexes and enable auto configuration. Once the indexes are configured, review the fields and report the indexes again to ensure the changes are applied.

In the log management area, click the name of the created log group. The log stream page is displayed.
In the navigation pane, click Log Settings. On the Cloud Structuring Parsing page, select JSON.
Click Select from Existing Logs, select the required raw log events, and click OK.

Figure 14 Setting of cloud structuring parsing
Click Intelligent Extract and then Save.
Click the Index Settings tab and click Auto Configure.

Figure 15 Index settings
Click OK.

Configuring log dashboard

After logs are collected and structured, create a dashboard and add a chart for the dashboard on the LTS console. You can then run SQL statements in the visualization widget and interactive search to view the results. For details, see Creating a Dashboard.

When creating a dashboard, you can select Add Chart under Add Chart.