This is an automated email from the ASF dual-hosted git repository. madhan pushed a commit to branch ranger-2.6 in repository https://gitbox.apache.org/repos/asf/ranger.git
commit d06b18607415cf49e2737e36d64243c43542a931 Author: Rakesh Gupta <[email protected]> AuthorDate: Fri Oct 4 18:22:47 2024 +0530 RANGER-4809: Utility to migrate admin audit logs in x_trx_log table x_trx_log_v2 table Signed-off-by: Madhan Neethiraj <[email protected]> (cherry picked from commit c1aaffb63fc8690d3d194f8c4d7074b1b508d067) --- .../ranger-admin-transaction-log-migrate.sh | 48 +++ .../optimized/current/ranger_core_db_mysql.sql | 1 + .../patches/074-create-index-for-x_trx_log.sql | 26 ++ .../optimized/current/ranger_core_db_oracle.sql | 1 + .../patches/067-create-index-for-x_trx_log.sql | 25 ++ .../optimized/current/ranger_core_db_postgres.sql | 1 + .../patches/074-create-index-for-x_trx_log.sql | 33 ++ .../current/ranger_core_db_sqlanywhere.sql | 4 + .../patches/067-create-index-for-x_trx_log.sql | 19 + .../optimized/current/ranger_core_db_sqlserver.sql | 1 + .../patches/067-create-index-for-x_trx_log.sql | 26 ++ .../patch/cliutil/TrxLogV2MigrationUtil.java | 399 +++++++++++++++++++++ .../main/resources/META-INF/jpa_named_queries.xml | 26 ++ 13 files changed, 610 insertions(+) diff --git a/embeddedwebserver/scripts/ranger-admin-transaction-log-migrate.sh b/embeddedwebserver/scripts/ranger-admin-transaction-log-migrate.sh new file mode 100755 index 000000000..21691349a --- /dev/null +++ b/embeddedwebserver/scripts/ranger-admin-transaction-log-migrate.sh @@ -0,0 +1,48 @@ +#!/bin/bash + +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# Check if required environment variables are set +for var in RANGER_ADMIN_HOME RANGER_ADMIN_CONF RANGER_ADMIN_LOG_DIR; do + if [ -z "${!var}" ]; then + echo "Environment variable not found: ${var}" + exit 1 + else + echo "${var} : ${!var}" + fi +done + +RANGER_ADMIN_HEAP_DEFAULT_SIZE=1g +RANGER_ADMIN_HEAP_SIZE=${RANGER_ADMIN_HEAP_SIZE:-$RANGER_ADMIN_HEAP_DEFAULT_SIZE} +echo "Using heap size ${RANGER_ADMIN_HEAP_SIZE}" + +# Define JAVA_OPTS +JAVA_OPTS=" ${JAVA_OPTS} -XX:MetaspaceSize=100m -XX:MaxMetaspaceSize=200m -Xmx${RANGER_ADMIN_HEAP_SIZE} -Xms${RANGER_ADMIN_HEAP_SIZE} -Xloggc:${RANGER_ADMIN_LOG_DIR}/gc-worker.log -verbose:gc -XX:+PrintGCDetails " + +# Construct CLASSPATH +CLASSPATH="${RANGER_ADMIN_CONF}:${RANGER_ADMIN_HOME}/ews/webapp/WEB-INF/classes/:${SQL_CONNECTOR_JAR}:${RANGER_ADMIN_HOME}/ews/webapp/WEB-INF/classes/lib/*:${RANGER_ADMIN_HOME}/ews/webapp/WEB-INF/classes/META-INF:${RANGER_ADMIN_HOME}/ews/webapp/WEB-INF/lib/*:${RANGER_ADMIN_HOME}/ews/webapp/META-INF:${RANGER_ADMIN_HOME}/ews/lib/*:${JAVA_HOME}/lib/*" + +# Start Migration +nohup ${JAVA_HOME}/bin/java ${JAVA_OPTS} -Dlogdir="${RANGER_ADMIN_LOG_DIR}" -cp "${CLASSPATH}" org.apache.ranger.patch.cliutil.TrxLogV2MigrationUtil > ${RANGER_ADMIN_LOG_DIR}/trxlog_v1_migration.out 2>&1 & +VALUE_OF_PID=$! + +# Check if the command succeeded +if [ $? -ne 0 ]; then + echo "[Error] Migration failed. Please check ${RANGER_ADMIN_LOG_DIR} for details." + exit 1 +fi + +echo "Migrating Transaction logs has started with PID - ${VALUE_OF_PID}" diff --git a/security-admin/db/mysql/optimized/current/ranger_core_db_mysql.sql b/security-admin/db/mysql/optimized/current/ranger_core_db_mysql.sql index de262ef80..73ad54611 100644 --- a/security-admin/db/mysql/optimized/current/ranger_core_db_mysql.sql +++ b/security-admin/db/mysql/optimized/current/ranger_core_db_mysql.sql @@ -1834,6 +1834,7 @@ INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('069',UTC_TIMESTAMP(),'Ranger 2.5.0',UTC_TIMESTAMP(),'localhost','Y'); INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('072',UTC_TIMESTAMP(),'Ranger 2.5.0',UTC_TIMESTAMP(),'localhost','Y'); INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('073',UTC_TIMESTAMP(),'Ranger 2.5.0',UTC_TIMESTAMP(),'localhost','Y'); +INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('074',UTC_TIMESTAMP(),'Ranger 2.6.0',UTC_TIMESTAMP(),'localhost','Y'); INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('DB_PATCHES',UTC_TIMESTAMP(),'Ranger 1.0.0',UTC_TIMESTAMP(),'localhost','Y'); INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('J10001',UTC_TIMESTAMP(),'Ranger 1.0.0',UTC_TIMESTAMP(),'localhost','Y'); diff --git a/security-admin/db/mysql/patches/074-create-index-for-x_trx_log.sql b/security-admin/db/mysql/patches/074-create-index-for-x_trx_log.sql new file mode 100644 index 000000000..9a8f3c0d2 --- /dev/null +++ b/security-admin/db/mysql/patches/074-create-index-for-x_trx_log.sql @@ -0,0 +1,26 @@ +-- Licensed to the Apache Software Foundation (ASF) under one or more +-- contributor license agreements. See the NOTICE file distributed with +-- this work for additional information regarding copyright ownership. +-- The ASF licenses this file to You under the Apache License, Version 2.0 +-- (the "License"); you may not use this file except in compliance with +-- the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. + +drop procedure if exists create_index_for_x_trx_log; + +delimiter ;; +create procedure create_index_for_x_trx_log() begin +if not exists (SELECT * FROM INFORMATION_SCHEMA.STATISTICS WHERE table_schema=DATABASE() AND table_name='x_trx_log' AND index_name='x_trx_log_IDX_trx_id') then + CREATE INDEX x_trx_log_IDX_trx_id ON x_trx_log(trx_id); + end if; +end;; + +delimiter ; +call create_index_for_x_trx_log(); diff --git a/security-admin/db/oracle/optimized/current/ranger_core_db_oracle.sql b/security-admin/db/oracle/optimized/current/ranger_core_db_oracle.sql index b9a591f81..d9bd93c2f 100644 --- a/security-admin/db/oracle/optimized/current/ranger_core_db_oracle.sql +++ b/security-admin/db/oracle/optimized/current/ranger_core_db_oracle.sql @@ -1990,6 +1990,7 @@ INSERT INTO x_db_version_h (id,version,inst_at,inst_by,updated_at,updated_by,act INSERT INTO x_db_version_h (id,version,inst_at,inst_by,updated_at,updated_by,active) VALUES (X_DB_VERSION_H_SEQ.nextval, '060',sys_extract_utc(systimestamp),'Ranger 1.0.0',sys_extract_utc(systimestamp),'localhost','Y'); INSERT INTO x_db_version_h (id,version,inst_at,inst_by,updated_at,updated_by,active) VALUES (X_DB_VERSION_H_SEQ.nextval, '065',sys_extract_utc(systimestamp),'Ranger 1.0.0',sys_extract_utc(systimestamp),'localhost','Y'); INSERT INTO x_db_version_h (id,version,inst_at,inst_by,updated_at,updated_by,active) VALUES (X_DB_VERSION_H_SEQ.nextval, '066',sys_extract_utc(systimestamp),'Ranger 2.5.0',sys_extract_utc(systimestamp),'localhost','Y'); +INSERT INTO x_db_version_h (id,version,inst_at,inst_by,updated_at,updated_by,active) VALUES (X_DB_VERSION_H_SEQ.nextval, '067',sys_extract_utc(systimestamp),'Ranger 2.5.0',sys_extract_utc(systimestamp),'localhost','Y'); INSERT INTO x_db_version_h (id,version,inst_at,inst_by,updated_at,updated_by,active) VALUES (X_DB_VERSION_H_SEQ.nextval, '068',sys_extract_utc(systimestamp),'Ranger 2.5.0',sys_extract_utc(systimestamp),'localhost','Y'); INSERT INTO x_db_version_h (id,version,inst_at,inst_by,updated_at,updated_by,active) VALUES (X_DB_VERSION_H_SEQ.nextval, '069',sys_extract_utc(systimestamp),'Ranger 2.5.0',sys_extract_utc(systimestamp),'localhost','Y'); INSERT INTO x_db_version_h (id,version,inst_at,inst_by,updated_at,updated_by,active) VALUES (X_DB_VERSION_H_SEQ.nextval, '072',sys_extract_utc(systimestamp),'Ranger 2.5.0',sys_extract_utc(systimestamp),'localhost','Y'); diff --git a/security-admin/db/oracle/patches/067-create-index-for-x_trx_log.sql b/security-admin/db/oracle/patches/067-create-index-for-x_trx_log.sql new file mode 100644 index 000000000..1ef726b53 --- /dev/null +++ b/security-admin/db/oracle/patches/067-create-index-for-x_trx_log.sql @@ -0,0 +1,25 @@ +-- Licensed to the Apache Software Foundation (ASF) under one or more +-- contributor license agreements. See the NOTICE file distributed with +-- this work for additional information regarding copyright ownership. +-- The ASF licenses this file to You under the Apache License, Version 2.0 +-- (the "License"); you may not use this file except in compliance with +-- the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. +-- sync_source_info CLOB NOT NULL, + +DECLARE + v_index_exists number:=0; +BEGIN + SELECT COUNT(*) INTO v_index_exists FROM USER_INDEXES WHERE INDEX_NAME = upper('x_trx_log_IDX_trx_id') AND TABLE_NAME= upper('x_trx_log'); + IF (v_index_exists = 0) THEN + execute IMMEDIATE 'CREATE INDEX x_trx_log_IDX_trx_id ON x_trx_log(trx_id)'; + commit; + END IF; +END;/ diff --git a/security-admin/db/postgres/optimized/current/ranger_core_db_postgres.sql b/security-admin/db/postgres/optimized/current/ranger_core_db_postgres.sql index 66cf32f38..980f22f14 100644 --- a/security-admin/db/postgres/optimized/current/ranger_core_db_postgres.sql +++ b/security-admin/db/postgres/optimized/current/ranger_core_db_postgres.sql @@ -1924,6 +1924,7 @@ INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('069',current_timestamp,'Ranger 2.5.0',current_timestamp,'localhost','Y'); INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('072',current_timestamp,'Ranger 2.5.0',current_timestamp,'localhost','Y'); INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('073',current_timestamp,'Ranger 2.5.0',current_timestamp,'localhost','Y'); +INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('074',current_timestamp,'Ranger 2.6.0',current_timestamp,'localhost','Y'); INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('DB_PATCHES',current_timestamp,'Ranger 1.0.0',current_timestamp,'localhost','Y'); INSERT INTO x_user_module_perm (user_id,module_id,create_time,update_time,added_by_id,upd_by_id,is_allowed) VALUES diff --git a/security-admin/db/postgres/patches/074-create-index-for-x_trx_log.sql b/security-admin/db/postgres/patches/074-create-index-for-x_trx_log.sql new file mode 100644 index 000000000..34cf77ccc --- /dev/null +++ b/security-admin/db/postgres/patches/074-create-index-for-x_trx_log.sql @@ -0,0 +1,33 @@ +-- Licensed to the Apache Software Foundation (ASF) under one or more +-- contributor license agreements. See the NOTICE file distributed with +-- this work for additional information regarding copyright ownership. +-- The ASF licenses this file to You under the Apache License, Version 2.0 +-- (the "License"); you may not use this file except in compliance with +-- the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. + +select 'delimiter start'; +CREATE OR REPLACE FUNCTION create_index_for_x_trx_log() +RETURNS void AS $$ +DECLARE + v_attnum1 integer := 0; +BEGIN + select attnum into v_attnum1 from pg_attribute where attrelid in(select oid from pg_class where relname='x_trx_log') and attname in('trx_id'); + IF v_attnum1 > 0 THEN + IF not exists (select * from pg_index where indrelid in(select oid from pg_class where relname='x_trx_log') and indkey[0]=v_attnum1) THEN + CREATE INDEX x_trx_log_IDX_trx_id ON x_trx_log(trx_id); + END IF; + END IF; +END; +$$ LANGUAGE plpgsql; +select 'delimiter end'; + +select create_index_for_x_trx_log(); +select 'delimiter end'; diff --git a/security-admin/db/sqlanywhere/optimized/current/ranger_core_db_sqlanywhere.sql b/security-admin/db/sqlanywhere/optimized/current/ranger_core_db_sqlanywhere.sql index 9918e5040..649a0b250 100644 --- a/security-admin/db/sqlanywhere/optimized/current/ranger_core_db_sqlanywhere.sql +++ b/security-admin/db/sqlanywhere/optimized/current/ranger_core_db_sqlanywhere.sql @@ -2297,6 +2297,10 @@ INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active GO INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('073',CURRENT_TIMESTAMP,'Ranger 2.5.0',CURRENT_TIMESTAMP,'localhost','Y'); GO +INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('067',CURRENT_TIMESTAMP,'Ranger 3.0.0',CURRENT_TIMESTAMP,'localhost','Y'); +GO +INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('068',CURRENT_TIMESTAMP,'Ranger 3.0.0',CURRENT_TIMESTAMP,'localhost','Y'); +GO INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('DB_PATCHES',CURRENT_TIMESTAMP,'Ranger 1.0.0',CURRENT_TIMESTAMP,'localhost','Y'); GO INSERT INTO x_user_module_perm (user_id,module_id,create_time,update_time,added_by_id,upd_by_id,is_allowed) VALUES (dbo.getXportalUIdByLoginId('admin'),dbo.getModulesIdByName('Reports'),CURRENT_TIMESTAMP,CURRENT_TIMESTAMP,dbo.getXportalUIdByLoginId('admin'),dbo.getXportalUIdByLoginId('admin'),1); diff --git a/security-admin/db/sqlanywhere/patches/067-create-index-for-x_trx_log.sql b/security-admin/db/sqlanywhere/patches/067-create-index-for-x_trx_log.sql new file mode 100644 index 000000000..ae76a6de9 --- /dev/null +++ b/security-admin/db/sqlanywhere/patches/067-create-index-for-x_trx_log.sql @@ -0,0 +1,19 @@ +-- Licensed to the Apache Software Foundation (ASF) under one or more +-- contributor license agreements. See the NOTICE file distributed with +-- this work for additional information regarding copyright ownership. +-- The ASF licenses this file to You under the Apache License, Version 2.0 +-- (the "License"); you may not use this file except in compliance with +-- the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. + +CREATE INDEX IF NOT EXISTS x_trx_log_IDX_trx_id ON x_trx_log(trx_id); +GO + +EXIT diff --git a/security-admin/db/sqlserver/optimized/current/ranger_core_db_sqlserver.sql b/security-admin/db/sqlserver/optimized/current/ranger_core_db_sqlserver.sql index b44c683bf..9466f2783 100644 --- a/security-admin/db/sqlserver/optimized/current/ranger_core_db_sqlserver.sql +++ b/security-admin/db/sqlserver/optimized/current/ranger_core_db_sqlserver.sql @@ -4153,6 +4153,7 @@ INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('060',CURRENT_TIMESTAMP,'Ranger 1.0.0',CURRENT_TIMESTAMP,'localhost','Y'); INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('065',CURRENT_TIMESTAMP,'Ranger 1.0.0',CURRENT_TIMESTAMP,'localhost','Y'); INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('066',CURRENT_TIMESTAMP,'Ranger 2.5.0',CURRENT_TIMESTAMP,'localhost','Y'); +INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('067',CURRENT_TIMESTAMP,'Ranger 2.6.0',CURRENT_TIMESTAMP,'localhost','Y'); INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('068',CURRENT_TIMESTAMP,'Ranger 2.5.0',CURRENT_TIMESTAMP,'localhost','Y'); INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('069',CURRENT_TIMESTAMP,'Ranger 2.5.0',CURRENT_TIMESTAMP,'localhost','Y'); INSERT INTO x_db_version_h (version,inst_at,inst_by,updated_at,updated_by,active) VALUES ('072',CURRENT_TIMESTAMP,'Ranger 2.5.0',CURRENT_TIMESTAMP,'localhost','Y'); diff --git a/security-admin/db/sqlserver/patches/067-create-index-for-x_trx_log.sql b/security-admin/db/sqlserver/patches/067-create-index-for-x_trx_log.sql new file mode 100644 index 000000000..3efb9eebb --- /dev/null +++ b/security-admin/db/sqlserver/patches/067-create-index-for-x_trx_log.sql @@ -0,0 +1,26 @@ +-- Licensed to the Apache Software Foundation (ASF) under one or more +-- contributor license agreements. See the NOTICE file distributed with +-- this work for additional information regarding copyright ownership. +-- The ASF licenses this file to You under the Apache License, Version 2.0 +-- (the "License"); you may not use this file except in compliance with +-- the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. +GO +IF NOT EXISTS(SELECT * FROM sys.indexes WHERE name = 'x_trx_log_IDX_trx_id' AND object_id = OBJECT_ID('x_trx_log')) +BEGIN + CREATE NONCLUSTERED INDEX [x_trx_log_IDX_trx_id] ON [x_trx_log] + ( + [trx_id] ASC + ) + WITH (SORT_IN_TEMPDB = OFF, DROP_EXISTING = OFF, IGNORE_DUP_KEY = OFF, ONLINE = OFF) ON [PRIMARY] +END +Go + +EXIT; diff --git a/security-admin/src/main/java/org/apache/ranger/patch/cliutil/TrxLogV2MigrationUtil.java b/security-admin/src/main/java/org/apache/ranger/patch/cliutil/TrxLogV2MigrationUtil.java new file mode 100644 index 000000000..bb8ba0b0b --- /dev/null +++ b/security-admin/src/main/java/org/apache/ranger/patch/cliutil/TrxLogV2MigrationUtil.java @@ -0,0 +1,399 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.ranger.patch.cliutil; + +import java.sql.Timestamp; +import java.text.SimpleDateFormat; +import java.time.LocalDate; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Date; +import java.util.Iterator; +import java.util.List; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.AtomicReference; + +import org.apache.commons.collections.CollectionUtils; +import org.apache.ranger.db.RangerDaoManager; +import org.apache.ranger.entity.XXTrxLog; +import org.apache.ranger.entity.XXTrxLogV2; +import org.apache.ranger.patch.BaseLoader; +import org.apache.ranger.plugin.util.JsonUtilsV2; +import org.apache.ranger.util.CLIUtil; +import org.apache.ranger.view.VXTrxLog; +import org.apache.ranger.view.VXTrxLogV2; +import org.apache.ranger.view.VXTrxLogV2.ObjectChangeInfo; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Component; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; + +@Component +public class TrxLogV2MigrationUtil extends BaseLoader { + private static final Logger logger = LoggerFactory.getLogger(TrxLogV2MigrationUtil.class); + + private final Stats stats; + private TransactionTemplate txTemplate; + private Iterator<String> trxIdIter = Collections.emptyIterator(); + private int commitBatchSize = 25; + + + @Autowired + RangerDaoManager daoMgr; + + @Autowired + @Qualifier(value = "transactionManager") + PlatformTransactionManager txManager; + + + public static void main(String[] args) { + if (logger.isDebugEnabled()) { + logger.info("TrxLogV2MigrationUtil: main()"); + } + + try { + TrxLogV2MigrationUtil loader = (TrxLogV2MigrationUtil) CLIUtil.getBean(TrxLogV2MigrationUtil.class); + + loader.init(); + + while (loader.isMoreToProcess()) { + loader.load(); + } + + logger.info("Load complete. Exiting!!!"); + + System.exit(0); + } catch (Exception e) { + logger.error("Error loading", e); + + System.exit(1); + } + } + + public TrxLogV2MigrationUtil() { + this.stats = new Stats(); + } + + @Override + public void init() throws Exception { + txTemplate = new TransactionTemplate(txManager); + + txTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); + } + + @Override + public void execLoad() { + logger.info("==> TrxLogV2MigrationUtil.execLoad()"); + + try { + migrateTrxLogs(); + } catch (Exception e) { + logger.error("Error while migrating trx logs from v1 to v2", e); + } + + logger.info("<== TrxLogV2MigrationUtil.execLoad(): migration completed. Transaction counts(total: {}, migrated: {}, already-migrated: {}, failed: {})", stats.totalCount, stats.migratedCount, stats.alreadyMigratedCount, stats.failedCount); + } + + @Override + public void printStats() { + stats.logStats(); + } + + private void migrateTrxLogs() throws Exception { + logger.info("==> TrxLogV2MigrationUtil.migrateTrxLogs()"); + + int trxRetentionDays = config.getInt("ranger.admin.migrate.transaction_records.retention.days", -1); + int threadCount = config.getInt("ranger.admin.migrate.transaction_records.thread.count", 5); + + commitBatchSize = config.getInt("ranger.admin.migrate.transaction_records.commit.batch.size", 25); + + final List<String> uniqueTrxIdList; + + if (trxRetentionDays < 0) { + uniqueTrxIdList = daoMgr.getEntityManager().createNamedQuery("XXTrxLog.findDistinctTrxIds", String.class).getResultList(); + } else { + // Define start and end dates based on the retention period + Date startDate = Timestamp.valueOf(LocalDate.now().minusDays(trxRetentionDays).atStartOfDay()); + Date endDate = Timestamp.valueOf(LocalDate.now().atTime(23, 59, 59, 999999999)); + + uniqueTrxIdList = daoMgr.getEntityManager().createNamedQuery("XXTrxLog.findDistinctTrxIdsByTimeInterval", String.class) + .setParameter("startDate", startDate) + .setParameter("endDate", endDate) + .getResultList(); + } + + trxIdIter = uniqueTrxIdList.iterator(); + + stats.totalCount = uniqueTrxIdList.size(); + + logger.info("Found {} transactions to migrate", stats.totalCount); + + logger.info("Starting {} threads to migrate, commit batch size: {}", threadCount, commitBatchSize); + + LogMigrationThread[] migrationThreads = new LogMigrationThread[threadCount]; + + for (int i = 0; i < migrationThreads.length; i++) { + migrationThreads[i] = new LogMigrationThread(); + + migrationThreads[i].start(); + } + + for (LogMigrationThread migrationThread : migrationThreads) { + migrationThread.join(); + } + + logger.info("<== TrxLogV2MigrationUtil.migrateTrxLogs()"); + } + + private void migrateTrxLog(String trxId) { + if (logger.isDebugEnabled()) { + logger.debug("==> TrxLogV2MigrationUtil.createTransactionLogByTrxId()"); + } + + List<XXTrxLogV2> trxLogsV2 = daoMgr.getXXTrxLogV2().findByTransactionId(trxId); + + if (CollectionUtils.isNotEmpty(trxLogsV2)) { + if (logger.isDebugEnabled()) { + logger.debug("transaction({}): already migrated to v2", trxId); + } + + stats.incrAlreadyMigratedCount(); + } else { + List<XXTrxLog> v1TrxLogs = getV1TrxLogs(trxId); + + if (!v1TrxLogs.isEmpty()) { + ObjectChangeInfo objChangeInfo = new ObjectChangeInfo(); + + for (XXTrxLog v1TrxLog : v1TrxLogs) { + objChangeInfo.addAttribute(v1TrxLog.getAttributeName(), v1TrxLog.getPreviousValue(), v1TrxLog.getNewValue()); + } + + XXTrxLog firstTrxLog = v1TrxLogs.get(0); + + createTrxLog(firstTrxLog, objChangeInfo); + + if (logger.isDebugEnabled()) { + logger.debug("transaction({}): migrated {} v1 records", trxId, v1TrxLogs.size()); + } + + stats.incrMigratedCount(firstTrxLog.getId(), firstTrxLog.getCreateTime()); + } else { + if (logger.isDebugEnabled()) { + logger.debug("transaction({}): no v1 records found", trxId); + } + + stats.incrFailedCount(); + } + } + + if (logger.isDebugEnabled()) { + logger.debug("<== TrxLogV2MigrationUtil.createTransactionLogByTrxId()"); + } + } + + private List<XXTrxLog> getV1TrxLogs(String trxId) { + List<XXTrxLog> ret = Collections.emptyList(); + + try { + List<Object[]> rows = daoMgr.getEntityManager().createNamedQuery("XXTrxLog.findByTrxIdForMigration", Object[].class).setParameter("transactionId", trxId).getResultList(); + + if (rows != null) { + ret = new ArrayList<>(rows.size()); + + for (Object[] row : rows) { + XXTrxLog trxLog = new XXTrxLog(); + + trxLog.setId(toLong(row[0])); + trxLog.setCreateTime((Date) row[1]); + trxLog.setUpdateTime(trxLog.getCreateTime()); + trxLog.setAddedByUserId(toLong(row[2])); + trxLog.setUpdatedByUserId(trxLog.getAddedByUserId()); + trxLog.setObjectClassType(toInt(row[3])); + trxLog.setObjectId(toLong(row[4])); + trxLog.setObjectName((String) row[5]); + trxLog.setParentObjectClassType(toInt(row[6])); + trxLog.setParentObjectId(toLong(row[7])); + trxLog.setParentObjectName((String) row[8]); + trxLog.setAttributeName((String) row[9]); + trxLog.setPreviousValue((String) row[10]); + trxLog.setNewValue((String) row[11]); + trxLog.setTransactionId((String) row[12]); + trxLog.setAction((String) row[13]); + trxLog.setRequestId((String) row[14]); + trxLog.setSessionId((String) row[15]); + trxLog.setSessionType((String) row[16]); + + ret.add(trxLog); + } + } + } catch (Exception excp) { + logger.error("failed to get v1 transaction logs for trxId {}", trxId, excp); + + ret = Collections.emptyList(); + } + + return ret; + } + + private void createTrxLog(XXTrxLog v1TrxLog, ObjectChangeInfo objChangeInfo) { + VXTrxLogV2 trxLogV2 = new VXTrxLogV2(toVXTrxLog(v1TrxLog)); + + trxLogV2.setChangeInfo(objChangeInfo); + + XXTrxLogV2 dbObj = toDBObject(trxLogV2); + + dbObj.setAddedByUserId(v1TrxLog.getAddedByUserId()); + + daoMgr.getXXTrxLogV2().create(dbObj); + } + + private void fetchNextBatch(List<String> trxIds) { + trxIds.clear(); + + synchronized (this) { + for (int i = 0; i < commitBatchSize; i++) { + if (!trxIdIter.hasNext()) { + break; + } + + trxIds.add(trxIdIter.next()); + } + } + } + + private static VXTrxLog toVXTrxLog(XXTrxLog trxLog) { + VXTrxLog ret = new VXTrxLog(); + + ret.setId(trxLog.getId()); + ret.setCreateDate(trxLog.getCreateTime()); + ret.setUpdateDate(trxLog.getUpdateTime()); + ret.setObjectClassType(trxLog.getObjectClassType()); + ret.setObjectId(trxLog.getObjectId()); + ret.setObjectName(trxLog.getObjectName()); + ret.setParentObjectClassType(trxLog.getParentObjectClassType()); + ret.setParentObjectId(trxLog.getParentObjectId()); + ret.setParentObjectName(trxLog.getParentObjectName()); + ret.setAction(trxLog.getAction()); + ret.setRequestId(trxLog.getRequestId()); + ret.setTransactionId(trxLog.getTransactionId()); + ret.setSessionId(trxLog.getSessionId()); + ret.setSessionType(trxLog.getSessionType()); + + return ret; + } + + private static XXTrxLogV2 toDBObject(VXTrxLogV2 vObj) { + XXTrxLogV2 ret = new XXTrxLogV2(vObj.getObjectClassType(), vObj.getObjectId(), vObj.getObjectName(), + vObj.getParentObjectClassType(), vObj.getParentObjectId(), vObj.getParentObjectName(), vObj.getAction()); + + ret.setCreateTime(vObj.getCreateDate()); + ret.setChangeInfo(toJson(vObj.getChangeInfo())); + ret.setTransactionId(vObj.getTransactionId()); + ret.setSessionId(vObj.getSessionId()); + ret.setRequestId(vObj.getRequestId()); + ret.setSessionType(vObj.getSessionType()); + + return ret; + } + + private static String toJson(ObjectChangeInfo changeInfo) { + try { + return JsonUtilsV2.objToJson(changeInfo); + } catch (Exception excp) { + logger.error("Error converting ObjectChangeInfo to JSON", excp); + + return null; + } + } + + private static int toInt(Object obj) { + return obj instanceof Number ? ((Number) obj).intValue() : 0; + } + + private static long toLong(Object obj) { + return obj instanceof Long ? ((Number) obj).longValue() : 0L; + } + + class LogMigrationThread extends Thread { + @Override + public void run() { + List<String> trxIds = new ArrayList<>(commitBatchSize); + + for (fetchNextBatch(trxIds); !trxIds.isEmpty(); fetchNextBatch(trxIds)) { + txTemplate.execute((TransactionCallback<Void>) status -> { + for (String trxId : trxIds) { + migrateTrxLog(trxId); + } + + return null; + }); + } + } + } + + public static class Stats { + private long totalCount; + private final AtomicLong migratedCount = new AtomicLong(); + private final AtomicLong failedCount = new AtomicLong(); + private final AtomicLong alreadyMigratedCount = new AtomicLong(); + private final AtomicLong processedCount = new AtomicLong(); + private final AtomicReference<Long> lastTrxId = new AtomicReference<>(); + private final AtomicReference<Date> lastTrxDate = new AtomicReference<>(); + + private final ThreadLocal<SimpleDateFormat> dateFormatter = ThreadLocal.withInitial(() -> new SimpleDateFormat("yyyy/MM/dd HH:mm:ss Z")); + + public void incrMigratedCount(Long trxId, Date trxDate) { + migratedCount.incrementAndGet(); + lastTrxId.set(trxId); + lastTrxDate.set(trxDate); + + incrProcessedCount(); + } + + public void incrFailedCount() { + failedCount.incrementAndGet(); + + incrProcessedCount(); + } + + public void incrAlreadyMigratedCount() { + alreadyMigratedCount.incrementAndGet(); + + incrProcessedCount(); + } + + private void incrProcessedCount() { + if (processedCount.incrementAndGet() % 1000 == 0) { + logStats(); + } + } + + public void logStats() { + logger.info("PROGRESS: {} of {} transactions processed. Last migrated transaction(id={}, time={}). Counts(migrated: {}, failed: {}, already-migrated: {})", + processedCount.get(), totalCount, lastTrxId.get(), toString(lastTrxDate.get()), migratedCount.get(), failedCount.get(), alreadyMigratedCount.get()); + } + + private String toString(Date date) { + return date != null ? dateFormatter.get().format(date) : null; + } + } +} diff --git a/security-admin/src/main/resources/META-INF/jpa_named_queries.xml b/security-admin/src/main/resources/META-INF/jpa_named_queries.xml index 210c7952d..616ce9924 100755 --- a/security-admin/src/main/resources/META-INF/jpa_named_queries.xml +++ b/security-admin/src/main/resources/META-INF/jpa_named_queries.xml @@ -334,6 +334,32 @@ </query> </named-query> + <named-query name="XXTrxLog.findByTrxIdForMigration"> + <query>SELECT obj.id, obj.createTime, obj.addedByUserId, + obj.objectClassType, obj.objectId, obj.objectName, + obj.parentObjectId, obj.parentObjectClassType, obj.parentObjectName, + obj.attributeName, obj.previousValue, obj.newValue, + obj.transactionId, obj.action, obj.requestId, obj.sessionId, obj.sessionType + FROM XXTrxLog obj + WHERE obj.transactionId = :transactionId + ORDER BY obj.id + </query> + </named-query> + + <named-query name="XXTrxLog.findDistinctTrxIds"> + <query>SELECT obj.transactionId FROM XXTrxLog obj GROUP BY obj.transactionId ORDER BY MIN(obj.id) + </query> + </named-query> + + <named-query name="XXTrxLog.findDistinctTrxIdsByTimeInterval"> + <query> + SELECT obj.transactionId FROM XXTrxLog obj + WHERE obj.createTime BETWEEN :startDate AND :endDate + GROUP BY obj.transactionId + ORDER BY MIN(obj.id) + </query> + </named-query> + <named-query name="XXTrxLogV2.findByTrxId"> <query>SELECT obj FROM XXTrxLogV2 obj WHERE obj.transactionId = :transactionId order by obj.id</query> </named-query>
