將DMS內的動作記錄匯出到SLS - Data Management

Data Management支援將動作記錄匯出至阿里雲Log Service（SLS），便於對動作記錄進行加工和分析。

前提條件

已開通Log Service。更多資訊，請參見開通Log Service。
已建立Log ServiceProject和Logstore。具體操作，請參見管理Project和建立基礎Logstore。
已在DMS的執行個體管理中將建立的Project錄入至DMS。錄入執行個體的操作，請參見雲資料庫錄入。
匯出的目標Logstore必須為空白，且全文索引和欄位索引必須至少啟用一種。全文索引和欄位索引的詳細資料，請參見建立索引。

背景資訊

動作記錄指使用者通過DMS系統發起所有操作的流水賬式日誌，詳情請參見功能說明。

費用說明

匯出DMS動作記錄到SLS功能，暫不收費。
當涉及的Logstore的計費模式為按使用功能計費時，Log Service採集到日誌後，將根據儲存空間、讀取流量、請求數量、資料加工、資料投遞等進行收費。
當涉及的Logstore的計費模式為按寫入資料量計費時，Log Service採集到日誌後，Log Service將根據原始寫入資料量等進行收費。

操作步驟

登入Data Management 5.0。
單擊控制台左上方的表徵圖，選擇全部功能 > 安全與容災（DBS） > Action Trail。
說明
若您使用的是非極簡模式的控制台，在頂部功能表列中，選擇安全與容災（DBS） > Action Trail。
選擇匯出日誌頁簽，單擊右上方建立任務。

在建立匯出任務對話方塊中，配置如下資訊。

配置項	是否必填	說明
任務名稱	是	匯出任務名稱，便於後續尋找。
目標SLS	是	Log Service的資源嵌入式管理單元Project。
SLS Logstore	是	將日誌匯出至建立的Logstore。該Logstore需要建立索引並不含任何日誌條目。說明您可單擊同步字典，再單擊確認，DMS可自動採集Logstore中的中繼資料資訊。
功能模組	是	選擇需要匯出DMS哪些功能模組的日誌（與動作記錄中的模組對應），包含不限、許可權、資料Owner、資料查詢、查詢結果匯出、跨庫查詢結果匯出等功能，預設選擇不限。
調度方式	是	選擇本次任務的調度方式。單次：指成功建立匯出任務後，僅匯出一次。周期：可選擇按日、周或月迴圈多次匯出日誌至Logstore。周期調度任務第一次會匯出從日誌開始時間到第一次調度開始時間範圍內，您在DMS產生的所有動作記錄，後續僅匯出增量的日誌。具體配置，請參見周期調度。
日誌時間範圍	否	說明調度方式選擇單次時會出現此配置項。匯出某時間範圍內的日誌。不填寫該配置項則預設匯出三年內的日誌。
日誌開始時間	否	說明調度方式選擇周期時會出現此配置項。周期任務沒有截止時間。 DMS日誌記錄的開始時間，不填寫則預設為建立匯出任務時間對應三年前的時間。

單擊確認，會建立一個匯出日誌任務，同時，系統還會在您的Logstore中建立一些用於後續查詢分析資料的索引欄位，例如dbId、dbName、dbUser等。
- 對於單次任務，僅匯出一次日誌，當任務的狀態為運行成功時，表示日誌匯出成功。
  說明
  因Logstore索引延遲生效，所以單次調度任務會在建立成功後的90秒左右開始執行。
- 對於周期任務，會多次匯出日誌，且匯出前和匯出後的任務狀態均為待調度。您可通過查看任務日誌，判斷某次任務是否執行成功。
您還可以在目標任務行操作列下進行如下操作。
- 查詢：單擊查詢，系統自動跳轉至SQL Console頁面，單擊查詢，在頁面下方的執行結果地區可查看匯出至Logstore的日誌。
- 任務日誌：單擊任務日誌，查看任務開始、結束時間、投遞日誌數量、任務狀態等資訊。
- 暫停：單擊暫停，在彈出的提示對話方塊中，單擊確認，周期任務會被暫停執行。
- 重啟：單擊重啟，在彈出的提示對話方塊中，單擊確認，可重新啟動已被暫停執行的周期任務。

周期調度

配置項	說明
調度周期	選擇調度任務的周期：日：按每日一次的頻率執行任務調度，需要配置每日調度的具體時間。周：以周為周期，每個指定天執行一次任務調度，需要配置指定時間和具體時間。月：以月為周期，每個指定天執行一次任務調度，需要配置指定時間和具體時間。
指定時間	如果調度周期為周，選擇星期幾執行任務，支援多選。如果調度周期為月，選擇每月幾號執行任務，支援多選。
具體時間	設定執行任務流的具體時間。例如配置02:55，系統將在指定天的02時55分執行任務。
cron運算式	不需要手動設定，系統會根據您配置的周期、具體時間自動展現。

代碼實現

也可以通過Java代碼的方式實現日誌的匯出。

說明

對應的帳號需具備DMS的GetOpLog許可權，以及 SLS 的PutLogs和CreateIndex許可權。
為提升安全性，建議在工程代碼中採用無 AK 的憑據配置方式，具體配置方法請參見：管理訪問憑據。

<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);
    }

    // 建立DMS OpenAPI Client，用於調用GetOpLog
    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());
        // Endpoint 請參考 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;
    }

    // 建立SLS OpenAPI Client，用於調用PutLogs
    public static synchronized com.aliyun.openservices.log.Client createSlsClient() throws Exception {
        if (slsClient != null) {
            return slsClient;
        }
        com.aliyun.credentials.Client credentialClient = getCredentialClient();
        // Endpoint 請參考 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 OpenAPI 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";
        // DMS 動作記錄結束時間（精確到秒）
        String endStr = "2025-01-13 09:15:00";
        // SLS project名
        String project = "project";
        // SLS logStore名
        String logStore = "logStore";

        // 建立sls logStore索引，一個logStore建立一次即可
        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日誌量較大，為避免OpenAPI請求逾時，每次請求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);

            // 移動到下一天的 00:00:00
            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);
        }
    }
}

Data Management：匯出DMS動作記錄到阿里雲Log Service

前提條件

背景資訊

費用說明

操作步驟

周期調度

代碼實現

相關文檔