基于操作日志实现业务资源事件关联分析

方案概述

本方案以多账号操作日志统一归集与审计及资源标签为基础，通过搭建日志处理服务，为操作日志新增具有业务语义的资源标签，向企业提供业务视角的审计日志查询方案。

操作审计支持将日志归集到日志服务SLS，部署日志处理服务对日志消息进行监听消费，对数据清洗过滤后实时处理成用户关注的内容后再次写入SLS，支持生成用户关注的业务审计视图。日志处理过程主要为过滤掉用户不关心的操作事件、查询资源标签，从用户视角重新构建新的操作日志写入SLS，通过SLS的查询分析能力支持按标签查询用户关注的操作记录。

方案优势

提升审计效率

基于已归集的审计日志和资源标签，产出业务项目视角的审计视图，提升客户的操作日志审计效率和分析体验。

客户场景

从业务视角对操作日志进行审计

场景描述

企业在内部有多个业务单元或项目，存在资源被多个账号或业务单元共享的场景，多个业务单元或项目用户会对资源有变更操作。现有操作审计日志中能看到的只是事件类型、资源、账号维度的操作记录，不具备业务属性，在进行操作审计日志统计分析的时候缺乏业务视角，无法快速统计用户操作的资源属于什么业务单元、某个项目进行了哪些变更，只能线下手动进行关联，审计操作繁琐且成本高昂。

适用客户

完成操作审计配置，从业务视角对操作审计日志有统计分析诉求的企业客户。

客户案例

客户背景

某国内公司X为互联网电动汽车知名品牌，专注于针对一线城市年轻人的互联网时尚电动汽车的研发。

客户痛点

在一个UID下面，有多个RAM用户，这些用户对资源有变更操作，需要按项目的角度来统计哪些不同项目分别被哪些RAM User操作过，现有操作审计日志无法快速按项目统计。

实施方案

客户收益

基于已归集的审计日志和资源标签，产出业务项目视角的审计视图，有效提升用户的操作日志审计效率和审计体验。

方案架构

关键流程：

• 在企业管理主账号已完成操作审计日志归集到SLS。

• 在相应账号下对关注的资源打标签，使资源具备业务属性。

• 在审计账号SLS创建新LogStore用于业务视角的日志存储，创建RAM用户及AccessKey用于供日志处理服务监听SLS消息、写入新LogStore。

• 在企业管理账号创建只读权限的RAM用户及AccessKey，用于日志处理服务读取所有账号的资源标签。

• 在ECS服务器部署启动Python日志处理服务。

• 日志处理服务对SLS日志消息进行过滤及操作事件处理、获取资源标签，将用户关注字段写入新LogStore。

• 在日志审计账号对新LogStore构建业务查询视图。

产品费用及名词

产品费用

名词解释

安全性

资源目录权限

资源目录服务关联角色（AliyunServiceRoleForResourceDirectory）为资源目录集成服务提供可信访问通道，详情参见资源目录服务关联角色。

操作审计

操作审计管理权限相关，如RAM用户授权、服务关联角色、系统权限策略等详情参见管理权限。

AccessKey使用

• 需要使用企业管理主账号下RAM用户的AccessKey，该RAM用户需要拥有资源只读权限 ReadOnlyAccess和角色扮演AliyunSTSAssumeRoleAccess两个权限，用于日志处理服务读取资源目录账号下的资源标签。

• 需要使用日志审计账号下RAM用户的AccessKey，该RAM用户需要拥有SLS的读取及写入权限，用于监听日志消息及写入处理后的日志。

注意事项

操作审计

• 操作审计相关限制详情参见使用限制，地域支持详情参见支持操作审计的地域，云产品支持情况详情参见支持操作审计的云服务及事件。

• 企业需要为用于存储操作事件的SLS付费，根据自身需求设置新写入的日志保留天数。

日志处理服务可用性

方案中部署的日志处理Python服务若需要高可用，需要根据实际情况进行配置消费组信息、进行多套部署并进行异常处理加固。SLS SDK概述及异常处理等详情参见SDK参考。

标签使用限制

云产品对标签支持情况详情参见支持标签的云服务。

资源目录使用限制

资源目录及资源共享使用限制详情参见使用限制。

实施步骤

实施步骤中以ECS、SLB打标签及其操作事件为例进行部署验证。

实施准备

已完成审计日志归集

• 按照多账号操作日志统一归集与审计完成审计日志配置。

• 企业管理账号需确保已开启操作审计服务。

拥有运行Python环境的ECS及SLB资源

• ECS用于部署日志处理服务，具备Python3运行环境。

• 已有ECS、SLB，用于打标签并进行管控操作，进行日志写入验证。

确保资源结构已创建，本方案中相应的账号UID及用途如下表

实施时长

在实施准备工作完成的情况下，本方案实施预计时长：45分钟。

操作步骤

为资源添打标签

1. 登录成员账号ECS控制台，为ECS打标签。如果资源已有标签可跳过此步骤。

   

   

   

2. 登录成员账号负载均衡控制台，为SLB打标签。如果资源已有标签可跳过此步骤。

   

创建日志审计目标LogStore

1. 使用RAM用户登录管理账号资源目录控制台，选择日志审计账号登入。

2. 登入日志审计账号后进入日志服务控制台，能够查看到已归集的审计日志记录。

3. 在日志服务控制台新建Project并记录名称备用，该Project用于存储处理后的日志，也可不新建使用已有Project。

4. 在新建的Project下新增LogStore，作为存储处理后的日志的LogStore，记录LogStore名称备用。

新建RAM用户及AccessKey

1. 登录日志审计账号RAM控制台，新建RAM用户后复制保存AccessKey备用，该AccessKey用于日志处理服务的SLS消息消费以及处理后的日志写入。

2. 为日志审计账号新建的RAM用户授予日志服务管控权限。

3. 切换到管理账号RAM控制台，新建RAM用户后保存AccessKey备用，用于日志处理服务读取所有账号的ECS/SLB等资源信息。

4. 按最小授权原则，为管理账号新建的RAM用户授予只读和STS角色扮演权限，用于日志处理服务读取管理账号的资源信息、扮演资源目录角色读取成员账号的ECS/SLB等资源标签信息。日志处理服务读取成员账号的资源信息时，会使用该管理账号的RAM用户，扮演为成员账号的资源目录角色ResourceDirectoryAccountAccessRole进行资源数据读取，该角色在成员账号纳入管理账号资源目录时已经自动创建。

部署日志处理服务

1. 确保要部署日志处理服务的服务器已安装Python环境，在服务器上安装阿里云相关Python SDK，本方案使用Python3。

   # 安装阿里云 SDK 核心库
   pip3 install aliyun-python-sdk-core
   
   # 安装阿里云 STS SDK
   pip3 install aliyun-python-sdk-sts
   
   # 安装阿里云 日志服务SLS SDK 
   pip3 install aliyun-log-python-sdk
   
   # 安装阿里云 ECS SDK 
   pip3 install aliyun-python-sdk-ecs
   
   # 安装阿里云 SLB SDK 
   pip3 install aliyun-python-sdk-slb

2. 修改下面代码配置，填入管理账号UID、源LogStore、目标LogStore、AccessKey等配置信息，配置完成后运行Python脚本启动日志处理服务，服务会持续消费SLS源日志库，当操作事件类型为“写入”时会进行处理并写入目标LogStore。代码示例中收到ECS及SLB的操作日志时会去查询资源标签，每个标签都会向目标LogStore写入单独字段，需要查询其它资源类型的标签时引入相应sdk依赖，在query_resource_tags函数中添加类型支持即可。

   # encoding: utf-8
   
   import time
   import json
   
   from aliyun.log.consumer import *
   from aliyun.log import *
   
   from aliyunsdkcore.client import AcsClient
   from aliyunsdkcore.auth.credentials import AccessKeyCredential
   from aliyunsdkcore.auth.credentials import StsTokenCredential
   
   from aliyunsdkecs.request.v20140526.ListTagResourcesRequest import ListTagResourcesRequest as EcsListTagResourcesRequest
   from aliyunsdkslb.request.v20140515.ListTagResourcesRequest import ListTagResourcesRequest as SlbListTagResourcesRequest
   from aliyunsdksts.request.v20150401.AssumeRoleRequest import AssumeRoleRequest
   
   # SLS消息消费组
   sls_consumer_group_name = 'g2'
   sls_consumer_name = 'c2'
   
   # SLS endpoint
   sls_endpoint = 'cn-hangzhou.log.aliyuncs.com'
   
   # SLS 源project
   sls_source_project = 'testl1-trail-project'
   sls_source_logstore = 'actiontrail_test212'
   
   # SLS 目的project
   sls_target_project = 'test1-trail-project'
   sls_target_logstore = 'processed_test1'
   
   # SLS AccessKey
   sls_access_key = ''
   sls_secret_key = ''
   
   # 管理账号UID
   admin_account_id = '1140931609457592'
   
   # 查询资源(如ecs/slb/vpc)标签所需AK，使用管理账号RAM用户的AK，RAM用户需具备所有资源读取和STS角色扮演权限
   resource_access_key = ''
   resource_secret_key = ''
   
   # 需要处理的event的serviceName，如ecs/slb/vpc/oss/cloudsso/sts，为空表示处理所有
   need_process_service = []
   # need_process_service = ['ecs', 'slb', 'vpc', 'oss']
   
   # 是否只处理写事件，建议开启True
   only_process_write_event = True
   
   
   class SampleConsumer(ConsumerProcessorBase):
       shard_id = -1
       last_check_time = 0
   
       def initialize(self, shard):
           self.shard_id = shard
   
       def process(self, log_groups, check_point_tracker):
           for log_group in log_groups.LogGroups:
               items = []
               for log in log_group.Logs:
                   item = dict()
                   item['time'] = log.Time
                   for content in log.Contents:
                       item[content.Key] = content.Value
                   items.append(item)
   
               log_items = dict()
               log_items['topic'] = log_group.Topic
               log_items['source'] = log_group.Source
               log_items['logs'] = items
   
               for item in log_items['logs']:
   
                   """
                   item格式
                   {'time': 1632708969, 'event': '{"acsRegion":"cn-beijing","serviceName":"Ecs","xxx":"xxx"}'}
                   """
                   if 'event' not in item:
                       continue
   
                   log_item_event = parse_json(item['event'])
                   # 只处理写事件
                   if only_process_write_event and log_item_event['eventRW'] != 'Write':
                       continue
   
                   # 只处理关心的service event
                   if need_process_service and str(log_item_event['serviceName']).lower() not in need_process_service:
                       continue
   
                   print('\noriginEvent:' + str(item))
   
                   try:
                       # 构建新事件，只包含所关心的字段
                       new_log_item_event = build_new_event(log_item_event)
                       print('\nnewEvent:' + str(new_log_item_event))
   
                       # 查询资源标签
                       res_tags = query_resource_tags(new_log_item_event['accountId'],
                                                      new_log_item_event['serviceName'],
                                                      new_log_item_event['acsRegion'],
                                                      new_log_item_event['resourceName'])
                       print('\nresTags:' + str(res_tags))
   
                       # 构建写入sls的tuple
                       log_tuples = build_log_item_tuples(new_log_item_event, res_tags)
                       print('\nlogTuples:' + str(log_tuples))
   
                       # 写入sls
                       put_logs(log_tuples)
                   except Exception:
                       import traceback
                       traceback.print_exc()
                       return check_point_tracker.get_check_point()
   
           current_time = time.time()
           if current_time - self.last_check_time > 3:
               try:
                   self.last_check_time = current_time
                   check_point_tracker.save_check_point(True)
               except Exception:
                   import traceback
                   traceback.print_exc()
           else:
               try:
                   check_point_tracker.save_check_point(False)
               except Exception:
                   import traceback
                   traceback.print_exc()
   
           # None means succesful process
           # if need to roll-back to previous checkpoint，return check_point_tracker.get_check_point()
           return None
   
       def shutdown(self, check_point_tracker):
           try:
               check_point_tracker.save_check_point(True)
           except Exception:
               import traceback
               traceback.print_exc()
   
   
   def build_new_event(log_item_event):
       if 'resourceType' not in log_item_event or 'resourceName' not in log_item_event:
           log_item_event['resourceType'] = ''
           log_item_event['resourceName'] = ''
   
       if 'principalId' not in log_item_event['userIdentity']:
           log_item_event['userIdentity']['principalId'] = ''
   
       new_log_item_event = {
           'eventId': log_item_event['eventId'],
           'eventName': log_item_event['eventName'],
           'eventRW': log_item_event['eventRW'],
           'eventSource': log_item_event['eventSource'],
           'eventTime': log_item_event['eventTime'],
           'eventType': log_item_event['eventType'],
   
           'acsRegion': log_item_event['acsRegion'],
           'serviceName': log_item_event['serviceName'],
           'resourceType': log_item_event['resourceType'],
           'resourceName': log_item_event['resourceName'],
   
           'accountId': log_item_event['userIdentity']['accountId'],
           'principalId': log_item_event['userIdentity']['principalId'],
           'userName': log_item_event['userIdentity']['userName'],
           'userType': log_item_event['userIdentity']['type'],
       }
   
       return new_log_item_event
   
   
   def assume_res_dir_role(account_id):
   
       request = AssumeRoleRequest()
       request.set_accept_format('json')
       request.set_RoleSessionName("ResourceDirectoryAccountAccessRoleSession")
       request.set_RoleArn("acs:ram::" + str(account_id) + ":role/ResourceDirectoryAccountAccessRole")
   
       client = AcsClient(resource_access_key, resource_secret_key)
       response = client.do_action_with_exception(request)
   
       print(str(response, encoding='utf-8'))
   
       return parse_json(str(response, encoding='utf-8'))
   
   
   def query_resource_tags(account_id, resource_type, region_id, resource_id):
       # 若需新增其它资源标签查询在此添加代码
       if 'ecs' in str(resource_type).lower():
           request = EcsListTagResourcesRequest()
       elif 'slb' in str(resource_type).lower():
           request = SlbListTagResourcesRequest()
       else:
           return []
   
       request.set_accept_format('json')
       request.set_ResourceType("instance")
       request.set_ResourceIds([resource_id])
   
       if account_id == admin_account_id:
           # 若是管理账号下资源，直接使用自己的ak
           resource_access_credentials = AccessKeyCredential(resource_access_key, resource_secret_key)
       else:
           # 成员账号，管理账号扮演资源目录角色获取STS令牌
           rd_role = assume_res_dir_role(account_id)
           resource_access_credentials = StsTokenCredential(rd_role['Credentials']['AccessKeyId'],
                                                            rd_role['Credentials']['AccessKeySecret'],
                                                            rd_role['Credentials']['SecurityToken'])
   
       client = AcsClient(region_id=region_id, credential=resource_access_credentials)
   
       response = client.do_action_with_exception(request)
       print(region_id + ' originTagResponse:' + str(response, encoding='utf-8'))
   
       response_json = parse_json(str(response, encoding='utf-8'))
   
       if response_json is not None and 'TagResources' in response_json and 'TagResource' in response_json['TagResources']:
           return response_json['TagResources']['TagResource']
       else:
           return []
   
   
   def parse_json(content):
       try:
           return json.loads(content)
       except Exception as e:
           return None
   
   
   def build_log_item_tuples(log_dict, tags):
       log_tuples = []
       for k, v in log_dict.items():
           log_tuple = (k, v)
           log_tuples.append(log_tuple)
   
       for tag in tags:
           log_tuple = (tag['TagKey'], tag['TagValue'])
           log_tuples.append(log_tuple)
   
       return log_tuples
   
   
   def put_logs(log_tuples):
       print("ready to put logs for %s" % sls_target_logstore)
   
       log_item = LogItem()
       log_item.set_contents(log_tuples)
       log_group = [log_item]
   
       request = PutLogsRequest(sls_target_project, sls_target_logstore, "", "", log_group, compress=False)
   
       sls_log_client = LogClient(sls_endpoint, sls_access_key, sls_secret_key)
       sls_log_client.put_logs(request)
   
       print("put logs for %s success " % sls_target_logstore)
   
   
   def main():
       client_worker = None
       try:
           option = LogHubConfig(sls_endpoint, sls_access_key, sls_secret_key, sls_source_project, sls_source_logstore,
                                 sls_consumer_group_name, sls_consumer_name,
                                 cursor_position=CursorPosition.END_CURSOR,
                                 heartbeat_interval=6,
                                 data_fetch_interval=1)
   
           client_worker = ConsumerWorker(SampleConsumer, consumer_option=option)
           print('********start consume********')
           client_worker.start()
           client_worker.join()
       except Exception as ex:
           print(ex)
           print('********end consume********')
           client_worker.shutdown()
   
   
   if __name__ == '__main__':
       main()

目标日志库添加索引及验证

1. 登录日志审计账号日志服务控制台，可查看目标日志库写入情况。

2. 建立字段索引，用于支持单字段查询分析。如果是标签字段，点击自动生成索引后可能无法自动出现在列表内，需要点击“+”号手动添加标签Key到列表中，例如company、role标签。

3. 按需重建索引。建立索引后只对之后的日志写入生效，如果需要按字段查询历史数据需要进行索引重建。

4. 验证资源变更日志数据写入。对SLB、ECS进行变更类操作后等待片刻，可按资源类型及标签等字段查询处理后的日志。

5. 按业务标签查询，快速构建统计视图。

故障排除

SLS

日志服务问题详情参见常见问题，SDK异常处理详情参见错误处理。

操作审计

操作审计常见问题详情参见常见问题。

资源目录

资源目录常见问题处理详情参见常见问题。

相关内容

• SLS SDK参考