为 Amazon DMS 数据库迁移任务建立自动化监控机制

Amazon Database Migration Service(DMS)是一项云服务，可轻松完成对关系数据库、数据仓库、NoSQL 数据库以及其他类型数据存储的迁移工作。Amazon DMS 专门用于从多个本地实例，或者云与本地实例的组合将数据迁移至亚马逊云。

• Database Migration Service

http://aws.amazon.com/dms

在使用 Amazon DMS 进行数据迁移的过程中，最重要的是对当前正在进行的复制任务的状态执行监控。您可以通过任务的控制表以及 Amazon CloudWatch 服务进行这项操作。您可以通过 Amazon 管理控制台、Amazon 命令行界面（Amazon CLI）或者 Amazon DMS API 监控任务进度、实际使用的资源与网络连接。

您通常使用多个任务执行迁移。这些任务之间彼此独立，能够同时运行，而复制任务的数量则根据实际场景而有所区别。当面对大量同时进行的复制任务时，靠人工去监控每项任务的进度无疑是一项既枯燥、又容易出错的工作。

📢 想要了解更多亚马逊云科技最新技术发布和实践创新，敬请关注在上海、北京、深圳三地举办的 2021 亚马逊云科技中国峰会！点击图片报名吧～

在本文中，我们为您提供一套使用 Amazon CloudFormation 模板的自动化解决方案。此解决方案包含以下步骤：

1. 为复制任务创建一个 CloudWatch 告警。

2. 创建 Amazon DMS 事件订阅。

3. 配置 Amazon Simple Notification Service (Amazon SNS)，将任务的 CloudWatch 日志中发现的错误通知给您。

4. 创建一个 Amazon Lambda 函数，发送 SNS 通知以重现 CloudWatch 告警。

• Amazon Simple Notification Service 

http://aws.amazon.com/sns

前提条件

在开始之前，您必须拥有以下资源：

• Amazon DMS 源与目标端点

• 一个 Amazon DMS 复制实例

• 启用日志记录的 Amazon DMS 复制任务（关于详细说明，请参阅使用 Amzon DMS 创建持续复制任务）

• 一个 SNS 主题

• 使用 Amzon DMS 创建持续复制任务

https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Task.CDC.html

在具备以上先决条件之后，大家即可开始自动执行复制任务监控。

用于 Amazon DMS 复制任务的 CloudWatch 告警

对任务状态进行监控的首选方法，就是为复制任务创建 CloudWatch 告警。因为只要复制任务的指标发生改变，告警就会被立即触发。

我们建议您为以下 DMS 指标设置告警：

• CDCLatencySource

• CDCLatencyTarget

• CDCChangesDiskSource

• CDCChangesDiskTarget

关于 Amazon DMS 指标的更多详细信息，请参阅 Amazon Database Migration Service 指标。

• Amazon Database Migration Service 指标

https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Monitoring.html#CHAP_Monitoring.Metrics

CDCLatencySource

CDCLatencySource 是指从源终端节点中捕获的最后一个事件与 Amazon DMS 实例的当前系统时间戳之间的间隔 (秒)。如果因为任务范围限制原因而没有从源终端节点处捕捉到任何变更，则 Amazon DMS 将此值设置为零。

在复制期间，Amazon DMS 会从源数据库事务日志中读取变更。

根据源数据库的实际引擎，源事务日志可能包含未提交的数据。在复制进行期间，Amazon DMS 会从事务日志中读取传入的变更，但仅将已经提交的变更转发至目标，这最终将引发源数据库发送延迟。

CDCLatencyTarget

CDCLatencyTarget 是指在目标上等待提交的第一个事件时间戳与 Amazon DMS 实例的当前系统时间戳之间的间隔（秒）。如果存在目标没有处理的事务，则会产生此值。如果所有事务都得到及时处理，则目标延迟将与源延迟相同。目标延迟不应低于源延迟。

目标延迟之所以高于源延迟，是因为前者代表的是从源数据库中的插入时间、直到对应行发生提交为止的当前记录的总延迟时长。

CDCChangesDiskSource

CDCChangesDiskSource 是指在磁盘上累积并等待从源处进行实际提交的总行数。

CDCChangesDiskSource 当中的所有行都曾经处于内存内，只是由于达到了允许驻留在内存内的最长时间阈值而被逐出保存在磁盘上了。我们的目标是了解引擎的内部结构，并使用任务设置将 CDCChangesDiskSource 的值最小化。例如我们可以尝试设置 MemoryLimitTotal 与 MemoryKeepTime 的值来达到目的。如果需要了解更多详细信息，请参阅调试 Amazon DMS 迁移：出现问题时应采取的各项措施（第二部分）。

• 调试 Amazon DMS 迁移：出现问题时应采取的各项措施（第二部分）

https://aws.amazon.com/blogs/database/debugging-your-aws-dms-migrations-what-to-do-when-things-go-wrong-part-2/

CDCChangesDiskTarget

CDCChangesDiskTarget 是指在磁盘上累积并等待被提交至目标的总行数。

我们应尽量确保在内存内处理这些行。如果 CDCChangesDiskTarget 的值不断增加，则可能代表出现了两个问题：复制实例上的内存可能被过度使用；或者目标数据库实例已经无法同步接收 Amazon DMS 发送来的变更。

创建 CloudWatch 告警

以下 CloudFormation 堆栈将为您的 Amazon DMS 任务创建 CloudWatch 警报：

https://console.aws.amazon.com/cloudformation/home?#/stacks/new?stackName=DMS-Task-Alarms&templateURL=https://aws-database-blog.s3.amazonaws.com/artifacts/automate_monitoring_db_migration_DMS/1.DMS-Task-Alarms.yaml

向 CloudFormation 堆栈提供以下信息：

• 栈名称

• Amazon DMS 任务标识符

• Amazon DMS 复制实例名称

• SNS 主题 ARN

其余所有设置皆可保留为默认值。

创建 Amazon DMS 事件订阅

当复制任务或复制实例发生特定事件时（例如创建或删除实例时），您可以通过创建的 Amazon DMS 事件订阅接收到相应通知。

对于复制任务，请为以下事件创建订阅：

• 配置变更

• 创建

• 删除

• 失败

• 状态变更

对于复制实例，请为以下事件创建订阅：

• 配置变更

• 创建

• 删除

• 故障转移

• 失败

• 低存储容量

• 维护

创建事件订阅后，Amazon DMS 会将事件通知发送至您所提供的目标地址处。您可能希望创建多个不同的订阅，例如由一个订阅接收所有事件通知，再由另一订阅处理仅涉及生产环境 Amazon DMS 资源的关键事件。

您可以通过在 Amazon DMS 控制台将 Enabled 选项设置为 No，或者使用 Amazon DMS API 将 Enabled 参数设置为 false 来轻松关闭通知但又继续保留订阅。关于更多详细信息，请参阅在 Amazon 数据库迁移服务中使用事件与通知。

• 在 Amazon 数据库迁移服务中使用事件与通知

https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Events.html

以下 CloudFormation 堆栈，将为您的 Amazon DMS 任务创建事件订阅：

https://console.aws.amazon.com/cloudformation/home?#/stacks/new?stackName=DMS-Task-Errors&templateURL=https://aws-database-blog.s3.amazonaws.com/artifacts/automate_monitoring_db_migration_DMS/3.DMS-Task-Errors.yaml

需要为堆栈提供以下信息：

• 堆栈名称

• Amazon DMS 任务名称

• SNS 主题 ARN

其余所有设置皆可保留为默认值。

为 CloudWatch 日志中的错误消息创建 SNS 通知

Amazon DMS 可以将详细任务信息发布至 CloudWatch 日志。您可以利用这一点来在任务运行的时候监控其运行状况并且诊断发生的任何问题

在默认情况下，日志将被存储在日志组 dms-tasks-的 dms-task-日志流当中。关于更多详细信息，请参阅对任务设置进行日志记录。

• 对任务设置进行日志记录

https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Tasks.CustomizingTasks.TaskSettings.Logging.html

要获取关于 CloudWatch 日志中出现错误消息的通知，请在日志组上创建订阅过滤器，具体参阅以下 Python 脚本：

from __future__ import print_functionimport jsonimport base64, zlibimport boto3import os
def logstream_handler(event, context):    bstream_data = event.get("awslogs").get("data")    decoded_data = json.loads(zlib.decompress(base64.b64decode(bstream_data),16 + zlib.MAX_WBITS))    client = boto3.client('sns')    subscriptionFilters = decoded_data.get("subscriptionFilters")    subject = ""    if subscriptionFilters:        subject = "Log Filter Alert : {0}".format(subscriptionFilters[0])    decoded_msg = decoded_data.get("logEvents")    msg = "logGroup : {0}\nlogStream : {1}".format(        decoded_data.get("logGroup"),        decoded_data.get("logStream"))    msg = "{0}\n\nMessages: \n".format(msg)    for m in decoded_msg:        msg = "{0}\n{1}".format(msg,m.get("message"))    topicARN=os.environ.get("topicARN")    args = {}    args["TargetArn"]=topicARN    args["Message"]=msg    if subject:        args["Subject"]=subject    response = client.publish(**args)    return {        "statusCode": 200,        "body": json.dumps('Sent Message.')    }

以下 CloudFormation 堆栈可创建用于发送 SNS 错误日志通知的环境：

https://console.aws.amazon.com/cloudformation/home?#/stacks/new?stackName=DMS-Task-Errors&templateURL=https://aws-database-blog.s3.amazonaws.com/artifacts/automate_monitoring_db_migration_DMS/3.DMS-Task-Errors.yaml

为堆栈提供以下信息：

• 堆栈名称

• 日志组名称

• SNS 主题 ARN

• 过滤模式

其余所有设置皆可保留为默认值。

创建 Lambda 函数，为重复出现的 CloudWatch 告警发送 SNS 通知

您可以创建多个 CloudWatch 告警，借此了解告警状态何时发生变更。在某些情况下，告警可能长时间处于活跃状态，导致您可能错过此前已发送的告警。要获取重复告警，我们为您提供一个 Lambda 函数，用于检查告警的状态以及当前状态的持续时间，并据此发送通知。

以下可以发送 SNS 通知的 Python 脚本将由 CloudWatch Events 规则调用：

import jsonimport boto3import os
cloudwatch = boto3.client('cloudwatch')sns = boto3.client('sns')
subject_str = '{}: "{}" in {}'message_str = """You are receiving this email because your Amazon CloudWatch Alarm "{}" in the {} region has entered the {} state, because "{}".
Alarm Details :    - Name: {}    - Description: {}    - Reason for State Change: {}

Monitored Metric:    - MetricNamespace: {}    - MetricName: {}    - Dimensions: {}    - Period: {}    - Statistic: {}    - Unit: {}    - TreatMissingData: {}"""
def send_alarm(topic, subject, message):    """ Sends SNS Notification to given topic """    response = sns.publish(                    TopicArn=topic,                    Message=message,                    Subject=subject                )    print("Alarm Sent Subject : {}".format(subject))    return
def main_handler(event, context):    """        Describes existing alarms in current region and check it's state        If state matches to alarmState that sent as input send alarm.
        Parameters        ----------        alarmNames - ['string']        alarmState - string (Alarm/OK/INSUFFICIENT_DATA)    """    alarm_names = event['alarmNames']    alarm_state = event.get('alarmState', 'Alarm').lower()    region = os.environ["AWS_REGION"]    response = cloudwatch.describe_alarms(        AlarmNames=alarm_names        )    metric_alarms = response["MetricAlarms"]    if len(metric_alarms) == 0:        return {            'statusCode': 200,            'body': json.dumps('No Alarms Configured')        }    for alarm in metric_alarms:        if alarm["StateValue"].lower() != alarm_state:            continue        topics = alarm["AlarmActions"] if alarm_state == 'alarm' else alarm["OKActions"] if alarm_state == 'ok' else alarm['InsufficientDataActions'] if alarm_state == 'insufficient_data' else []        if len(topics) == 0:            print('No Topics Configured for state %s to %s' %(alarm_state, alarm['AlarmName']))            continue        subject = subject_str.format(alarm["StateValue"], alarm['AlarmName'], region)        message = message_str.format(alarm['AlarmName'], region,                                    alarm['StateValue'], alarm['StateReason'],                                    alarm['AlarmName'], alarm['AlarmDescription'],                                    alarm['StateReason'], alarm['Namespace'],                                    alarm['MetricName'], str(["{}={}".format(d['Name'], d['Value']) for d in alarm["Dimensions"]]),                                    alarm['Period'], alarm['Statistic'],                                    alarm.get('Unit', 'not specified'), alarm['TreatMissingData'])        for topic in topics:            send_alarm(topic, subject, message)    return {        'statusCode': 200,        'body': json.dumps('Success')    }Python

以下 CloudFormation 堆栈可创建用于发送 SNS 错误日志消息通知的环境：

https://console.aws.amazon.com/cloudformation/home?#/stacks/new?stackName=DMS-Recurring-Alarms&templateURL=https://aws-database-blog.s3.amazonaws.com/artifacts/automate_monitoring_db_migration_DMS/4.DMS-Recurring-Alarms.yaml

向堆栈提供以下信息：

• 堆栈名称

• Amazon DMS 任务名称

• SNS 主题 ARN

其余所有设置皆可保留为默认值。

总结

在本文中，我们向您介绍了如何使用 CloudWatch、Amazon DMS 事件订阅、Amazon SNS 以及 Amazon Lambda 以自动化方式对 Amazon DMS 复制任务进行监控与告警。

使用这套解决方案，您无需使用控制台即可轻松跟踪复制任务的状态。系统会向您通报每个变更事件；如果发生错误，您还会收到相应的告警。

希望本文能够帮助大家了解如何监控 Amazon DMS 数据库迁移过程.

本篇作者

Venkata Naveen Koppula

亚马逊云科技

专业服务助理顾问

他专注于 Aamzon DMS、SCT 以及 Aurora PostgreSQL 等项目，致力于为客户提供最佳体验。

Vijaya Diddi

亚马逊云科技

专业服务助理顾问

她专注于 Amazon DMS、SCT、Amazon Config 以及 SSM Documents。她乐于使用 Python 语言开发自动化工具，借此大大降低迁移工作的实施难度。