DMS の操作ログを SLS にエクスポート - Data Management - Alibaba Cloud ドキュメントセンター

Data Management (DMS) では、操作ログを Simple Log Service (SLS) にエクスポートして、一元的に保存・クエリ・データ変換・分析できます。

前提条件

作業を開始する前に、以下の要件を満たしていることを確認してください。

Simple Log Service アカウントが有効化されていること。詳細については、「Simple Log Service の有効化」をご参照ください。
SLS プロジェクトおよび Logstore が作成済みであること。詳細については、「プロジェクトの管理」および「基本的な Logstore の作成」をご参照ください。
インスタンス管理 ページで、データベースインスタンスが DMS に追加済みであること。詳細については、「ApsaraDB インスタンスの追加」をご参照ください。
ログエントリが存在せず、フルテキストインデックスまたはフィールドインデックスが有効化された Logstore が用意されていること。詳細については、「インデックスの作成」をご参照ください。
(Java 実装の場合のみ) DMS の GetOpLog 権限、および SLS の PutLogs 権限と CreateIndex 権限が必要です。

課金

DMS の操作ログを SLS にエクスポートする機能は無料です。
ログが SLS に収集された後は、Logstore の課金方法に基づいて料金が発生します。機能単位課金 の場合、ストレージ容量、読み取りトラフィック、リクエスト数、データ変換、データ転送に対して課金されます。詳細については、課金ドキュメントをご参照ください。
取り込みデータ量課金 の場合、取り込まれた生データ量に対して課金されます。詳細については、課金ドキュメントをご参照ください。

Java を使用したログのエクスポート

コンソールの代わりに、Java SDK を使用してプログラムでログをエクスポートすることもできます。このコードでは、DMS の GetOpLog API を呼び出してログを取得し、PutLogs を使用して SLS に書き込みます。

説明

(Java 実装の場合のみ) DMS の GetOpLog 権限、および SLS の PutLogs 権限と CreateIndex 権限が必要です。
プロジェクトコード内に AccessKey をハードコードしないでください。認証情報を安全に設定する方法の詳細については、「アクセス認証情報の管理」をご参照ください。

以下の Maven 依存関係を追加します。

<dependency>
  <groupId>com.aliyun</groupId>
  <artifactId>dms_enterprise20181101</artifactId>
  <version>2.0.0</version>
</dependency>

<dependency>
  <groupId>com.aliyun.openservices</groupId>
  <artifactId>aliyun-log</artifactId>
  <version>0.6.100</version>
</dependency>

package org.example;


import com.aliyun.dms_enterprise20181101.models.GetOpLogResponse;
import com.aliyun.dms_enterprise20181101.models.GetOpLogResponseBody;
import com.aliyun.openservices.log.common.LogItem;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Optional;


public class ExportDmsOperLogExample {

    private static com.aliyun.dms_enterprise20181101.Client dmsClient = null;

    private static com.aliyun.openservices.log.Client slsClient = null;

    public static com.aliyun.credentials.Client getCredentialClient() {
        com.aliyun.credentials.models.Config credentialConfig = new com.aliyun.credentials.models.Config();
        credentialConfig.setType("access_key");
        // AccessKey ID を環境変数から取得します。
        credentialConfig.setAccessKeyId(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"));
        // AccessKey Secret を環境変数から取得します。
        credentialConfig.setAccessKeySecret(System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET"));

        return new com.aliyun.credentials.Client(credentialConfig);
    }

    // GetOpLog を呼び出すための DMS クライアントを作成します。
    public static synchronized com.aliyun.dms_enterprise20181101.Client createDmsClient() throws Exception {
        if (dmsClient != null) {
            return dmsClient;
        }
        com.aliyun.teaopenapi.models.Config config = new com.aliyun.teaopenapi.models.Config()
                .setCredential(getCredentialClient());
        // エンドポイントについては、https://api.aliyun.com/product/dms-enterprise をご参照ください。
        config.endpoint = "dms-enterprise.cn-hangzhou.aliyuncs.com";
        dmsClient = new com.aliyun.dms_enterprise20181101.Client(config);
        return dmsClient;
    }

    // PutLogs を呼び出すための SLS クライアントを作成します。
    public static synchronized com.aliyun.openservices.log.Client createSlsClient() throws Exception {
        if (slsClient != null) {
            return slsClient;
        }
        com.aliyun.credentials.Client credentialClient = getCredentialClient();
        // エンドポイントについては、https://api.aliyun.com/product/Sls をご参照ください。
        String endpoint = "cn-hangzhou.log.aliyuncs.com";
        slsClient = new com.aliyun.openservices.log.Client(endpoint, credentialClient.getAccessKeyId(), credentialClient.getAccessKeySecret());
        return slsClient;
    }

    // 指定された時間範囲およびページの DMS 操作ログを取得するために GetOpLog を呼び出します。
    public static List<GetOpLogResponseBody.GetOpLogResponseBodyOpLogDetailsOpLogDetail> getLogs(com.aliyun.dms_enterprise20181101.Client client, String startTime, String endTime, Integer pageNumber, Integer pageSize) {
        com.aliyun.dms_enterprise20181101.models.GetOpLogRequest getOpLogRequest = new com.aliyun.dms_enterprise20181101.models.GetOpLogRequest()
                .setStartTime(startTime)
                .setEndTime(endTime)
                .setPageSize(pageSize)
                .setPageNumber(pageNumber);
        com.aliyun.teautil.models.RuntimeOptions runtime = new com.aliyun.teautil.models.RuntimeOptions();
        try {
            GetOpLogResponse response = client.getOpLogWithOptions(getOpLogRequest, runtime);
            return Optional.ofNullable(response.getBody())
                    .map(GetOpLogResponseBody::getOpLogDetails)
                    .map(GetOpLogResponseBody.GetOpLogResponseBodyOpLogDetails::getOpLogDetail)
                    .orElse(new ArrayList<>());
        } catch (Exception e) {
            System.out.println(e.getMessage());
            throw new RuntimeException(e.getMessage());
        }
    }

    // PutLogs を呼び出してログを SLS Logstore に書き込みます。
    public static void putLogs(com.aliyun.openservices.log.Client client, String project, String logStore, List<GetOpLogResponseBody.GetOpLogResponseBodyOpLogDetailsOpLogDetail> logDetailList) {
        List<LogItem> logItemList = new ArrayList<>();
        for (GetOpLogResponseBody.GetOpLogResponseBodyOpLogDetailsOpLogDetail logDetail : logDetailList) {
            LogItem logItem = new LogItem((int) (new Date().getTime() / 1000));
            logItem.PushBack("module", logDetail.getModule());
            logItem.PushBack("database", logDetail.getDatabase());
            logItem.PushBack("userId", logDetail.getUserId());
            logItem.PushBack("opUserId", String.valueOf(logDetail.getOpUserId()));
            logItem.PushBack("userNick", logDetail.getUserNick());
            logItem.PushBack("opTime", logDetail.getOpTime());
            logItem.PushBack("opContent", logDetail.getOpContent());
            logItem.PushBack("orderId", String.valueOf(logDetail.getOrderId()));
            logItemList.add(logItem);
        }
        try {
            client.PutLogs(project, logStore, "", logItemList, "");
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println(e.getMessage());
            throw new RuntimeException(e.getMessage());
        }
    }

    // インデックスを作成します。Logstore は一度だけ初期化する必要があります。
    public static void createIndex(com.aliyun.openservices.log.Client client, String project, String logStore) {
        try {
            GetIndexResponse getIndexResponse = client.GetIndex(project, logStore);
            Index index = getIndexResponse.GetIndex();
            IndexLine indexLine = index.GetLine();
            List<String> indexToken = null;
            if (indexLine != null) {
                indexToken = indexLine.GetToken();
            }
            IndexKeys keys = new IndexKeys();
            List<String> logStoreKeys = List.of("module", "database", "userId", "opUserId",
                    "userNick", "opTime", "opContent", "orderId");
            for (String logStoreKey : logStoreKeys) {
                IndexKey key = new IndexKey();
                key.SetType("text");
                key.SetChn(true);
                key.SetDocValue(false);
                key.SetToken(indexToken);
                keys.AddKey(logStoreKey, key);
                index.SetKeys(keys);
            }
            client.UpdateIndex(project, logStore, index);
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println(e.getMessage());
            throw new RuntimeException(e.getMessage());
        }
    }

    public static void main(String[] args) throws Exception {
        // DMS 操作ログの開始時刻および終了時刻 (秒単位で正確)。
        String startStr = "2025-01-10 14:30:00";
        String endStr = "2025-01-13 09:15:00";
        // SLS プロジェクトおよび Logstore の名前。
        String project = "project";
        String logStore = "logStore";

        // Logstore インデックスを初期化します。Logstore ごとに 1 回だけ実行する必要があります。
        createIndex(createSlsClient(), project, logStore);

        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        LocalDateTime start = LocalDateTime.parse(startStr, formatter);
        LocalDateTime end = LocalDateTime.parse(endStr, formatter);

        if (start.isAfter(end)) {
            throw new IllegalArgumentException("開始時刻は終了時刻より後にはできません。");
        }

        LocalDateTime current = start;

        // DMS のログ量が多いため、API タイムアウトを回避するために、コードは 1 日分ずつログを取得します。
        while (!current.isAfter(end)) {
            // 各日について、23:59:59 または全体の終了時刻のいずれか早い方を使用します。
            LocalDateTime dayEnd = current.toLocalDate().atTime(23, 59, 59);
            if (dayEnd.isAfter(end)) {
                dayEnd = end;
            }

            String segmentStartStr = current.format(formatter);
            String segmentEndStr = dayEnd.format(formatter);

            // DMS からログを取得し、SLS に書き込みます。
            process(segmentStartStr, segmentEndStr, project, logStore);

            // 次の日の開始時刻に進めます。
            current = dayEnd.toLocalDate().plusDays(1).atStartOfDay();
        }
    }

    public static void process(String startTime, String endTime, String project, String logStore) throws Exception {
        int pageNumber = 1;
        int pageSize = 100;
        com.aliyun.dms_enterprise20181101.Client dmsClient = createDmsClient();
        com.aliyun.openservices.log.Client slsClient = createSlsClient();
        List<GetOpLogResponseBody.GetOpLogResponseBodyOpLogDetailsOpLogDetail> logs = getLogs(dmsClient, startTime, endTime, pageNumber, pageSize);
        while(logs.size() >= 100) {
            putLogs(slsClient, project, logStore, logs);
            pageNumber++;
            logs = getLogs(dmsClient, startTime, endTime, pageNumber, pageSize);
        }
    }
}

次のステップ

DMS の操作ログを SLS にエクスポートした後は、クエリと分析を実行できます。詳細については、「Simple Log Service での DMS 操作ログのクエリと分析」をご参照ください。