Proxy datasource & change data

Proxy JDBC datasource thu thập dữ liệu thay đổi (before/after image) ở tầng câu lệnh, phục vụ audit và revision. Ứng dụng thường không dùng trực tiếp — dependency hoạt động ngầm khi có govex-cloud-data-audit hoặc govex-cloud-data-revision.

Khi nào sử dụng

Ứng dụng dùng govex-cloud-data-audit, govex-cloud-data-revision hoặc govex-cloud-transaction — dependency được kéo theo, không cần khai báo thêm.
Cần tự lắng nghe dữ liệu thay đổi trước commit để xử lý nghiệp vụ riêng (qua bean ConnectionListener).
Cần tắt proxy datasource vì xung đột với datasource đặc thù hoặc để giảm overhead.
Cần loại trừ một lớp datasource cụ thể khỏi cơ chế proxy.
Cần điều tra vì sao một thao tác ghi không được ghi nhận audit — proxy chỉ thu change data khi luồng hiện tại có cid.

Cài đặt

<dependency>
    <groupId>vn.govex.cloud</groupId>
    <artifactId>govex-cloud-datasource</artifactId>
</dependency>

Version quản lý qua BOM — xem Cài đặt. Dependency kéo theo govex-cloud-sqlparser (parse SQL, metadata bảng) qua transitive dependency.

Cấu hình

Property	Mô tả	Mặc định
`govex.datasource.enabled`	Bật/tắt cấu hình tự động tạo proxy datasource	`true`
`govex.datasource.use-jdk-proxy`	Dùng proxy JDK thay cho CGLIB khi bọc datasource	`false`
`govex.datasource.excludes-for-auto-proxying`	Danh sách tên lớp datasource không tham gia tạo proxy tự động	rỗng

govex:
  datasource:
    enabled: true
    use-jdk-proxy: false
    excludes-for-auto-proxying:
      - com.example.CustomDataSource

Cơ chế proxy

Tạo proxy tự động

DataSourceProxyAutoConfiguration là auto-configuration duy nhất của module và chỉ hoạt động khi:

ứng dụng có ít nhất một bean DataSource;
govex.datasource.enabled không bị đặt thành false.

Auto-configuration chạy sau DataSourceAutoConfiguration và ở thứ tự ưu tiên thấp nhất (@AutoConfigureOrder(Ordered.LOWEST_PRECEDENCE)), nên mọi datasource do ứng dụng khai báo đều đã sẵn sàng trước khi bị bọc.

AutoDataSourceProxyCreator là một AbstractAutoProxyCreator duyệt các bean DataSource và bọc proxy cho từng bean, trừ khi:

lớp datasource nằm trong excludes-for-auto-proxying;
bean đã là vn.govex.cloud.api.Proxy.

Mỗi cặp được đăng ký vào DataSourceProxyHolder theo datasource gốc. Khi ứng dụng dừng, @PreDestroy xóa toàn bộ đăng ký khỏi holder.

Khi nào lời gọi đi qua proxy

Proxy không can thiệp vào mọi lời gọi. AutoDataSourceProxyAdvice chỉ chuyển lời gọi từ datasource gốc sang DataSourceProxy khi luồng hiện tại đang có cid (DataSourceContext.inContext()); ngoài phạm vi này, method được gọi trực tiếp trên datasource gốc.

cid (KEY_CID = "TX_CID") được lưu trong RootContext và bind theo một trong các cách:

govex-cloud-transaction bind xid làm cid khi vào method gắn @XidTransactional;
govex-cloud-data-audit bind trace id của request khi bắt đầu ghi audit;
ứng dụng tự gọi DataSourceContext.openScope(cid) / bind(cid) khi cần phạm vi dữ liệu riêng.

Hệ quả: ngoài cid, DataSourceProxy không nhận được lời gọi nào; change data chỉ được thu khi thao tác dữ liệu nằm trong một phạm vi có cid.

Khởi tạo proxy

DataSourceProxy đọc metadata của datasource thật ngay khi khởi tạo: jdbcUrl (đã loại bỏ thông tin xác thực), dbType, phiên bản database và resource id. Resource id dùng làm khóa cache metadata bảng:

MySQL: chuẩn hóa jdbcUrl dạng loadbalance (nhiều host ngăn cách bởi dấu phẩy) thành một resource id;
PostgreSQL: chỉ giữ tham số currentSchema;
Oracle: gắn thêm tên user để phân biệt schema;
còn lại: jdbcUrl bỏ phần tham số truy vấn.

DataSourceProxy.getConnection() trả về ConnectionProxy — connection này bọc statement, prepared statement, savepoint và autoCommit để phục vụ thu change data.

Thu before/after image

Khi connection được lấy từ proxy, mọi Statement/PreparedStatement tạo ra đều là proxy. ExecuteTemplate quyết định cách thực thi:

ngoài phạm vi cid: chạy thẳng xuống statement gốc;
trong phạm vi cid: parse SQL thành SQLRecognizer, chọn executor theo SQLType và dbType.

Các executor tương ứng với từng loại câu lệnh:

SQLType	Executor
`INSERT`	`InsertExecutor` (chọn theo dbType qua SPI: MySQL, MariaDB, Oracle, PostgreSQL)
`UPDATE`	`UpdateExecutor`
`DELETE`	`DeleteExecutor`
`INSERT_ON_DUPLICATE_UPDATE`	`MySQLInsertOnDuplicateUpdateExecutor` / `MariadbInsertOnDuplicateUpdateExecutor`
`UPDATE_JOIN`	`MySQLUpdateJoinExecutor` / `MariadbUpdateJoinExecutor`
khác	`PlainExecutor` (không thu change data)
nhiều recognizer	`MultiExecutor`

Luồng thực thi một câu lệnh DML:

Dựng before image bằng truy vấn SELECT trên dữ liệu liên quan.
Gọi beforeExecute của các listener.
Thực thi câu lệnh gốc.
Dựng after image.
Ghi ChangeData vào ConnectionContext của connection (appendChangeData).
Gọi afterExecute của các listener.

Nếu connection đang ở chế độ autoCommit, executor tự chuyển sang non-autocommit, chạy câu lệnh rồi commit, sau đó luôn khôi phục trạng thái autoCommit ban đầu. Khi lỗi, giao dịch đang mở được rollback trước khi khôi phục; ngoại lệ gốc được giữ nguyên và ngoại lệ rollback được đính kèm dưới dạng suppressed.

Với câu lệnh INSERT, connection proxy đọc metadata bảng trước để lấy danh sách cột khóa chính, phục vụ việc đọc generated keys; nếu không đọc được metadata thì câu lệnh vẫn được tạo theo cách thông thường.

Cache metadata bảng

AbstractTableMetaCache dùng Caffeine cache dùng chung cho mọi dialect:

tối đa 100.000 entry, expire sau 900 giây kể từ lần ghi, giá trị lưu dạng soft reference;
cache key gồm resource id, schema/catalog và tên bảng nên cùng tên bảng ở các schema khác nhau không ghi đè lẫn nhau;
khi gặp TableMetaException (ví dụ bảng vừa đổi cấu trúc), executor phát sự kiện làm mới metadata của datasource (tableMetaRefreshEvent) rồi ném lại ngoại lệ.

API `ConnectionListener`

Khai báo bean ConnectionListener; tất cả listener được gom tự động qua ObjectProvider và truyền vào mỗi connection khi connection được mở.

public interface ConnectionListener {

    default void beforeCommit(Connection connection, ConnectionContext context) throws SQLException {}

    default void afterCommit(ConnectionContext context) {}

    default void onCommitError(ConnectionContext context, Throwable throwable) {}

    default void beforeExecute(ConnectionContext context) {}

    default void afterExecute(ConnectionContext context) {}

    default void onExecuteError(ConnectionContext context, Throwable throwable) {}
}

Thời điểm các callback được gọi trong source hiện tại:

Callback	Thời điểm gọi	Ghi chú
`beforeCommit`	Trước khi commit thật, khi connection đang ở giao dịch theo xid	Ném `SQLException` sẽ hủy commit và kích hoạt `onCommitError`
`afterCommit`	Sau khi commit thành công	—
`onCommitError`	Commit hoặc `beforeCommit` thất bại	Ngoại lệ sau đó được bọc lại thành `SQLException` và ném tiếp
`beforeExecute`	Trước khi thực thi câu lệnh DML ở nhánh non-autocommit	Chỉ trong các executor DML
`afterExecute`	Sau khi dựng after image và ghi change data	Chỉ trong các executor DML
`onExecuteError`	Chưa được gọi ở source hiện tại	Method có trong interface nhưng chưa có call site — không nên dựa vào callback này

Điểm quan trọng: beforeCommit/afterCommit/onCommitError chỉ chạy khi connection đã được bind xid (ConnectionContext.inXidTransaction()). xid được bind tự động từ cid hiện hành ngay trước khi thực thi DML (BaseTransactionalExecutor.execute), nên chuỗi này chỉ có hiệu lực khi thao tác nằm trong phạm vi cid.

`ConnectionContext` và change data

ConnectionContext là ngữ cảnh giao dịch gắn với một ConnectionProxy, lưu xid, danh sách savepoint và change data theo từng savepoint.

API	Mô tả
`appendChangeData(ChangeData)`	Ghi nhận change data vào savepoint hiện hành
`hasChangeData()`	Kiểm tra ngữ cảnh có change data nào không
`getChangeData()`	Lấy toàn bộ change data theo thứ tự savepoint (đã gộp)
`inXidTransaction()`	Kiểm tra connection đã bind xid
`getXid()`	Lấy xid hiện hành, hoặc `null` nếu chưa bind
`isAutoCommitChanged()`	Cờ cho biết autoCommit đã từng bị đổi trong ngữ cảnh
`reset()`	Đặt lại ngữ cảnh, xóa xid và toàn bộ change data

Hành vi theo savepoint:

setSavepoint ghi nhận savepoint mới và chuyển change data về savepoint đó;
releaseSavepoint chuyển change data của savepoint bị giải phóng về savepoint hiện tại;
rollback(savepoint) xóa change data của savepoint đó và các savepoint sau nó;
rollback() toàn giao dịch luôn dọn sạch ngữ cảnh, kể cả khi rollback lỗi;
commit thành công cũng reset() ngữ cảnh sau khi chạy xong listener.

ChangeData mô tả thay đổi của một bảng trong một câu lệnh:

Trường	Kiểu	Mô tả
`sqlType`	`SQLType`	Loại câu lệnh (`INSERT`, `UPDATE`, `DELETE`, ...)
`tableName`	`String`	Tên bảng bị thay đổi
`beforeImage`	`TableRecords`	Dữ liệu trước thay đổi (có với `UPDATE`, `DELETE`)
`afterImage`	`TableRecords`	Dữ liệu sau thay đổi (có với `INSERT`, `UPDATE`)

TableRecords gồm metadata bảng, tên bảng và danh sách Row; mỗi Row gồm danh sách Field (tên, kiểu JDBC, giá trị, loại khóa). Giá trị Blob/Clob/Array/Ref được chuyển sang dạng serialize (SerialBlob, SerialClob, SerialArray...) trước khi lưu vào bản ghi. DataCompareUtils cung cấp tiện ích so sánh hai ảnh dữ liệu theo tên bảng, số dòng và giá trị từng cột.

Ví dụ listener thật

Audit — chuyển change data thành thay đổi audit

AuditChangeConnectionListener của govex-cloud-data-audit đọc change data trong beforeCommit, bỏ qua bảng nằm trong govex.audit.ignore-tables và chuyển từng dòng thành AuditRowChange gắn vào AuditContext đang bật capture:

@Override
public void beforeCommit(Connection connection, ConnectionContext context) {
    AuditContext auditContext = AuditContext.get();
    if (auditContext == null || !auditContext.isCapture() || !context.hasChangeData()) {
        return;
    }
    for (ChangeData changeData : context.getChangeData()) {
        if (changeData == null || isIgnoredTable(changeData.getTableName())) {
            continue;
        }
        convertToRowChanges(changeData).forEach(auditContext::addRowChange);
    }
}

Listener chỉ xử lý khi thao tác thuộc entity đích của audit; mọi lỗi chuyển đổi dữ liệu được ghi log và bỏ qua để không ảnh hưởng giao dịch nghiệp vụ. Chi tiết xem Lịch sử thao tác dữ liệu.

Undo log — ghi log hoàn tác trước commit

UndoLogConnectionListener của govex-cloud-transaction đẩy change data sang UndoLogManager trong beforeCommit:

@Override
public void beforeCommit(Connection connection, ConnectionContext context) throws SQLException {
    if (!context.hasChangeData()) {
        return;
    }
    undoLogManager.flushUndoLogs(TransactionContext.getXID(), context.getChangeData(), connection);
}

Sử dụng

Tắt proxy datasource

Khi không cần thu thập change data (ví dụ môi trường test hoặc datasource đặc thù):

govex:
  datasource:
    enabled: false

Loại trừ datasource khỏi proxy

govex:
  datasource:
    excludes-for-auto-proxying:
      - com.example.CustomDataSource

Luồng hoạt động

Sơ đồ dưới đây mô tả luồng proxy datasource: JDBC call đi qua proxy thu before/after image rồi chuyển tới ConnectionListener.

Lưu ý hiệu năng

Mỗi câu lệnh DML trong phạm vi cid phát sinh thêm truy vấn SELECT cho before image và after image; chi phí này chỉ xuất hiện khi thao tác nằm trong phạm vi cid, ngoài phạm vi gần như pass-through.
AutoDataSourceProxyAdvice cache method của DataSource theo tên và kiểu tham số (declaredMethodCache) nên không phản chiếu lặp lại trên mỗi lời gọi.
Cache metadata bảng giảm truy vấn metadata, nhưng có độ trễ tối đa 15 phút khi schema thay đổi ngoài Liquibase; khi phát hiện lệch metadata, executor tự phát sự kiện làm mới.
Change data được giữ trong bộ nhớ connection đến lúc commit; giao dịch dài hoặc batch lớn làm tăng bộ nhớ. Các cột Blob/Clob lớn bị chuyển sang dạng serialize nên càng tốn bộ nhớ hơn.
releaseSavepoint chỉ chuyển dữ liệu về savepoint hiện tại, không giải phóng ngay; chỉ commit/rollback mới xóa sạch ngữ cảnh.
Câu lệnh CallableStatement bị chặn khi luồng đang giữ cid (prepareCall ném ShouldNeverHappenException).
Câu lệnh rơi vào PlainExecutor (SQLType không xác định) không được thu change data.

Lưu ý

Dependency là nền tảng cho govex-cloud-data-audit và govex-cloud-transaction; ứng dụng thông thường không cần khai báo trực tiếp.
Không tự đăng ký bean DataSourceProxy thủ công; trường hợp này sẽ có log cảnh báo.
Hỗ trợ MySQL, MariaDB, Oracle, PostgreSQL thông qua SPI executor/table meta cache. File SPI EscapeHandler liệt kê handler cho Oracle, MySQL, MariaDB, PostgreSQL nhưng source hiện chỉ có OracleEscapeHandler và PostgresqlEscapeHandler — hai handler MySQL/MariaDB chưa tồn tại, cần kiểm tra khi dùng hai database này.
Proxy chỉ được tạo khi ứng dụng có ít nhất một DataSource.
Khi gặp vấn đề tương thích proxy CGLIB với datasource tùy biến, có thể thử govex.datasource.use-jdk-proxy: true.
Cần govex-cloud-sqlparser (parse SQL, metadata bảng) — đi kèm qua transitive dependency.

Khi nào sử dụng​

Cài đặt​

Cấu hình​

Cơ chế proxy​

Tạo proxy tự động​

Khi nào lời gọi đi qua proxy​

Khởi tạo proxy​

Thu before/after image​

Cache metadata bảng​

API ConnectionListener​

ConnectionContext và change data​

Ví dụ listener thật​

Audit — chuyển change data thành thay đổi audit​

Undo log — ghi log hoàn tác trước commit​

Sử dụng​

Tắt proxy datasource​

Loại trừ datasource khỏi proxy​

Luồng hoạt động​

Lưu ý hiệu năng​

Lưu ý​