# 配置接入LTS统计云桌面监控指标 #### 操作场景 Workspace支持提供多个维度的桌面云统计报表，可以按照用户、用户组、终端类型、时间等多个维度进行统计汇总，以实现对桌面云运行使用状态的全局把控。 #### 前提条件 已参考VPC终端节点服务[购买终端节点](https://support.huaweicloud.com/usermanual-vpcep/zh-cn_topic_0131645189.html)。 ![](https://support.huaweicloud.com/bestpractice-workspace/public_sys-resources/note_3.0-zh-cn.png) - 函数访问VPC内资源时，需额外购买VPC终端节点，并为函数配置具备VPC管理权限的委托。需购买以下三个服务的终端节点： - com.myhuaweicloud.cn-south-1.ces - com.myhuaweicloud.cn-south-1.lts-access - com.myhuaweicloud.cn-south-1.workspace - 购买VPC终端节点时，请确保三者选择的VPC完全一致。 - 创建函数时，若网络配置为公网访问，则无需创建 VPC 终端节点。 #### 步骤一：创建函数及配置 请参见[创建事件函数](https://support.huaweicloud.com/usermanual-functiongraph/functiongraph_01_1441.html)章节，完成函数的创建。 ![](https://support.huaweicloud.com/bestpractice-workspace/public_sys-resources/note_3.0-zh-cn.png) 在创建函数的配置步骤中，请将参数"运行时"环境设置为"Python 3.9"。 **函数的常规设置** 根据业务逻辑、数据量及执行频率动态调整超时时间，常规业务场景建议设置为60秒。 1. 在函数列表单击创建的函数名称，进入函数基本信息页面。 2. 单击"设置"页签，将常规设置下的"执行超时时间"参数设置为"60"秒。 图1执行超时时间设置 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002511311090.png "点击放大") 3. 单击"保存"。 **函数的触发器设置** 根据业务创建触发器，触发规则建议使用corn表达式指定，例如：录屏记录上报设置为每小时开始执行一次，桌面记录由于需要用到CES Agent上报信息, 建议在每小时的第四十分钟上报上一小时的数据，用户记录上报可在每天凌晨执行一次。 4. 在函数基本信息页面单击"设置 \> 触发器 \> 创建触发器"。 图2创建定时触发器 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002542817989.png "点击放大") 表1触发器参数说明 | 参数 | 说明 | 示例 | |:---|:---|:---| | 触发器类型 | 选择"定时触发器 (TIMER)"。 | 定时触发器 (TIMER) | | 定时器名称 | 自定义的定时器名称。支持字母、数字、下划线和中划线，必须以字母开头，且长度不能超过64个字符。 | Timer-Workspace | | 触发规则 | 触发器的触发规则，可选择使用"固定频率"或"Cron表达式"。 - 固定频率：固定时间间隔触发函数，该类型下支持配置单位为分、时、天，每种类型仅支持整数配置，其中分钟支持范围(0，60\]，小时支持范围(0，24\]，天支持范围(0，30\]。 - Cron表达式：设置更为复杂的函数执行计划，例如：周一到周五上午08:30:00执行函数等，具体请参见[Cron表达式规则](https://support.huaweicloud.com/usermanual-functiongraph/functiongraph_01_0207.html#functiongraph_01_0207__section1237972211317)。 | Cron表达式 40 \* \* \* \* ? | | 是否开启 | 选择是否开启定时触发器。不开启则不会按照设定频率触发函数。 | 开启 | | 附加信息 | 可选参数：如果用户配置了触发事件，会将该事件填写到TIMER事件源的"user_event"字段，详情请参见[支持的事件源](https://support.huaweicloud.com/devg-functiongraph/functiongraph_02_0102.html)。 | - | 5. 单击"确定"。 **函数的权限配置** 6. 在函数基本信息页面单击"设置 \> 权限 "。 7. 在委托右侧单击"创建委托"。 图3创建委托 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002542824731.png "点击放大") 8. 在统一身份认证服务的左侧导航窗格中，选择"委托"页签，单击右上方的"创建委托"。 图4创建委托 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002511185890.png "点击放大") 9. 配置委托信息。 - 委托名称：自定义委托名称。 - 委托类型：选择"云服务"。 - 云服务：选择"函数工作流 FunctionGraph"。 - 持续时间：默认为"永久" 图5委托配置 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002511347192.png "点击放大") 10. 单击"完成"。 11. 在授权的确认弹窗中，单击"立即授权"。 12. 在授权页面单击"新建策略"。 图6新建策略 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002511189324.png "点击放大") 13. 自定义策略名称，策略配置方式选择"JSON视图"。 图7自定义策略配置 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002511194628.png "点击放大") 14. 将如下内容拷贝至"策略内容"中。 ``` { "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" ] } ] } ``` 15. 单击"下一步"，创建完成。 16. 勾选新创建的委托，单击"下一步"。 17. 默认"选择授权范围方案"，单击"确定"。 18. 在函数基本信息页面单击"设置 \> 权限 "。 19. 在委托下拉框选择创建的委托。 图8权限设置 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002511357652.png "点击放大") 20. 单击"保存"。 **函数的网络配置** 网络配置可选择函数公网访问或者函数访问VPC内资源。函数使用默认网卡访问公网时，访问带宽为用户间共享，仅适合小量调用的业务场景使用；对带宽、性能、可靠性有高要求的业务场景，建议开启"函数访问VPC内资源"功能。 21. 在函数基本信息页面单击"设置 \> 网络配置"。 22. 在"网络配置"页面，开启允许函数访问VPC内资源并配置。 图9开启函数访问VPC资源 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002511378782.png "点击放大") - VPC：下拉选择[前提条件]中创建VPCEP终端节。 - 子网：默认为VPC下的子网。 23. 单击"保存"。 **函数的环境变量配置** 24. 在函数基本信息页面单击"设置 \> 环境变量 "。 25. 依次单击"编辑环境变量 \> 使用JSON格式编辑"。 图10环境变量配置 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002542862997.png "点击放大") 26. 配置环境变量，按照以下模板，根据说明进行编辑后拷贝至"使用JSON格式编辑"中。 以下参数取值需要进行替换： ``` { "region": "cn-south-1", "ip": "100.xxx.xxx.xxx", "project_id": "xxx", "log_group_id": "xxx", "log_stream_id": "xxx" } ``` - region：请根据当前局点进行填写，例如：华南-广州为"cn-south-1"。 - ip：通过解析Workspace的访问域名获取。 1. 按"win+r"，在弹出的运行框中输入"cmd"，按"Enter" 2. 执行如下命令获取ip。 **ping workspace.{*region_id*}.myhuaweicloud.com** ![](https://support.huaweicloud.com/bestpractice-workspace/public_sys-resources/note_3.0-zh-cn.png) ***region_id***请根据实际局点进行替换。 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002543867007.png "点击放大") - project_id：请参见[获取项目ID](https://support.huaweicloud.com/api-workspace/workspace_09_0068.html)。 - log_group_id：日志组ID，参见[2]获取。 - log_stream_id：日志流ID，参见[2]获取。 **代码执行** 27. 在函数基本信息页面单击"代码"页签。 28. 依次单击"文件 \> 新建文件"。 图11新建文件 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002511913836.png "点击放大") 29. 将新建的文件重命名为"index.py"。 图12重命名文件 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002543674925.png "点击放大") 30. 根据需要对所需代码进行测试，将[步骤三：部署代码]步骤中的代码示例拷贝至"index.py"中。 ![](https://support.huaweicloud.com/bestpractice-workspace/public_sys-resources/note_3.0-zh-cn.png) 代码请根据实际情况进行修改。 31. 单击"测试"。 图13测试代码 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002543597963.png "点击放大") 32. 请参见[3]中配置LTS日志。 #### 步骤三：部署代码 通过以下代码示例，使用FG定时查询Workspace的接口，以查询用户连接统计、桌面性能统计、录屏操作统计等，并将数据上报至LTS。您可以在LTS服务页面查看这些数据。 #### 示例代码一：上报用户连接统计信息 handler为主函数入口，通过FG定时查询Workspace的接口，查询用户连接统计（用户名、用户描述、归属用户组、用户过期时间、连接终端类型、企业项目、连接次数、连接成功次数、连接失败次数、使用时长、平均时延），上报数据到LTS，按自然天维度统计。 1. 涉及workspace接口文档：[查询用户列表](https://support.huaweicloud.com/api-workspace/ListUsers.html)、[查询用户使用统计数据](https://support.huaweicloud.com/api-workspace/ListUserUsageMetric.html)、[查询用户详情信息](https://support.huaweicloud.com/api-workspace/ListUserDetail.html)、[查询登录信息](https://support.huaweicloud.com/api-workspace/ListLoginRecordsNew.html)。 2. 涉及LTS接口文档：[API接入概述](https://support.huaweicloud.com/usermanual-lts/lts_04_1032.html) 3. 代码示例如下所示，请根据实际情况进行修改： ``` # -*- 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 内容预览:", response.text) items = response_data.get("items", []) current_total = response_data.get("count", len(items)) # fallback: 使用长度 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() # 构造完整 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 内容预览:", 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) # 时间格式已标准化，去除微秒，统一表示为 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"] } # 遍历 metric 列表，把每个指标展开成 key-value 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"] = "永不过期" 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"} } ``` #### 示例代码二：上报桌面性能统计信息 handler为主函数入口，通过FG定时查询Workspace的接口，查询桌面性能统计（桌面名称、桌面ID、桌面IP、绑定的用户、规格、系统盘大小（GB）、数据盘大小（GB）、企业项目、CPU使用率、 内存使用率、 磁盘使用率、监控数据事件），上报数据到LTS，按每小时统计上报一次。 1. 涉及workspace接口文档：[查询桌面详情列表](https://support.huaweicloud.com/api-workspace/ListDesktopsDetail.html) 2. 涉及CES接口文档：[批量查询监控数据](https://support.huaweicloud.com/api-ces/ces_03_0034.html) 3. 涉及LTS接口文档：[API接入概述](https://support.huaweicloud.com/usermanual-lts/lts_04_1032.html)。 4. 代码示例如下所示，请根据实际情况进行修改： ``` # -*- 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') # 桌面需要查询的指标名 metric_names = ["cpu_usage"] # 获取当前 UTC 时间 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"} } ``` #### 示例代码三：上报录屏操作统计信息 handler为主函数入口，通过FG定时查询Workspace的接口，查询录屏操作统计（用户名、归属用户组、桌面名、桌面ID、录屏记录ID、事件类型、事件ID、事件级别、事件内容、事件发生时间、相对于视频开始的时间（秒）），上报数据到LTS，按每小时统计上报一次。 1. 涉及workspace接口文档：[查询录屏记录列表](https://support.huaweicloud.com/api-workspace/ListScreenRecords.html)、[查询桌面关键事件列表](https://support.huaweicloud.com/api-workspace/ListDesktopOperations.html)、[查询用户列表](https://support.huaweicloud.com/api-workspace/ListUsers.html)。 2. 涉涉及LTS接口文档：[API接入概述](https://support.huaweicloud.com/usermanual-lts/lts_04_1032.html)。 3. 代码示例如下所示，请根据实际情况进行修改： ``` # -*- coding:utf-8 -*- import json import time # 新增模块：time，用于记录时间戳 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 内容预览:", 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", [])) # 如果不是最后一轮，并且本轮处理了完整的 50 条，则等待 60 秒避免流控 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"} } ``` #### 步骤三：LTS日志相关配置 **创建日志组及日志流** 1. 请参见[创建日志组](https://support.huaweicloud.com/usermanual-lts/lts_04_0003.html#section1)和[创建日志流](https://support.huaweicloud.com/usermanual-lts/lts_04_0004.html#section1)》 2. 获取日志组和日志流的ID，请参见[获取日志组ID和日志流ID](https://support.huaweicloud.com/api-lts/lts_api_0014.html)。 **日志的云端化结构解析及索引配置** 配置完成云端化结构解析以后，需进行索引配置，点击自动配置，索引配置完成后进行字段检查。索引配置后，需要重新上报生效。 3. 在日志管理下单击创建的日志组名称，进入日志流页面。 4. 在页面左侧依次单击"日志设置 \> 云端结构化解析 \> JSON"。 5. 单击"已有日志中选择"，勾选所需原始日志，单击"确定"。 图14"云端结构化解析"设置 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002511877040.png "点击放大") 6. 单击"智能提取"，单击"保存"。 7. 单击"索引配置"页签，单击"自动配置"。 图15"索引配置"设置 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002512037036.png "点击放大") 8. 单击"确定"。 **日志的仪表盘配置** 已成功采集日志，且已对日志内容完成结构化配置，在LTS服务页面选择添加仪表盘，添加仪表盘以后进行图标添加，在可视化对象和交互式搜索中填入对应sql查看效果。请参见[创建日志仪表盘](https://support.huaweicloud.com/usermanual-lts/lts_05_0020.html)。 ![](https://support.huaweicloud.com/bestpractice-workspace/public_sys-resources/note_3.0-zh-cn.png) 在创建仪表盘时，如需添加图表，请在"添加图表"步骤中选择"添加可视化图表"。 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002560879103.png "点击放大") 9. 在左侧导航栏中选择"仪表盘"。 10. 在创建的仪表盘下单击仪表盘名称查看日志。 图16查看日志 ![](https://support.huaweicloud.com/bestpractice-workspace/zh-cn_image_0000002543681505.png "点击放大")