DM8 数据库安全监视探针部署与问题处理

数据库安全监视软件 DBSMS 通过探针采集数据库运行及安全检查信息，并发送至 Kafka。本文整理 DM8KafkaInspector V2.1 的部署步骤、HA 环境自动拉起脚本，以及部署过程中遇到的问题处理。

一、探针安装与配置

1. 检查 Java 环境

使用 dmdba 用户检查 Java 环境，以下启动命令使用 JDK 8。

su - dmdba
java -version

2. 创建目录并复制程序

mkdir -p /home/dmdba/dm/dmagent/bin

上传并解压安装包。在解压目录中，将 DMKafkaIniCreater 和 DM8KafkaInspector 的内容复制到探针目录：

cp -rf DMKafkaIniCreater/* /home/dmdba/dm/dmagent/bin/
cp -rf DM8KafkaInspector/* /home/dmdba/dm/dmagent/bin/

将授权文件命名为 launch.key，放在 /home/dmdba/dm/dmagent/bin/。保留安装包中的密钥文件和 lib 目录。

3. 执行预处理脚本

使用 SYSDBA 登录数据库管理工具，执行安装包配套的 预处理脚本.sql

4. 计算校验码

cd /home/dmdba/dm/dmagent/bin
java -Djava.ext.dirs=./lib -cp DMKafkaIniCreater.jar creater.IniCreater md5-8

将输出的校验码填入配置文件的 CHECK_CODE。

5. 修改配置文件

vim /home/dmdba/dm/dmagent/bin/init_conf_dm8.txt

配置示例：

vim init_conf_dm8.txt
修改以下内容：
CHECK_FILE = ./DM8KafkaInspector.jar
CHECK_CODE = 0f26733eea59bb0bb4038cd5c5e73f73 #粘贴校验码
SERVER_IP = 223.254.7.147  					#数据库IP地址
SERVER_PORT = 7000							#数据库端口号
LOGIN_NAME = SYSDBA							#数据库登陆用户名
<<<<<< LOGIN_PASSWORD = SYSDBA >>>>>>		#数据库登陆密码

##kafka集群连接配置
KAFKA_TOPIC = pf_database		         	#kafka消息主题topic配置
KAFKA_BROKER_LIST=127.0.0.1:9092 			#kafka集群代理IP端口号列表

#5.数据库磁盘信息-备份文件监控 运行信息配置
LIST_BACKUP_PATH = /dbbak	            #改为备份目录/dbbak
#12.数据库非法访问-非授权IP访问数据库 运行信息配置
ENABLE_UNAUTHORIZED_IP_ACCESS = 否	
#13.数据库总表数量 运行信息配置
ENABLE_TABLE_NUMBER = 否	
##数据库配置项核查交互配置
ENABLE_CONF_CHECK = 否			
##日志管理参数配置
TIMESPAN_LOG_RETAIN = 15 					#日志保留时长，过期日志自动删除（单位：天）
THRESHOLD_LOG_DIR_SIZE = 10 				#日志文件夹占用磁盘空间大小上限，超过此上限，停止更新日志（单位：G）

SERVER_IP 填数据库设备接入平台的实际地址；Kafka 地址按平台提供的连接信息配置。口令保留两侧 <<<<<<、>>>>>> 标记。

LOCAL_PRINT_MESSAGE、LOCAL_LOG_MESSAGE 用于输出和记录消息及运行信息。日志保留 15 天，日志目录达到 10 GB 后停止更新日志。

6. 加密配置文件

cd /home/dmdba/dm/dmagent/bin
java -Djava.ext.dirs=./lib -cp DMKafkaIniCreater.jar creater.IniCreater t2k-8

配置修改后重新执行加密。

7. 前台启动验证

cd /home/dmdba/dm/dmagent/bin
java -Djava.ext.dirs=./lib -cp DM8KafkaInspector.jar com.dm8.kafka.Launcher

检查启动输出、检查信息和消息发送情况。前台验证完成后退出进程，再切换为后台运行。

8. 后台运行

cd /home/dmdba/dm/dmagent/bin
nohup java -Xms256M -Xmx4096M -XX:PermSize=256M -XX:MaxPermSize=4096M -Djava.ext.dirs=./lib -cp DM8KafkaInspector.jar com.dm8.kafka.Launcher >/dev/null &

9. 配置 DMHS 过滤

数据库同时作为 DMHS 源端时，在同步配置中排除探针监控表：

vim /home/dmdba/dm/dmhs/dmdbms.xml

在原有 <disable> 节点中增加：

<disable>
    <item>SYSDBA.MONITOR_*</item>
</disable>

保留已有过滤项，修改完成后重启 DMHS 服务：

/home/dmdba/dm/dmhs/dmhs_serverd restart

二、HA 环境自动检测与拉起

两台 HA 节点配置检测脚本。本机数据库服务运行时，检查探针进程，进程不存在则启动；数据库服务停止时，关闭探针进程。

1. 配置检测脚本

vim /home/dmdba/dm/dmagent/dbsms_check.sh

沿用原脚本的服务状态判断、进程计数和启停方式，探针工作目录统一为前文的安装目录：

#!/bin/bash

export LD_LIBRARY_PATH="$LD_LIBRARY_PATH:/home/dmdba/dm/dmdbms/bin"

dmserver_status=$(/home/dmdba/dm/dmdbms/bin/dmserverd status)

if [[ ${dmserver_status} == *running* ]]; then
    echo "dmserver running, check sms status"

    pid=$(ps -ef | grep DM8KafkaInspector | grep -v grep | wc -l)
    if [ "$pid" -gt 0 ]; then
        echo 'sms running'
    else
        cd /home/dmdba/dm/dmagent/bin
        java -Xms256M -Xmx4096M -XX:PermSize=256M -XX:MaxPermSize=4096M -Djava.ext.dirs=./lib -cp DM8KafkaInspector.jar com.dm8.kafka.Launcher 2>&1 &
    fi
else
    echo "dmserver not running, stop sms"

    pid=$(ps -ef | grep DM8KafkaInspector | grep -v grep | wc -l)
    if [ "$pid" -gt 0 ]; then
        kill $(ps -ef | grep DM8KafkaInspector | grep -v grep | awk '{print $2}')
    else
        echo 'sms not running'
    fi
fi

2. 配置定时任务

以 root 用户修改系统 crontab：

vim /etc/crontab

每分钟执行一次检测：

*/1 * * * * dmdba /bin/bash /home/dmdba/dm/dmagent/dbsms_check.sh

三、部署问题处理

1. 探针启动后不再打印检查信息

探针进程已启动，但打印启动日志后没有继续输出检查信息。

查看进程并导出线程栈：

ps -ef | grep DM8KafkaInspector
jstack 探针Java进程PID

线程栈显示主线程停在数据库查询调用链：

SocketInputStream.socketRead0
→ JDBCManager.executeQuery
→ DeleteActionHandler.<clinit>
→ InitializationService.loadInspectionClasses

数据库会话中存在长时间未返回的查询：

SELECT ISNULL(MAX(SEQ_NO), -1) AS START_SEQNO FROM WW_KAFKA_DELETE_ACTION;

当时未查看到对应锁信息。安装规范记录了 FN_KAFKA_CUT 在输入为空时可能无法退出循环的问题。

将函数中的退出条件：

EXIT WHEN M = 0 OR N = 0;

修改为：

EXIT WHEN M = 0 OR N = 0 OR M IS NULL;

修改函数并重新编译，重启探针后，日志恢复打印检查信息。

2. 运行时视图报“数据转换丢失警告”

探针日志反复出现：

[executeQuery] SELECT DBVER, START_TIME, DURING_TIME FROM VW_KAFKA_DATABASE_RUNTIME 失败：数据转换丢失警告

VW_KAFKA_DATABASE_STATUS 存在同类问题。使用探针配置的数据库用户执行查询：

SELECT DBVER, START_TIME, DURING_TIME FROM VW_KAFKA_DATABASE_RUNTIME;
SELECT DBVER, START_TIME, DURING_TIME FROM VW_KAFKA_DATABASE_STATUS;

检查视图所属模式：

SELECT OWNER, VIEW_NAME FROM DBA_VIEWS WHERE UPPER(VIEW_NAME) IN ('VW_KAFKA_DATABASE_RUNTIME', 'VW_KAFKA_DATABASE_STATUS');

处理时调整两个视图的输出表达式：DBVER 转为 VARCHAR(128)；启动时间按秒格式化后转为 VARCHAR(19)；运行时长取整后拼接，最终转为 VARCHAR(64)。

以 SYSDBA 用户执行：

CREATE OR REPLACE VIEW VW_KAFKA_DATABASE_RUNTIME AS
SELECT
    CAST(
        (
            SELECT MAX(REPLACE(REPLACE(REPLACE(BANNER, ' ', ''), 'DMDatabaseServer', ''), '-Build', ''))
            FROM V$VERSION
            WHERE UPPER(BANNER) LIKE 'DM%'
        ) AS VARCHAR(128)
    ) AS DBVER,
    CAST(TO_CHAR(T.START_TIME, 'YYYY-MM-DD HH24:MI:SS') AS VARCHAR(19)) AS START_TIME,
    CAST(
        TO_CHAR(TRUNC(DATEDIFF(MI, T.START_TIME, NOW()) / 1440))
        || '天'
        || TO_CHAR(TRUNC(MOD(DATEDIFF(MI, T.START_TIME, NOW()), 1440) / 60))
        || '小时'
        || TO_CHAR(MOD(DATEDIFF(MI, T.START_TIME, NOW()), 60))
        || '分钟'
        AS VARCHAR(64)
    ) AS DURING_TIME
FROM SYS.V$THREADS T
WHERE UPPER(T.NAME) = 'DM_LSNR_THD';

CREATE OR REPLACE VIEW VW_KAFKA_DATABASE_STATUS AS
SELECT
    CAST(
        (
            SELECT MAX(REPLACE(REPLACE(REPLACE(BANNER, ' ', ''), 'DMDatabaseServer', ''), '-Build', ''))
            FROM V$VERSION
            WHERE UPPER(BANNER) LIKE 'DM%'
        ) AS VARCHAR(128)
    ) AS DBVER,
    CAST(TO_CHAR(T.START_TIME, 'YYYY-MM-DD HH24:MI:SS') AS VARCHAR(19)) AS START_TIME,
    CAST(
        TO_CHAR(TRUNC(DATEDIFF(MI, T.START_TIME, NOW()) / 1440))
        || '天'
        || TO_CHAR(TRUNC(MOD(DATEDIFF(MI, T.START_TIME, NOW()), 1440) / 60))
        || '小时'
        || TO_CHAR(MOD(DATEDIFF(MI, T.START_TIME, NOW()), 60))
        || '分钟'
        AS VARCHAR(64)
    ) AS DURING_TIME
FROM SYS.V$THREADS T
WHERE UPPER(T.NAME) = 'DM_LSNR_THD';

修改后重新查询两个视图，检查查询是否正常；重启探针，检查日志中是否仍有转换报错和重复重试信息。

3. dmdba 定时任务被 PAM 拒绝

以 dmdba 用户执行探针定时任务时，/var/log/cron 中出现：

(dmdba) PAM ERROR (拒绝权限)
(dmdba) FAILED to authorize user with PAM (拒绝权限)

检查用户权限和 PAM 配置：

cat /etc/cron.allow
cat /etc/pam.d/crond

发现 /etc/cron.allow 未包含 dmdba，且 /etc/pam.d/crond 启用了：

account required pam_access.so

以 root 用户进行处理。在 /etc/cron.allow 原用户列表中增加 dmdba：

vim /etc/cron.allow

dmdba

再修改访问控制文件：

vim /etc/security/access.conf

在拒绝规则之前增加：

+:dmdba:cron crond

完成配置后停止探针，等待定时任务执行，探针进程成功重新拉起。