diff --git a/.github/workflows/read-replica-integration-tests.yml b/.github/workflows/read-replica-integration-tests.yml new file mode 100644 index 000000000000..3921a8712e30 --- /dev/null +++ b/.github/workflows/read-replica-integration-tests.yml @@ -0,0 +1,120 @@ +name: Read-Replica Integration Tests + +on: + pull_request: + types: [labeled, synchronize, opened, reopened, ready_for_review] + +# Cancel running tests for previous commits on the same PR to save time/resources +concurrency: + group: read-replica-tests-${{ github.event.pull_request.number }} + cancel-in-progress: true + +jobs: + read-replica-integration-test: + # Trigger only if the 'read-replica' label is attached and the PR is not a draft + if: ${{ contains(github.event.pull_request.labels.*.name, 'read-replica') && !github.event.pull_request.draft }} + env: + REPLICA_BASE_DIR: ./dev-support/integration-test/read-replica + defaults: + run: + # Change default working directory from GITHUB_WORKSPACE to this + working-directory: ${{ env.REPLICA_BASE_DIR }} + runs-on: ubuntu-latest + steps: + - name: Checkout HBase Repository with Pull Request's SHA + uses: actions/checkout@v4 + + - name: Checkout HBase Respoitory Again Inside read-replica Directory + uses: actions/checkout@v4 + with: + # Place repo inside dev-support/integration-test/read-replica + # Use this repo to build the hbase-docker image + path: ${{ env.REPLICA_BASE_DIR }}/hbase + + - name: Show Working Directory + run: | + echo "The previous default working directory was: GITHUB_WORKSPACE=${GITHUB_WORKSPACE}" + echo "The current and default working directory is now: $(pwd)" + echo "The working directory now has an HBase repo:" + ls -la + echo "Showing the most recent commit in this HBase repo:" + cd hbase && git log --oneline | head + + - name: Load .env File into GitHub Actions Environment + run: | + # Turn on auto-export and source the file + set -a + source .env + set +a + + # Filter out comments, grab the keys, and write the expanded values to GITHUB_ENV + grep -E "^[A-Za-z0-9_]+=" .env | cut -d= -f1 | while read -r key; do + echo "$key=${!key}" + echo "$key=${!key}" >> $GITHUB_ENV + done + + - name: Set up Docker Compose + uses: docker/setup-compose-action@v2 + + - name: Install Python Requirements + run: | + pip install -r requirements.txt + + - name: Set Python Path + run: | + echo "Setting PYTHONPATH=$(pwd) for GITHUB_ENV=${GITHUB_ENV}" + echo "PYTHONPATH=$(pwd)" >> ${GITHUB_ENV} + + - name: Compile Protobuf + run: | + echo "Copying ActiveClusterSuffix.proto from HBase repo to have latest version" + cp hbase/hbase-protocol-shaded/src/main/protobuf/server/ActiveClusterSuffix.proto \ + python/proto + python3 python/proto/proto_compiler.py + + - name: Build Docker Image + run: | + echo "Building docker image" + ./build-images.sh + echo "Listing Docker images" + docker image ls + + - name: Verify Can't Start Two Active Clusters at Once + run: | + # This script automatically prepares the HBASE_DATA_STORE_ROOT defined in .env. + # This directory and its sub-directories need 777 permissions to prevent HBase + # from failing on startup. The directories are deleted when the containers are stopped + python3 python/scripts/test_dual_active_cluster_startup.py --clean-up-containers + + - name: Properly Start HBase Docker Containers + run: | + # This script also prepares HBASE_DATA_STORE_ROOT, similar to the previous step + echo "Starting one active and one replica HBase cluster and waiting for them to initialize..." + python3 python/scripts/verify_hbase_start.py + + - name: Test Table Create/Drop Behavior + run: python3 python/scripts/test_create_drop_behavior.py --skip-table-cleanup-on-start + + - name: Test Put/Get/Delete Behavior + run: python3 python/scripts/test_put_get_delete_behavior.py --skip-table-cleanup-on-start + + - name: Test Read-Only Flag Flipping + run: | + python3 python/scripts/test_read_only_flag_flipping.py --skip-container-start-or-restart + + - name: Test Cannot Promote Replica to Second Active Cluster + run: | + python3 python/scripts/test_cannot_promote_second_active_cluster.py --skip-container-start-or-restart + + - name: Test Bulkloaded Data and Region Splitting + run: | + python3 python/scripts/test_bulkloaded_data_and_region_splits.py --skip-container-start-or-restart + + - name: Clean up + run: docker compose -f docker-compose.yml down + + - name: Dump Logs on Failure + if: failure() + run: | + docker logs hbase-docker + docker logs hbase-docker-2 diff --git a/dev-support/integration-test/read-replica/.env b/dev-support/integration-test/read-replica/.env new file mode 100644 index 000000000000..904827f1652e --- /dev/null +++ b/dev-support/integration-test/read-replica/.env @@ -0,0 +1,25 @@ +# The name of the HBase Docker image +HBASE_IMAGE=kgeisz/hbase-docker:read-replica-gha +# The name of the HBase docker container +HBASE_CONTAINER_NAME=hbase-docker +# The directory within the docker container that contains the config files +HBASE_CONF_DIR=/opt/hbase/conf +# The directory containing the 'data-store/' directory. +# This directory is mounted with the HBase Docker container. +HBASE_DATA_STORE_ROOT=${RUNNER_TEMP}/tmp +# The port for the active cluster's HBase UI (used for cluster readiness) +ACTIVE_CLUSTER_PORT=16010 +# The port for the replica cluster's HBase UI (used for cluster readiness) +REPLICA_CLUSTER_PORT=26010 +# Local path to the active cluster's HBase config file. +# This file is part of a mounted volume, so modifying it +# locally also modifies it within the container (and vise-versa). +ACTIVE_CLUSTER_CONF_DIR=${GITHUB_WORKSPACE}/dev-support/integration-test/read-replica/conf1 +# Same as above, except for the replica cluster +REPLICA_CLUSTER_CONF_DIR=${GITHUB_WORKSPACE}/dev-support/integration-test/read-replica/conf2 +# The path to the docker-compose file +DOCKER_COMPOSE_FILE=${GITHUB_WORKSPACE}/dev-support/integration-test/read-replica/docker-compose.yml +# The location of the utils directory within the docker container +CONTAINER_UTILS_DIR=/opt/utils +# The log level for the Python scripts +LOG_LEVEL=DEBUG diff --git a/dev-support/integration-test/read-replica/Dockerfile b/dev-support/integration-test/read-replica/Dockerfile new file mode 100755 index 000000000000..b774e50c658e --- /dev/null +++ b/dev-support/integration-test/read-replica/Dockerfile @@ -0,0 +1,100 @@ +# Stage 0: Cache Maven dependencies +ARG BASE_IMAGE=registry.access.redhat.com/ubi8/openjdk-17 +FROM ${BASE_IMAGE} AS cache-stage + +USER root + +# Install necessary packages for building Maven dependencies +RUN INITRD=no DEBIAN_FRONTEND=noninteractive microdnf update -y && microdnf install -y maven git hostname diffutils + +# Copy the entire source code to cache dependencies +COPY ./hbase /opt/hbase-src + +WORKDIR /opt/hbase-src + +# Download and cache all dependencies +RUN mvn clean install -DskipTests -Dskip.license.check=true + +# Stage 1: Build the HBase source code +FROM ${BASE_IMAGE} AS build-stage + +USER root + +# Install necessary build packages +RUN INITRD=no DEBIAN_FRONTEND=noninteractive microdnf update -y && microdnf install -y maven git hostname diffutils + +# Copy the cached Maven dependencies +COPY --from=cache-stage /root/.m2 /root/.m2 + +# Copy the HBase source code +COPY ./hbase /opt/hbase-src + +WORKDIR /opt/hbase-src + +# Build HBase source code using cached dependencies and enable parallel build +RUN mvn clean package -DskipTests -Dskip.license.check=true assembly:single -T 1C + +# Stage 2: Create the final Docker image +FROM ${BASE_IMAGE} + +USER root + +# Set environment variables +ENV HBASE_HOME=/opt/hbase +ENV JAVA_HOME=/usr/lib/jvm/java-17-openjdk \ + HBASE_USER=hbase \ + HBASE_CONF_DIR=${HBASE_HOME}/conf \ + HBASE_LIB_DIR=${HBASE_HOME}/lib \ + HBASE_LOGS_DIR=${HBASE_HOME}/logs \ + DATA_DIR=/data-store + +# Install necessary runtime packages +RUN INITRD=no DEBIAN_FRONTEND=noninteractive microdnf update -y && microdnf install -y unzip gzip wget hostname maven git diffutils vim openssh-clients python3 procps + +# Copy the built HBase binaries from the build-stage +COPY --from=build-stage /opt/hbase-src/hbase-assembly/target/hbase-4.0.0-alpha-1-SNAPSHOT-bin.tar.gz /opt/ + +# Extract HBase binaries +RUN tar -xzf /opt/hbase-4.0.0-alpha-1-SNAPSHOT-bin.tar.gz -C /opt \ + && ln -s /opt/hbase-4.0.0-alpha-1-SNAPSHOT /opt/hbase \ + && rm /opt/hbase-4.0.0-alpha-1-SNAPSHOT-bin.tar.gz + +COPY --from=build-stage /root/.m2/repository/org/apache/hadoop/hadoop-mapreduce-client-common/*/hadoop-mapreduce-client-common-*.jar \ + ${HBASE_HOME}/lib/ + +# Apply custom HBase build steps +RUN sed -i "s,^. export JAVA_HOME.*,export JAVA_HOME=$JAVA_HOME," ${HBASE_CONF_DIR}/hbase-env.sh \ + && sed -E -i 's/(.*)hbase\-daemons\.sh(.*zookeeper)/\1hbase-daemon.sh\2/g' ${HBASE_HOME}/bin/start-hbase.sh \ + && echo -e "JAVA_HOME=$JAVA_HOME\nexport JAVA_HOME\nexport PATH=$JAVA_HOME/jre/bin:$PATH" > /etc/profile.d/defaults.sh \ + && ln -sf ${HBASE_HOME}/bin/* /usr/bin + +# Create HBase user and home directory +RUN useradd -u 1000 -m ${HBASE_USER} \ + && mkdir -p /home/${HBASE_USER} \ + && chown -R ${HBASE_USER}:${HBASE_USER} /opt /home/${HBASE_USER} \ + && mkdir "$DATA_DIR" && chown -R ${HBASE_USER}:${HBASE_USER} "$DATA_DIR" + +# Set permissions for jboss home directory +RUN chown -R ${HBASE_USER}:${HBASE_USER} /home/jboss + +# Copy configuration files - Removed in order to mount config files individually # + +# Expose required ports for HBase and related services +EXPOSE 8000 8080 8085 9090 9095 2181 16000 16010 16020 16030 + +# Set up the utils directory as a volume +VOLUME ["/opt/utils"] + +# Switch to the HBase user +USER ${HBASE_USER} + +# Create necessary directories for HBase to run +RUN mkdir -p "$DATA_DIR"/hbase "$DATA_DIR"/run "$DATA_DIR"/logs +RUN chmod -R 777 "$DATA_DIR" + +# Add the 'ls -l' alias +RUN echo 'alias ll="ls -l"' >> /home/hbase/.bashrc +RUN echo 'alias ll="ls -l"' >> /home/jboss/.bashrc + +# Start HBase and keep it running +ENTRYPOINT ["/bin/bash", "-c", "${HBASE_HOME}/bin/start-hbase.sh && tail -f ${HBASE_LOGS_DIR}/*.log"] diff --git a/dev-support/integration-test/read-replica/build-images.sh b/dev-support/integration-test/read-replica/build-images.sh new file mode 100755 index 000000000000..9d2884aeccf3 --- /dev/null +++ b/dev-support/integration-test/read-replica/build-images.sh @@ -0,0 +1,47 @@ +#!/bin/bash + +# Load environment variables from .env file +set -a +. ./.env +set +a + +# Check if required environment variables are set +if [ -z "$HBASE_IMAGE" ]; then + echo "Error: HBASE_IMAGE is not set in .env file." + exit 1 +fi + +# HBase source directory +HBASE_SOURCE_DIR="./hbase" + +# Check if HBase source directory exists +if [ ! -d "$HBASE_SOURCE_DIR" ]; then + echo "Error: HBase source directory does not exist at $HBASE_SOURCE_DIR." + exit 1 +fi + +# Run Maven clean to remove previous build artifacts +echo "Running 'mvn clean' in $HBASE_SOURCE_DIR..." +cd "$HBASE_SOURCE_DIR" || exit 1 +mvn clean + +if [ $? -ne 0 ]; then + echo "Error: 'mvn clean' failed." + exit 1 +else + echo "'mvn clean' completed successfully." +fi + +# Return to the original directory +cd - || exit 1 + +# Build HBase Docker image +echo "Building HBase Docker image: ${HBASE_IMAGE}" +docker build -t "${HBASE_IMAGE}" ./ + +if [ $? -ne 0 ]; then + echo "Error: Failed to build HBase Docker image." + exit 1 +else + echo "HBase Docker image built successfully." +fi diff --git a/dev-support/integration-test/read-replica/conf1/hbase-site.xml b/dev-support/integration-test/read-replica/conf1/hbase-site.xml new file mode 100755 index 000000000000..9c13265a4e81 --- /dev/null +++ b/dev-support/integration-test/read-replica/conf1/hbase-site.xml @@ -0,0 +1,54 @@ + + + + + hbase.zookeeper.quorum + hbase-docker + + + hbase.zookeeper.property.dataDir + /data-store/zk + + + hbase.rootdir + file:///data-store/hbase + + + hbase.wal.dir + file:///data-store/wal + + + hbase.store.file-tracker.impl + FILE + + + + hbase.master.info.bindAddress + hbase-docker + + + hbase.regionserver.info.bindAddress + hbase-docker + + + hbase.unsafe.stream.capability.enforce + false + + + hbase.cluster.distributed + true + + + hbase.rest.port + 8000 + + + + hbase.global.readonly.enabled + false + + + hbase.master.hbck.chore.interval + 0 + + diff --git a/dev-support/integration-test/read-replica/conf1/log4j2.properties b/dev-support/integration-test/read-replica/conf1/log4j2.properties new file mode 100644 index 000000000000..91957fdc8a31 --- /dev/null +++ b/dev-support/integration-test/read-replica/conf1/log4j2.properties @@ -0,0 +1,137 @@ +#/** +# * 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. +# */ + +status = warn +dest = err +name = PropertiesConfig + +# Console appender +appender.console.type = Console +appender.console.target = SYSTEM_ERR +appender.console.name = console +appender.console.layout.type = PatternLayout +appender.console.layout.pattern = %d{ISO8601} %-5p [%t%notEmpty{ %X}] %c{2}: %.1000m%n + +# Daily Rolling File Appender +appender.DRFA.type = RollingFile +appender.DRFA.name = DRFA +appender.DRFA.fileName = ${sys:hbase.log.dir:-.}/${sys:hbase.log.file:-hbase.log} +appender.DRFA.filePattern = ${sys:hbase.log.dir:-.}/${sys:hbase.log.file:-hbase.log}.%d{yyyy-MM-dd} +appender.DRFA.createOnDemand = true +appender.DRFA.layout.type = PatternLayout +appender.DRFA.layout.pattern = %d{ISO8601} %-5p [%t] %c{2}: %.1000m%n +appender.DRFA.policies.type = Policies +appender.DRFA.policies.time.type = TimeBasedTriggeringPolicy +appender.DRFA.policies.time.interval = 1 +appender.DRFA.policies.time.modulate = true +appender.DRFA.policies.size.type = SizeBasedTriggeringPolicy +appender.DRFA.policies.size.size = ${sys:hbase.log.maxfilesize:-256MB} +appender.DRFA.strategy.type = DefaultRolloverStrategy +appender.DRFA.strategy.max = ${sys:hbase.log.maxbackupindex:-20} + +# Rolling File Appender +appender.RFA.type = RollingFile +appender.RFA.name = RFA +appender.RFA.fileName = ${sys:hbase.log.dir:-.}/${sys:hbase.log.file:-hbase.log} +appender.RFA.filePattern = ${sys:hbase.log.dir:-.}/${sys:hbase.log.file:-hbase.log}.%i +appender.RFA.createOnDemand = true +appender.RFA.layout.type = PatternLayout +appender.RFA.layout.pattern = %d{ISO8601} %-5p [%t] %c{2}: %.1000m%n +appender.RFA.policies.type = Policies +appender.RFA.policies.size.type = SizeBasedTriggeringPolicy +appender.RFA.policies.size.size = ${sys:hbase.log.maxfilesize:-256MB} +appender.RFA.strategy.type = DefaultRolloverStrategy +appender.RFA.strategy.max = ${sys:hbase.log.maxbackupindex:-20} + +# Security Audit Appender +appender.RFAS.type = RollingFile +appender.RFAS.name = RFAS +appender.RFAS.fileName = ${sys:hbase.log.dir:-.}/${sys:hbase.security.log.file:-SecurityAuth.audit} +appender.RFAS.filePattern = ${sys:hbase.log.dir:-.}/${sys:hbase.security.log.file:-SecurityAuth.audit}.%i +appender.RFAS.createOnDemand = true +appender.RFAS.layout.type = PatternLayout +appender.RFAS.layout.pattern = %d{ISO8601} %-5p [%t] %c{2}: %.1000m%n +appender.RFAS.policies.type = Policies +appender.RFAS.policies.size.type = SizeBasedTriggeringPolicy +appender.RFAS.policies.size.size = ${sys:hbase.security.log.maxfilesize:-256MB} +appender.RFAS.strategy.type = DefaultRolloverStrategy +appender.RFAS.strategy.max = ${sys:hbase.security.log.maxbackupindex:-20} + +# Http Access Log RFA, uncomment this if you want an http access.log +# appender.AccessRFA.type = RollingFile +# appender.AccessRFA.name = AccessRFA +# appender.AccessRFA.fileName = /var/log/hbase/access.log +# appender.AccessRFA.filePattern = /var/log/hbase/access.log.%i +# appender.AccessRFA.createOnDemand = true +# appender.AccessRFA.layout.type = PatternLayout +# appender.AccessRFA.layout.pattern = %m%n +# appender.AccessRFA.policies.type = Policies +# appender.AccessRFA.policies.size.type = SizeBasedTriggeringPolicy +# appender.AccessRFA.policies.size.size = 200MB +# appender.AccessRFA.strategy.type = DefaultRolloverStrategy +# appender.AccessRFA.strategy.max = 10 + +# Null Appender +appender.NullAppender.type = Null +appender.NullAppender.name = NullAppender + +rootLogger = ${sys:hbase.root.logger:-INFO,console} + +logger.SecurityLogger.name = SecurityLogger +logger.SecurityLogger = ${sys:hbase.security.logger:-INFO,console} +logger.SecurityLogger.additivity = false + +# Custom Logging levels +# logger.zookeeper.name = org.apache.zookeeper +# logger.zookeeper.level = ERROR + +# logger.FSNamesystem.name = org.apache.hadoop.fs.FSNamesystem +# logger.FSNamesystem.level = DEBUG + +logger.hbase.name = org.apache.hadoop.hbase +logger.hbase.level = DEBUG + +# logger.META.name = org.apache.hadoop.hbase.META +# logger.META.level = DEBUG + +# Make these two classes below DEBUG to see more zk debug. +# logger.ZKUtil.name = org.apache.hadoop.hbase.zookeeper.ZKUtil +# logger.ZKUtil.level = DEBUG + +# logger.ZKWatcher.name = org.apache.hadoop.hbase.zookeeper.ZKWatcher +# logger.ZKWatcher.level = DEBUG + +# logger.dfs.name = org.apache.hadoop.dfs +# logger.dfs.level = DEBUG + +# Prevent metrics subsystem start/stop messages (HBASE-17722) +logger.MetricsConfig.name = org.apache.hadoop.metrics2.impl.MetricsConfig +logger.MetricsConfig.level = WARN + +logger.MetricsSinkAdapte.name = org.apache.hadoop.metrics2.impl.MetricsSinkAdapter +logger.MetricsSinkAdapte.level = WARN + +logger.MetricsSystemImpl.name = org.apache.hadoop.metrics2.impl.MetricsSystemImpl +logger.MetricsSystemImpl.level = WARN + +# Disable request log by default, you can enable this by changing the appender +logger.http.name = http.requests +logger.http.additivity = false +logger.http = INFO,NullAppender +# Replace the above with this configuration if you want an http access.log +# logger.http = INFO,AccessRFA diff --git a/dev-support/integration-test/read-replica/conf1/zoo.cfg b/dev-support/integration-test/read-replica/conf1/zoo.cfg new file mode 100755 index 000000000000..5677f0fbfa7c --- /dev/null +++ b/dev-support/integration-test/read-replica/conf1/zoo.cfg @@ -0,0 +1,3 @@ +clientPort=2181 +clientPortAddress=hbase-docker +server.1=hbase-docker:2181 diff --git a/dev-support/integration-test/read-replica/conf2/hbase-site.xml b/dev-support/integration-test/read-replica/conf2/hbase-site.xml new file mode 100755 index 000000000000..8481306089d1 --- /dev/null +++ b/dev-support/integration-test/read-replica/conf2/hbase-site.xml @@ -0,0 +1,58 @@ + + + + + hbase.zookeeper.quorum + hbase-docker-2 + + + hbase.zookeeper.property.dataDir + /data-store/zk + + + hbase.rootdir + file:///data-store/hbase + + + hbase.wal.dir + file:///data-store/wal + + + hbase.store.file-tracker.impl + FILE + + + + hbase.master.info.bindAddress + hbase-docker-2 + + + hbase.regionserver.info.bindAddress + hbase-docker-2 + + + hbase.unsafe.stream.capability.enforce + false + + + hbase.cluster.distributed + true + + + hbase.rest.port + 8000 + + + + hbase.meta.table.suffix + replica1 + + + hbase.global.readonly.enabled + true + + + hbase.master.hbck.chore.interval + 0 + + diff --git a/dev-support/integration-test/read-replica/conf2/log4j2.properties b/dev-support/integration-test/read-replica/conf2/log4j2.properties new file mode 100644 index 000000000000..91957fdc8a31 --- /dev/null +++ b/dev-support/integration-test/read-replica/conf2/log4j2.properties @@ -0,0 +1,137 @@ +#/** +# * 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. +# */ + +status = warn +dest = err +name = PropertiesConfig + +# Console appender +appender.console.type = Console +appender.console.target = SYSTEM_ERR +appender.console.name = console +appender.console.layout.type = PatternLayout +appender.console.layout.pattern = %d{ISO8601} %-5p [%t%notEmpty{ %X}] %c{2}: %.1000m%n + +# Daily Rolling File Appender +appender.DRFA.type = RollingFile +appender.DRFA.name = DRFA +appender.DRFA.fileName = ${sys:hbase.log.dir:-.}/${sys:hbase.log.file:-hbase.log} +appender.DRFA.filePattern = ${sys:hbase.log.dir:-.}/${sys:hbase.log.file:-hbase.log}.%d{yyyy-MM-dd} +appender.DRFA.createOnDemand = true +appender.DRFA.layout.type = PatternLayout +appender.DRFA.layout.pattern = %d{ISO8601} %-5p [%t] %c{2}: %.1000m%n +appender.DRFA.policies.type = Policies +appender.DRFA.policies.time.type = TimeBasedTriggeringPolicy +appender.DRFA.policies.time.interval = 1 +appender.DRFA.policies.time.modulate = true +appender.DRFA.policies.size.type = SizeBasedTriggeringPolicy +appender.DRFA.policies.size.size = ${sys:hbase.log.maxfilesize:-256MB} +appender.DRFA.strategy.type = DefaultRolloverStrategy +appender.DRFA.strategy.max = ${sys:hbase.log.maxbackupindex:-20} + +# Rolling File Appender +appender.RFA.type = RollingFile +appender.RFA.name = RFA +appender.RFA.fileName = ${sys:hbase.log.dir:-.}/${sys:hbase.log.file:-hbase.log} +appender.RFA.filePattern = ${sys:hbase.log.dir:-.}/${sys:hbase.log.file:-hbase.log}.%i +appender.RFA.createOnDemand = true +appender.RFA.layout.type = PatternLayout +appender.RFA.layout.pattern = %d{ISO8601} %-5p [%t] %c{2}: %.1000m%n +appender.RFA.policies.type = Policies +appender.RFA.policies.size.type = SizeBasedTriggeringPolicy +appender.RFA.policies.size.size = ${sys:hbase.log.maxfilesize:-256MB} +appender.RFA.strategy.type = DefaultRolloverStrategy +appender.RFA.strategy.max = ${sys:hbase.log.maxbackupindex:-20} + +# Security Audit Appender +appender.RFAS.type = RollingFile +appender.RFAS.name = RFAS +appender.RFAS.fileName = ${sys:hbase.log.dir:-.}/${sys:hbase.security.log.file:-SecurityAuth.audit} +appender.RFAS.filePattern = ${sys:hbase.log.dir:-.}/${sys:hbase.security.log.file:-SecurityAuth.audit}.%i +appender.RFAS.createOnDemand = true +appender.RFAS.layout.type = PatternLayout +appender.RFAS.layout.pattern = %d{ISO8601} %-5p [%t] %c{2}: %.1000m%n +appender.RFAS.policies.type = Policies +appender.RFAS.policies.size.type = SizeBasedTriggeringPolicy +appender.RFAS.policies.size.size = ${sys:hbase.security.log.maxfilesize:-256MB} +appender.RFAS.strategy.type = DefaultRolloverStrategy +appender.RFAS.strategy.max = ${sys:hbase.security.log.maxbackupindex:-20} + +# Http Access Log RFA, uncomment this if you want an http access.log +# appender.AccessRFA.type = RollingFile +# appender.AccessRFA.name = AccessRFA +# appender.AccessRFA.fileName = /var/log/hbase/access.log +# appender.AccessRFA.filePattern = /var/log/hbase/access.log.%i +# appender.AccessRFA.createOnDemand = true +# appender.AccessRFA.layout.type = PatternLayout +# appender.AccessRFA.layout.pattern = %m%n +# appender.AccessRFA.policies.type = Policies +# appender.AccessRFA.policies.size.type = SizeBasedTriggeringPolicy +# appender.AccessRFA.policies.size.size = 200MB +# appender.AccessRFA.strategy.type = DefaultRolloverStrategy +# appender.AccessRFA.strategy.max = 10 + +# Null Appender +appender.NullAppender.type = Null +appender.NullAppender.name = NullAppender + +rootLogger = ${sys:hbase.root.logger:-INFO,console} + +logger.SecurityLogger.name = SecurityLogger +logger.SecurityLogger = ${sys:hbase.security.logger:-INFO,console} +logger.SecurityLogger.additivity = false + +# Custom Logging levels +# logger.zookeeper.name = org.apache.zookeeper +# logger.zookeeper.level = ERROR + +# logger.FSNamesystem.name = org.apache.hadoop.fs.FSNamesystem +# logger.FSNamesystem.level = DEBUG + +logger.hbase.name = org.apache.hadoop.hbase +logger.hbase.level = DEBUG + +# logger.META.name = org.apache.hadoop.hbase.META +# logger.META.level = DEBUG + +# Make these two classes below DEBUG to see more zk debug. +# logger.ZKUtil.name = org.apache.hadoop.hbase.zookeeper.ZKUtil +# logger.ZKUtil.level = DEBUG + +# logger.ZKWatcher.name = org.apache.hadoop.hbase.zookeeper.ZKWatcher +# logger.ZKWatcher.level = DEBUG + +# logger.dfs.name = org.apache.hadoop.dfs +# logger.dfs.level = DEBUG + +# Prevent metrics subsystem start/stop messages (HBASE-17722) +logger.MetricsConfig.name = org.apache.hadoop.metrics2.impl.MetricsConfig +logger.MetricsConfig.level = WARN + +logger.MetricsSinkAdapte.name = org.apache.hadoop.metrics2.impl.MetricsSinkAdapter +logger.MetricsSinkAdapte.level = WARN + +logger.MetricsSystemImpl.name = org.apache.hadoop.metrics2.impl.MetricsSystemImpl +logger.MetricsSystemImpl.level = WARN + +# Disable request log by default, you can enable this by changing the appender +logger.http.name = http.requests +logger.http.additivity = false +logger.http = INFO,NullAppender +# Replace the above with this configuration if you want an http access.log +# logger.http = INFO,AccessRFA diff --git a/dev-support/integration-test/read-replica/conf2/zoo.cfg b/dev-support/integration-test/read-replica/conf2/zoo.cfg new file mode 100755 index 000000000000..2071752be5b2 --- /dev/null +++ b/dev-support/integration-test/read-replica/conf2/zoo.cfg @@ -0,0 +1,3 @@ +clientPort=2181 +clientPortAddress=hbase-docker-2 +server.1=hbase-docker-2:2181 diff --git a/dev-support/integration-test/read-replica/docker-compose.yml b/dev-support/integration-test/read-replica/docker-compose.yml new file mode 100644 index 000000000000..a9961b64569a --- /dev/null +++ b/dev-support/integration-test/read-replica/docker-compose.yml @@ -0,0 +1,29 @@ +# version: '3.8' + +# Variables are created in .env file +services: + hbase: + image: ${HBASE_IMAGE} + container_name: ${HBASE_CONTAINER_NAME} + hostname: ${HBASE_CONTAINER_NAME} + ports: + - ${ACTIVE_CLUSTER_PORT}:16010 # Master UI + volumes: + - ./utils:${CONTAINER_UTILS_DIR} + - ${HBASE_DATA_STORE_ROOT}/data-store/hbase:/data-store/hbase + - ${ACTIVE_CLUSTER_CONF_DIR}/hbase-site.xml:${HBASE_CONF_DIR}/hbase-site.xml + - ${ACTIVE_CLUSTER_CONF_DIR}/zoo.cfg:${HBASE_CONF_DIR}/zoo.cfg + - ${ACTIVE_CLUSTER_CONF_DIR}/log4j2.properties:${HBASE_CONF_DIR}/log4j2.properties + + hbase2: + image: ${HBASE_IMAGE} + container_name: ${HBASE_CONTAINER_NAME}-2 + hostname: ${HBASE_CONTAINER_NAME}-2 + ports: + - ${REPLICA_CLUSTER_PORT}:16010 # Master UI + volumes: + - ./utils:${CONTAINER_UTILS_DIR} + - ${HBASE_DATA_STORE_ROOT}/data-store/hbase:/data-store/hbase + - ${REPLICA_CLUSTER_CONF_DIR}/hbase-site.xml:${HBASE_CONF_DIR}/hbase-site.xml + - ${REPLICA_CLUSTER_CONF_DIR}/zoo.cfg:${HBASE_CONF_DIR}/zoo.cfg + - ${REPLICA_CLUSTER_CONF_DIR}/log4j2.properties:${HBASE_CONF_DIR}/log4j2.properties diff --git a/dev-support/integration-test/read-replica/python/__init__.py b/dev-support/integration-test/read-replica/python/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/dev-support/integration-test/read-replica/python/proto/proto_compiler.py b/dev-support/integration-test/read-replica/python/proto/proto_compiler.py new file mode 100644 index 000000000000..a657e52f606b --- /dev/null +++ b/dev-support/integration-test/read-replica/python/proto/proto_compiler.py @@ -0,0 +1,30 @@ +#!/usr/bin/env python3 +""" +Compiles all *.proto files in the 'python/proto' directory. The compiled output is sent to 'python/proto/generated'. +This script removes any existing 'generated' directory and creates a new one every time. +""" +import os +from grpc_tools import protoc +from python.src.logger_config import get_logger + +logger = get_logger(__name__) + + +if __name__ == '__main__': + proto_dir = os.path.dirname(__file__) + generated_dir = os.path.join(proto_dir, 'generated') + + if os.path.exists(generated_dir): + os.rmdir(generated_dir) + os.mkdir(generated_dir) + + proto_files = [file for file in os.listdir(proto_dir) if file.endswith('.proto')] + for file in proto_files: + logger.info(f"Compiling {file} and sending output to {generated_dir}") + protoc.main(( + '', + f'-I{proto_dir}', + f'--python_out={generated_dir}/.', + f'--pyi_out={generated_dir}/.', + os.path.join(proto_dir, file), + )) diff --git a/dev-support/integration-test/read-replica/python/scripts/__init__.py b/dev-support/integration-test/read-replica/python/scripts/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/dev-support/integration-test/read-replica/python/scripts/test_bulkloaded_data_and_region_splits.py b/dev-support/integration-test/read-replica/python/scripts/test_bulkloaded_data_and_region_splits.py new file mode 100644 index 000000000000..465cd496fbaa --- /dev/null +++ b/dev-support/integration-test/read-replica/python/scripts/test_bulkloaded_data_and_region_splits.py @@ -0,0 +1,227 @@ +#!/usr/bin/env python3 +""" +This script tests bulk-loading data with Read-Replica HBase clusters. +""" +import argparse +import time + +from python.src import get_logger, HBaseDockerClient +from python.src.environment_loader import get_env +from python.src.hbase_docker_client import DockerExecCommandError +from python.src.utils import (add_common_skip_container_stop_or_restart_arg, reset_cluster_setup, + load_env_and_set_up_clients, swap_cluster_roles, + log_script_start, log_script_end) + +logger = get_logger(__name__) + + +class Bulkloader: + def __init__(self, bulkload_script: str): + self.bulkload_script = bulkload_script + + def bulkload_data(self, active_cluster: HBaseDockerClient, table_name: str, column_family: str = 'cf', + num_rows: int = 500, initial_row_num: int = 0): + logger.info(f"Running {self.bulkload_script} to bulkload {num_rows} rows into table '{table_name}' " + f"on {active_cluster.name}, starting with row {initial_row_num}") + active_cluster.run_docker_exec_command( + f"{self.bulkload_script} {table_name} {column_family} -n {num_rows} -i {initial_row_num}") + + +def assert_cannot_bulkload_data_onto_replica(bulkloader: Bulkloader, replica_cluster: HBaseDockerClient): + logger.info(f"Verifying data cannot be loaded onto {replica_cluster.name} because read-only mode is enabled") + try: + bulkloader.bulkload_data(replica_cluster, table_name='replica-blt1', column_family='cf') + raise RuntimeError(f"Expected bulkloading data onto replica cluster {replica_cluster.name} " + f"to result in an error") + except DockerExecCommandError as e: + expected_error_msg = ("org.apache.hadoop.hbase.WriteAttemptedOnReadOnlyClusterException: " + "Operation not allowed in Read-Only Mode") + assert expected_error_msg in str(e), (f"Expected exception to contain the following error message after " + f"attempting to bulkload data on to a replica cluster:\n" + f"{expected_error_msg}\n" + f"The actual exception was:\n{e}") + logger.info(f"Bulkload onto replica cluster {replica_cluster.name} failed as expected") + + +def assert_cannot_split_regions_on_replica(replica_cluster: HBaseDockerClient, table: str): + logger.info(f"Verifying regions cannot be split on {replica_cluster.name} because read-only mode is enabled") + try: + replica_cluster.split(table) + raise RuntimeError(f"Expected region split on replica cluster {replica_cluster.name} to result in an error") + except DockerExecCommandError as e: + expected_error_msg = ("org.apache.hadoop.hbase.WriteAttemptedOnReadOnlyClusterException: " + "Operation not allowed in Read-Only Mode") + assert expected_error_msg in str(e), (f"Expected exception to contain the following error message after " + f"attempting to split a region on a replica cluster:\n" + f"{expected_error_msg}\n" + f"The actual exception was:\n{e}") + logger.info(f"Region splitting on replica cluster {replica_cluster.name} failed as expected") + + +def main(): + log_script_start(__file__, logger) + + parser = argparse.ArgumentParser() + parser = add_common_skip_container_stop_or_restart_arg(parser) + args = parser.parse_args() + + skip_container_restart = args.skip_container_start_or_restart + + if skip_container_restart: + logger.info("Docker containers will NOT be started/restarted at the beginning of this test run") + else: + logger.info("Docker containers will be started/restarted at the beginning of this test run") + + cluster1, cluster2 = load_env_and_set_up_clients() + + data_store_root = get_env("HBASE_DATA_STORE_ROOT") + docker_compose_file = get_env("DOCKER_COMPOSE_FILE") + container_utils_dir = get_env("CONTAINER_UTILS_DIR") + + table1 = 'blt1' + table2 = 'blt2' + table3 = 'blt3' + tables = [table1, table2, table3] + + bulkloader = Bulkloader(bulkload_script=f"{container_utils_dir}/bulkload.sh") + + reset_cluster_setup(active_cluster=cluster1, replica_cluster=cluster2, + skip_container_restart=skip_container_restart, docker_compose_file=docker_compose_file, + data_store_root=data_store_root, sudo=True) + + logger.info(f"The active cluster is {cluster1.name} and the replica cluster is {cluster2.name}") + + assert_cannot_bulkload_data_onto_replica(bulkloader, replica_cluster=cluster2) + + # Bulkload data to active cluster and verify the data is there + logger.info(f"Bulkloading data to '{table1}' on the active cluster and verifying the data is there") + bulkloader.bulkload_data(active_cluster=cluster1, table_name=table1) + cluster1.assert_table_exists(table1) + cluster1.assert_table_row_count(table1, expected_row_count=500) + + # Replica cluster should not see bulkloaded data until meta and HFiles have been refreshed + logger.info(f"The replica cluster {cluster2.name} should not see bulkloaded data until " + f"meta and HFiles have been refreshed") + cluster2.assert_table_does_not_exist(table1) + cluster2.refresh_meta_and_hfiles() + cluster2.assert_table_exists(table1) + cluster2.assert_table_row_count(table1, expected_row_count=500) + + # Cluster 1 is now a replica and Cluster 2 is now the active cluster + swap_cluster_roles(new_active_cluster=cluster2, new_replica_cluster=cluster1) + + # Bulkload more data into the existing table on Cluster 2 + logger.info(f"Bulkloading more data into the existing table on {cluster2.name}") + bulkloader.bulkload_data(active_cluster=cluster2, table_name=table1, num_rows=300, initial_row_num=500) + cluster2.assert_table_row_count(table1, expected_row_count=800) + + # Cluster 1 should not see the newly bulkloaded data until its meta and HFiles have been refreshed + logger.info(f"The replica cluster {cluster1.name} should not see bulkloaded data until " + f"meta and HFiles have been refreshed") + cluster1.assert_table_row_count(table1, expected_row_count=500) + cluster1.refresh_meta_and_hfiles() + cluster1.assert_table_row_count(table1, expected_row_count=800) + + # Bulkload data into a new table on Cluster 2 + logger.info(f"Bulkloading data into new table '{table2}' on active cluster {cluster2.name}") + bulkloader.bulkload_data(active_cluster=cluster2, table_name=table2, num_rows=600) + cluster2.assert_table_exists(table2) + cluster2.assert_table_row_count(table2, expected_row_count=600) + + # Cluster 1 should not see this new table until after refreshing meta and HFiles + logger.info(f"The replica cluster {cluster1.name} should not see '{table2}' until after refreshing meta and HFiles") + cluster1.assert_table_does_not_exist(table2) + cluster1.refresh_meta_and_hfiles() + cluster1.assert_table_exists(table2) + cluster1.assert_table_row_count(table2, expected_row_count=600) + cluster1.assert_table_row_count(table1, expected_row_count=800) + + # Cluster 1 is back to being the active cluster and Cluster 2 is once again the replica cluster + swap_cluster_roles(new_active_cluster=cluster1, new_replica_cluster=cluster2) + + # Bulkload data onto both existing tables, and a new third table + logger.info(f"Bulkloading data onto '{table1}' and '{table2}', as well as a new table '{table3}'") + bulkloader.bulkload_data(active_cluster=cluster1, table_name=table1, num_rows=400, initial_row_num=800) + bulkloader.bulkload_data(active_cluster=cluster1, table_name=table2, num_rows=600, initial_row_num=600) + bulkloader.bulkload_data(active_cluster=cluster1, table_name=table3, num_rows=1200) + for table in tables: + cluster1.assert_table_row_count(table, expected_row_count=1200) + + # Cluster 2 should see the old row counts for the existing tables. It won't see the new table + # or the updated row counts until after its meta and HFiles have been refreshed. + logger.info(f"The replica cluster {cluster2.name} should not see '{table3}' or updated values for " + f"'{table1}' and '{table2}' until after refreshing meta and HFiles") + cluster2.assert_table_row_count(table1, expected_row_count=800) + cluster2.assert_table_row_count(table2, expected_row_count=600) + cluster2.assert_table_does_not_exist(table3) + cluster2.refresh_meta_and_hfiles() + for table in tables: + cluster2.assert_table_row_count(table, expected_row_count=1200) + + # Cluster 2 is now the active cluster and Cluster 1 is the replica cluster + swap_cluster_roles(new_active_cluster=cluster2, new_replica_cluster=cluster1) + + # Split regions on two tables on the active cluster + for table in [table1, table2]: + logger.info(f"Splitting table '{table}' on {cluster2.name}") + cluster2.flush_and_split(table) + cluster2.major_compact_and_wait(table) + cluster2.catalogjanitor_run() + time.sleep(5) + cluster2.assert_region_count_for_table(table, expected_region_count=2) + + # Bulkload more rows into each table on Cluster 2 and into a new table + table4 = 'blt4' + logger.info(f"Bulkloading data onto '{table1}', '{table2}', and '{table3}', as well as a new table '{table4}'") + for table in tables: + bulkloader.bulkload_data(active_cluster=cluster2, table_name=table, num_rows=1200, initial_row_num=1200) + bulkloader.bulkload_data(active_cluster=cluster2, table_name=table4, num_rows=2400) + tables.append(table4) + for table in tables: + cluster2.assert_table_row_count(table, expected_row_count=2400) + + # The replica cluster should see the old row counts for the existing tables. It won't see the new table + # or the updated row counts until after its meta and HFiles have been refreshed. + logger.info(f"The replica cluster {cluster1.name} should not see '{table4}' or row counts/region splits for " + f"'{table1}', '{table2}', and '{table3}' until after refreshing meta and HFiles") + for table in tables[:-1]: + cluster1.assert_table_row_count(table, expected_row_count=1200) + cluster1.assert_table_does_not_exist(table4) + + # The replica cluster should not see any region splits until after refreshing meta and HFiles + for table in tables[:-1]: + cluster1.assert_region_count_for_table(table, expected_region_count=1) + + # The replica cluster will now see updated row counts and region splits + cluster1.refresh_meta_and_hfiles() + logger.info(f"Replica cluster {cluster1.name} should now see updated row counts and region splits") + for table, num_regions in zip(tables, [2, 2, 1, 1]): + cluster1.assert_table_row_count(table, expected_row_count=2400) + cluster1.assert_region_count_for_table(table, expected_region_count=num_regions) + + # Make Cluster 1 the active cluster and Cluster 2 the replica cluster + swap_cluster_roles(new_active_cluster=cluster1, new_replica_cluster=cluster2) + + assert_cannot_split_regions_on_replica(replica_cluster=cluster2, table=table3) + + # Split regions on the active cluster. The replica cluster won't see the updated region count until meta and HFiles + # have been refreshed + for table, num_regions in zip(tables, [4, 4, 2, 2]): + cluster1.flush_and_split(table) + cluster1.major_compact(table) + cluster1.catalogjanitor_run() + time.sleep(5) + cluster1.assert_region_count_for_table(table, num_regions) + + # Replica cluster still has old region count + cluster2.assert_region_count_for_table(table, num_regions/2) + + # Update the replica cluster and verify new region count + cluster2.refresh_meta_and_hfiles() + cluster2.assert_region_count_for_table(table, num_regions) + + log_script_end(__file__, logger) + + +if __name__ == '__main__': + main() diff --git a/dev-support/integration-test/read-replica/python/scripts/test_cannot_promote_second_active_cluster.py b/dev-support/integration-test/read-replica/python/scripts/test_cannot_promote_second_active_cluster.py new file mode 100644 index 000000000000..9ea92ebdee95 --- /dev/null +++ b/dev-support/integration-test/read-replica/python/scripts/test_cannot_promote_second_active_cluster.py @@ -0,0 +1,103 @@ +#!/usr/bin/env python3 +""" +Verifies a cluster cannot be promoted to an active cluster when another active cluster already exists. + +The test starts with two Read-Replica HBase clusters, where one cluster is the active cluster and the other cluster is +the replica cluster. The test tries to promote the replica cluster to a second active cluster and expects an error to +occur. It then verifies this "second active cluster" is still in read-only mode and that data can still be added to the +actual active cluster. + +This test script verifies the fix for: + +HBASE-30220: A replica cluster can have read-only mode disabled even when another active cluster already exists +https://issues.apache.org/jira/browse/HBASE-30220 + +Before implementing the fix for HBASE-30220, a cluster could be promoted to from a replica cluster to an active cluster +even when another active cluster already existed. +""" +import argparse + +from python.src.environment_loader import get_env +from python.src.hbase_docker_client import HBaseDockerClient, DockerExecCommandError +from python.src.logger_config import get_logger +from python.src.utils import (assert_crud_operations_work_on_active_cluster, assert_correct_active_cluster_suffix, + add_common_skip_container_stop_or_restart_arg, clean_up_tables, reset_cluster_setup, + load_env_and_set_up_clients, create_table_and_test_active_and_replica_clusters, + log_script_start, log_script_end) +from time import sleep + +logger = get_logger(__name__) + +COLUMN_FAMILY = "cf" +EXPECTED_ERROR_MSG = ("ReadOnlyTransitionException: Cannot disable read-only mode because another active cluster " + "already exists on this storage location. The read-only coprocessors have not been removed.") + + +def assert_error_when_trying_to_have_second_active_cluster(replica_cluster: HBaseDockerClient, expected_error: str): + try: + replica_cluster.disable_read_only_mode() + raise RuntimeError(f"Expected an DockerExecCommandError with the following error message:\n\n" + f"{expected_error}") + except DockerExecCommandError as e: + assert expected_error in str(e), (f"Expected DockerExecCommandError to contain the following message:\n\n" + f"{str(expected_error)}\n\n" + f"Got the following message instead:\n\n{str(e)}") + logger.info(f"Successfully prevented {replica_cluster.name} from becoming a second active cluster") + + +def run_test_iteration(active_cluster: HBaseDockerClient, replica_cluster: HBaseDockerClient, data_root: str): + create_table_and_test_active_and_replica_clusters(active_cluster, replica_cluster, column_family='cf') + assert_error_when_trying_to_have_second_active_cluster(replica_cluster, EXPECTED_ERROR_MSG) + + # Cluster should still be in read-only mode after failed transition from read-only to read-write mode + replica_cluster.assert_read_only_error_occurs('create', 'test_table', COLUMN_FAMILY) + + assert_crud_operations_work_on_active_cluster(active_cluster) + + # Demote active cluster to replica and promote original replica to be the new active cluster + active_cluster.enable_read_only_mode() + replica_cluster.disable_read_only_mode() + active_cluster = replica_cluster + + # Wait for active cluster file to be updated and verify its contents + sleep(3) + assert_correct_active_cluster_suffix(active_cluster, data_root) + + +def main(): + log_script_start(__file__, logger) + parser = argparse.ArgumentParser() + parser = add_common_skip_container_stop_or_restart_arg(parser) + args = parser.parse_args() + + skip_container_restart = args.skip_container_start_or_restart + + if skip_container_restart: + logger.info("Docker containers will NOT be started/restarted at the beginning of this test run") + else: + logger.info("Docker containers will be started/restarted at the beginning of this test run") + + cluster1, cluster2 = load_env_and_set_up_clients() + data_store_root = get_env("HBASE_DATA_STORE_ROOT") + docker_compose_file = get_env("DOCKER_COMPOSE_FILE") + + reset_cluster_setup(active_cluster=cluster1, replica_cluster=cluster2, + skip_container_restart=skip_container_restart, docker_compose_file=docker_compose_file, + data_store_root=data_store_root, sudo=True) + + assert_correct_active_cluster_suffix(cluster1, data_store_root) + clean_up_tables(active_cluster=cluster1, replica_cluster=cluster2) + + test_iterations = 5 + for i in range(1, test_iterations+1): + logger.info(f"---------- Iteration {i} ----------") + if i % 2 == 1: + run_test_iteration(active_cluster=cluster1, replica_cluster=cluster2, data_root=data_store_root) + else: + run_test_iteration(active_cluster=cluster2, replica_cluster=cluster1, data_root=data_store_root) + logger.info(f"Finished iteration {i} of {test_iterations}") + log_script_end(__file__, logger) + + +if __name__ == '__main__': + main() diff --git a/dev-support/integration-test/read-replica/python/scripts/test_create_drop_behavior.py b/dev-support/integration-test/read-replica/python/scripts/test_create_drop_behavior.py new file mode 100644 index 000000000000..30bc87c46787 --- /dev/null +++ b/dev-support/integration-test/read-replica/python/scripts/test_create_drop_behavior.py @@ -0,0 +1,81 @@ +#!/usr/bin/env python3 +""" +Tests table creation behavior for read-replica clusters. It creates a table on the active +cluster, and then runs refresh_meta on the replica cluster and verifies the table's existence. +It does a similar process when dropping the table on the active cluster. It also verifies +tables cannot be created/dropped on the replica cluster. +""" +import argparse + +from python.src.logger_config import get_logger +from python.src.utils import (add_common_skip_table_cleanup_arg, clean_up_tables, load_env_and_set_up_clients, + log_script_start, log_script_end) + +logger = get_logger(__name__) + + +def test_table_creation_behavior(active_cluster, replica_cluster, table_name, column_family): + # We should not be able to create a new table on the read-replica cluster + replica_cluster.assert_read_only_error_occurs('create', table_name, column_family) + + active_cluster.create_table(table_name, column_family) + + # Read-Replica cluster should not see the newly created table yet + logger.info(f"Verifying {active_cluster.name} now has table '{table_name}', " + f"while {replica_cluster.name} cluster does not") + active_cluster.assert_table_exists(table_name) + replica_cluster.assert_table_does_not_exist(table_name) + + # Read-Replica cluster should now see the newly created table + replica_cluster.refresh_meta() + logger.info(f"Verifying {replica_cluster.name} has table '{table_name}' after refreshing meta") + replica_cluster.assert_table_exists(table_name) + active_cluster.assert_table_exists(table_name) + + # Cannot drop the table on the Read-Replica cluster. A WriteAttemptedOnReadOnlyClusterException should occur + replica_cluster.disable_table(table_name) + replica_cluster.assert_read_only_error_occurs('drop', table_name, column_family) + # The table should still exist on the read-replica cluster since drops are not allowed + replica_cluster.assert_table_exists(table_name) + + # Drop the table on the active cluster + active_cluster.disable_table(table_name) + active_cluster.drop_table(table_name) + + # The read-replica cluster should still have the table that was dropped on the active + # cluster since 'refresh_meta' has not been run yet. + logger.info(f"Verifying {replica_cluster.name} still has table '{table_name}'") + active_cluster.assert_table_does_not_exist(table_name) + replica_cluster.assert_table_exists(table_name) + + # The read-replica cluster no longer has the dropped table after running 'refresh_meta'. + logger.info(f"Verifying {replica_cluster.name} no longer has table '{table_name}' after " + f"refreshing meta") + replica_cluster.refresh_meta() + replica_cluster.assert_table_does_not_exist(table_name) + + +def main(): + log_script_start(__file__, logger) + + parser = argparse.ArgumentParser() + parser = add_common_skip_table_cleanup_arg(parser) + args = parser.parse_args() + + active_cluster, replica_cluster = load_env_and_set_up_clients(cluster1_name="Active Cluster", + cluster2_name="Read-Replica Cluster") + table_name = "t1" + column_family = "cf" + if not args.skip_table_cleanup_on_start: + # Delete any lingering tables + logger.info(f"Checking if table '{table_name}' already exists on {active_cluster.name} " + f"and dropping it if necessary") + clean_up_tables(active_cluster, replica_cluster) + + test_table_creation_behavior(active_cluster, replica_cluster, table_name, column_family) + + log_script_end(__file__, logger) + + +if __name__ == "__main__": + main() diff --git a/dev-support/integration-test/read-replica/python/scripts/test_dual_active_cluster_startup.py b/dev-support/integration-test/read-replica/python/scripts/test_dual_active_cluster_startup.py new file mode 100644 index 000000000000..1f6cd9e410e0 --- /dev/null +++ b/dev-support/integration-test/read-replica/python/scripts/test_dual_active_cluster_startup.py @@ -0,0 +1,124 @@ +#!/usr/bin/env python3 +""" +Verifies that two clusters cannot both start with read-only mode disabled (both as active clusters) +on the same shared data store. One cluster must fail to start, with the HMaster process not +running, and an error logged to the master log. + +Usage: python3 ./python/scripts/test_dual_active_cluster_startup.py +""" +import argparse +import time + +from python.src.environment_loader import get_env +from python.src.hbase_docker_client import HBaseDockerClient +from python.src.logger_config import get_logger +from python.src.utils import load_env_and_set_up_clients, log_script_start, log_script_end + +logger = get_logger(__name__) + +STARTUP_WAIT_SECONDS = 60 +EXPECTED_ERROR_MSG = "Another cluster is running in active (read-write) mode on this storage location" + + +def is_process_running(cluster: HBaseDockerClient, process_name: str) -> bool: + output = cluster.run_docker_exec_command("jps") + return process_name in output + + +def check_cluster_processes(cluster: HBaseDockerClient) -> bool: + hmaster_running = is_process_running(cluster, "HMaster") + logger.info(f" {cluster.name}: HMaster={'running' if hmaster_running else 'down'}") + return hmaster_running + + +def assert_error_in_master_log(cluster: HBaseDockerClient): + logger.info(f"Checking {cluster.name} master log for expected error message") + log_output = cluster.run_docker_exec_command( + "cat /opt/hbase/logs/hbase-*-master-*.log || true" + ) + assert EXPECTED_ERROR_MSG in log_output, ( + f"Expected {cluster.name}'s master log to contain:\n" + f" '{EXPECTED_ERROR_MSG}'\n" + f"but it was not found.\nLog tail:\n{log_output[-2000:]}" + ) + logger.info(f" [PASS] Found expected error message in {cluster.name}'s master log") + + +def main(): + log_script_start(__file__, logger) + + parser = argparse.ArgumentParser() + parser.add_argument('-c', '--clean-up-containers', action='store_true', + help='Stop Docker containers and revert cluster configurations to one ' + 'active cluster and one replica cluster after the test finishes') + args = parser.parse_args() + + cluster1, cluster2 = load_env_and_set_up_clients() + data_store_root = get_env("HBASE_DATA_STORE_ROOT") + docker_compose_file = get_env("DOCKER_COMPOSE_FILE") + + test_iterations = 3 + for i in range(1, test_iterations+1): + logger.info(f"---------- Iteration {i} ----------") + + HBaseDockerClient.stop_containers(docker_compose_file=docker_compose_file, data_dir=f'{data_store_root}/*', + sudo=True) + + # Make both clusters an active cluster (read-only disabled) + cluster1.disable_read_only_mode(run_update_all_config=False) + cluster2.disable_read_only_mode(run_update_all_config=False) + + # Start or restart containers so both attempt to start as active + HBaseDockerClient.start_or_restart_containers(docker_compose_file=docker_compose_file, + data_store_root=f'{data_store_root}') + + # Wait for HBase to attempt startup on both containers + logger.info(f"Waiting {STARTUP_WAIT_SECONDS}s for clusters to attempt startup...") + time.sleep(STARTUP_WAIT_SECONDS) + + # Determine which cluster failed + logger.info("Checking HBase processes on both clusters") + cluster1_running = check_cluster_processes(cluster1) + cluster2_running = check_cluster_processes(cluster2) + + if cluster1_running and not cluster2_running: + failed_cluster = cluster2 + running_cluster = cluster1 + elif cluster2_running and not cluster1_running: + failed_cluster = cluster1 + running_cluster = cluster2 + elif not cluster1_running and not cluster2_running: + raise RuntimeError("Both clusters appear to be down — this is unexpected") + else: + raise RuntimeError( + "Both clusters appear to be running — the test expects exactly one to have failed. " + "This may indicate the clusters are using separate data stores or the feature is not working. " + "Note: There is a rare occasion where this may occur due to a race condition, but it should " + "not happen often." + ) + + logger.info(f"[PASS] {running_cluster.name} is running as the active cluster") + logger.info(f"[PASS] {failed_cluster.name} failed to start (HMaster is down)") + + # Verify the failed cluster's master log contains the expected error + assert_error_in_master_log(failed_cluster) + + logger.info(f"Finished iteration {i} of {test_iterations}") + + logger.info("=" * 70) + logger.info("TEST PASSED: All dual active cluster startups were correctly rejected") + logger.info("=" * 70) + + if args.clean_up_containers: + logger.info("Stopping Docker containers and reverting test environment to having " + "one active cluster and one replica cluster") + HBaseDockerClient.stop_containers(docker_compose_file=docker_compose_file, data_dir=f'{data_store_root}/*', + sudo=True) + cluster1.disable_read_only_mode(run_update_all_config=False) + cluster2.enable_read_only_mode(run_update_all_config=False) + + log_script_end(__file__, logger) + + +if __name__ == '__main__': + main() diff --git a/dev-support/integration-test/read-replica/python/scripts/test_put_get_delete_behavior.py b/dev-support/integration-test/read-replica/python/scripts/test_put_get_delete_behavior.py new file mode 100644 index 000000000000..2d2ee4eb01f4 --- /dev/null +++ b/dev-support/integration-test/read-replica/python/scripts/test_put_get_delete_behavior.py @@ -0,0 +1,115 @@ +#!/usr/bin/env python3 +""" +Verifies data can be added to/deleted from the active cluster and the read-replica cluster +does not see this data until refresh_hfiles has been run. Also verifies put and delete +operations on the read-replica cluster result in an error. +""" +import argparse + +from python.src.hbase_docker_client import HBaseDockerClient, DockerExecCommandError, DockerExecCommandTimeoutError +from python.src.logger_config import get_logger +from python.src.utils import (add_common_skip_table_cleanup_arg, clean_up_tables, load_env_and_set_up_clients, + log_script_start, log_script_end) + +logger = get_logger(__name__) + + +def assert_cannot_flush_table_on_replica(replica_cluster: HBaseDockerClient, table: str, timeout: int = 60): + logger.info(f"Verifying table '{table}' cannot be flushed on {replica_cluster.name} " + f"because read-only mode is enabled") + try: + replica_cluster.flush(table, timeout=timeout) + raise RuntimeError(f"Expected flush on replica cluster {replica_cluster.name} to result in an error") + except DockerExecCommandTimeoutError: + raise RuntimeError( + f"TIMEOUT: flush on replica cluster '{replica_cluster.name}' did not complete within " + f"{timeout} seconds. This may indicate HBASE-30301 has not been fixed on this cluster." + ) + except DockerExecCommandError as e: + expected_error_msg = ("org.apache.hadoop.hbase.WriteAttemptedOnReadOnlyClusterException: " + "Operation not allowed in Read-Only Mode") + assert expected_error_msg in str(e), (f"Expected exception to contain the following error message after " + f"attempting a flush on replica cluster {replica_cluster.name}:\n" + f"{expected_error_msg}\n" + f"The actual exception was:\n{e}") + logger.info(f"Flush for table '{table}' on replica cluster {replica_cluster.name} failed as expected") + + +def test_put_delete_behavior(active_cluster, replica_cluster, table_name, column): + # Add data to the table on the active cluster + logger.info(f"Adding data to '{table_name}' on {active_cluster.name} and verifying it exists") + active_cluster.put(table_name, "row1", column, "value1") + active_cluster.assert_table_row_count(table_name, 1) + active_cluster.assert_get_output(table_name, "row1", column, "value1") + + # Verify the read-replica cluster does not see this new data + logger.info(f"Verifying '{table_name}' on {replica_cluster.name} still has 0 rows") + replica_cluster.assert_table_row_count(table_name, 0) + + # Flush the table's data on the active cluster + logger.info(f"Flushing '{table_name}' on {active_cluster.name} and refreshing meta and " + f"HFiles on {replica_cluster.name}") + active_cluster.flush(table_name) + + # Refresh meta and HFiles, and verify the read-replica cluster now sees the data + logger.info(f"Refreshing meta and HFiles on {replica_cluster.name}") + replica_cluster.refresh_meta_and_hfiles() + logger.info(f"Verifying '{table_name}' on {replica_cluster.name} has data after refreshing HFiles") + replica_cluster.assert_table_row_count(table_name, 1) + replica_cluster.assert_get_output(table_name, "row1", column, "value1") + + # Verify replica clusters cannot flush tables + assert_cannot_flush_table_on_replica(replica_cluster, table_name) + + # Verify data cannot be added to the table on the read-replica cluster + logger.info(f"Verifying data cannot be added to '{table_name}' on {replica_cluster.name}") + replica_cluster.assert_read_only_error_occurs('put', table_name, column, 'row2', 'value2') + + # Verify data cannot be deleted from the table on the read-replica cluster + logger.info(f"Verifying data cannot be deleted from '{table_name}' on {replica_cluster.name}") + replica_cluster.assert_read_only_error_occurs('delete', table_name, column, 'row2') + + # Delete data from the active cluster + logger.info(f"Deleting row from '{table_name}' on {active_cluster.name} " + f"and verifying it is gone") + active_cluster.delete(table_name, "row1", column) + active_cluster.flush(table_name) + active_cluster.assert_table_row_count(table_name, 0) + + # Verify deleted data still exists on the read-replica cluster + logger.info(f"Verifying deleted row still exists on {replica_cluster.name}") + replica_cluster.assert_table_row_count(table_name, 1) + replica_cluster.assert_get_output(table_name, "row1", column, "value1") + + # Verify the read-replica cluster no longer has the data after refreshing HFiles + replica_cluster.refresh_hfiles() + replica_cluster.assert_table_row_count(table_name, 0) + + +def main(): + log_script_start(__file__, logger) + + parser = argparse.ArgumentParser() + parser = add_common_skip_table_cleanup_arg(parser) + args = parser.parse_args() + + active_cluster, replica_cluster = load_env_and_set_up_clients(cluster1_name="Active Cluster", + cluster2_name="Read-Replica Cluster") + table_name = "t1" + column_family = "cf" + column = f"{column_family}:c1" + + if not args.skip_table_cleanup_on_start: + clean_up_tables(active_cluster, replica_cluster) + + # Create a table on the active cluster and have it appear on the read-replica cluster + active_cluster.create_table(table_name, column_family) + replica_cluster.refresh_meta() + + test_put_delete_behavior(active_cluster, replica_cluster, table_name, column) + + log_script_end(__file__, logger) + + +if __name__ == '__main__': + main() diff --git a/dev-support/integration-test/read-replica/python/scripts/test_read_only_flag_flipping.py b/dev-support/integration-test/read-replica/python/scripts/test_read_only_flag_flipping.py new file mode 100644 index 000000000000..e1051e570a52 --- /dev/null +++ b/dev-support/integration-test/read-replica/python/scripts/test_read_only_flag_flipping.py @@ -0,0 +1,116 @@ +#!/usr/bin/env python3 +""" +This test starts with two Read-Replica HBase clusters, where one cluster is the active cluster and the other cluster is +the replica cluster. The test creates a table on the active cluster, adds data to the cluster, and verifies this data +is consistent on the replica cluster after refreshing the meta HFiles. It also verifies write operations cannot be +performed on the replica cluster. Then, the two clusters swap roles, where the active cluster becomes a replica and the +former replica becomes the new active cluster. The previous steps then repeat in an iterative fashion. + +This test script verifies behavior for multiple bug fixes: + +1. HBASE-30090: Table on replica cluster not refreshing after flipping read-only flag twice + https://issues.apache.org/jira/browse/HBASE-30090 + + Before implementing this fix, an existing table on a read-replica cluster was not getting updated after making that + cluster the active cluster and then making it read-only again. + +2. HBASE-30180: Can still add data to read-only region after flipping read-only flag multiple times + https://issues.apache.org/jira/browse/HBASE-30180 + + Before implementing this fix, this cluster setup and series of steps would eventually get to a scenario where data + could be added to a table on cluster with read-only mode disabled. +""" +import argparse + +from python.src.utils import (assert_correct_active_cluster_suffix, add_common_skip_container_stop_or_restart_arg, + clean_up_tables, reset_cluster_setup, load_env_and_set_up_clients, + create_table_and_test_active_and_replica_clusters, + log_script_start, log_script_end) + +from python.src.environment_loader import get_env +from python.src.hbase_docker_client import HBaseDockerClient +from python.src.logger_config import get_logger + +COLUMN_FAMILY = "cf" +logger = get_logger(__name__) + + +def flip_read_only_flag(new_active_cluster: HBaseDockerClient, + new_replica_cluster: HBaseDockerClient): + # Make cluster read-only and verify it cannot create a table or put data + new_replica_cluster.enable_read_only_mode() + new_replica_cluster.assert_read_only_error_occurs('create', 'testTable', COLUMN_FAMILY) + new_replica_cluster.assert_read_only_error_occurs( + 'put', 't1', COLUMN_FAMILY, row='r2', data='2') + + # Make cluster active + new_active_cluster.disable_read_only_mode() + + +def create_table_and_test_clusters_then_flip_read_only_flag(cluster1, cluster2, data_store_root): + create_table_and_test_active_and_replica_clusters(active_cluster=cluster1, replica_cluster=cluster2, + column_family='cf') + flip_read_only_flag(new_active_cluster=cluster2, new_replica_cluster=cluster1) + assert_correct_active_cluster_suffix(cluster2, data_store_root) + + +def main(): + log_script_start(__file__, logger) + + parser = argparse.ArgumentParser() + parser = add_common_skip_container_stop_or_restart_arg(parser) + args = parser.parse_args() + + skip_container_restart = args.skip_container_start_or_restart + + if skip_container_restart: + logger.info("Docker containers will NOT be started/restarted at the beginning of this test run") + else: + logger.info("Docker containers will be started/restarted at the beginning of this test run") + + cluster1, cluster2 = load_env_and_set_up_clients() + data_store_root = get_env("HBASE_DATA_STORE_ROOT") + docker_compose_file = get_env("DOCKER_COMPOSE_FILE") + + reset_cluster_setup(active_cluster=cluster1, replica_cluster=cluster2, + skip_container_restart=skip_container_restart, docker_compose_file=docker_compose_file, + data_store_root=data_store_root, sudo=True) + + if not args.skip_container_start_or_restart: + HBaseDockerClient.start_or_restart_containers(docker_compose_file=docker_compose_file, + data_store_root=f'{data_store_root}') + HBaseDockerClient.wait_for_clusters_to_start([cluster1, cluster2]) + + test_iterations = 1 + read_only_flag_flips_per_iteration = 15 + for i in range(1, test_iterations + 1): + logger.info(f"---------- Iteration {i} ----------") + if i > 1: + logger.info(f"Ensuring clusters are in proper modes. " + f"Making both clusters a replica, and then making {cluster1.name} the active cluster") + cluster1.enable_read_only_mode() + cluster2.enable_read_only_mode() + cluster1.disable_read_only_mode() + + # Create table on active cluster + clean_up_tables(cluster1, cluster2) + + # One iteration flips the read-only flag on each cluster and then flips it back. + flip_num = 1 + while flip_num <= read_only_flag_flips_per_iteration: + logger.info(f"*** Testing read-only flag flip number {flip_num} ***") + if flip_num % 2 == 1: + # Cluster 1 is active and Cluster 2 is replica + create_table_and_test_clusters_then_flip_read_only_flag(cluster1, cluster2, data_store_root) + else: + # Cluster 2 is active and Cluster 1 is replica + create_table_and_test_clusters_then_flip_read_only_flag(cluster2, cluster1, data_store_root) + logger.info(f"Finished read-only flag flip {flip_num} of {read_only_flag_flips_per_iteration}") + flip_num += 1 + logger.info(f"Finished iteration {i} of {test_iterations}") + + log_script_end(__file__, logger) + + +if __name__ == '__main__': + main() diff --git a/dev-support/integration-test/read-replica/python/scripts/verify_hbase_start.py b/dev-support/integration-test/read-replica/python/scripts/verify_hbase_start.py new file mode 100644 index 000000000000..e9c852b9a4ba --- /dev/null +++ b/dev-support/integration-test/read-replica/python/scripts/verify_hbase_start.py @@ -0,0 +1,40 @@ +#!/usr/bin/env python3 +""" +Verifies the hbase-docker containers have started properly. For each cluster, the script first +curls the HBase UI until it receives a 200 response and then gets the server status to verify +there are no dead clusters +""" +import argparse + +from python.src import get_env +from python.src.hbase_docker_client import HBaseDockerClient +from python.src.logger_config import get_logger +from python.src.utils import (load_env_and_set_up_clients, log_script_start, log_script_end, + add_common_skip_container_stop_or_restart_arg) + +logger = get_logger(__name__) + + +def main(): + log_script_start(__file__, logger) + + parser = argparse.ArgumentParser() + parser = add_common_skip_container_stop_or_restart_arg(parser) + args = parser.parse_args() + + active_cluster, replica_cluster = load_env_and_set_up_clients(cluster1_name="Active Cluster", + cluster2_name="Read-Replica Cluster") + data_store_root = get_env("HBASE_DATA_STORE_ROOT") + docker_compose_file = get_env("DOCKER_COMPOSE_FILE") + + if not args.skip_container_start_or_restart: + HBaseDockerClient.start_or_restart_containers(docker_compose_file=docker_compose_file, + data_store_root=f'{data_store_root}') + + HBaseDockerClient.wait_for_clusters_to_start([active_cluster, replica_cluster]) + + log_script_end(__file__, logger) + + +if __name__ == "__main__": + main() diff --git a/dev-support/integration-test/read-replica/python/src/__init__.py b/dev-support/integration-test/read-replica/python/src/__init__.py new file mode 100644 index 000000000000..64f912d1e5cc --- /dev/null +++ b/dev-support/integration-test/read-replica/python/src/__init__.py @@ -0,0 +1,3 @@ +from .hbase_docker_client import HBaseDockerClient +from .environment_loader import get_env +from .logger_config import get_logger diff --git a/dev-support/integration-test/read-replica/python/src/environment_loader.py b/dev-support/integration-test/read-replica/python/src/environment_loader.py new file mode 100644 index 000000000000..49de063f4ebf --- /dev/null +++ b/dev-support/integration-test/read-replica/python/src/environment_loader.py @@ -0,0 +1,10 @@ +#!/usr/bin/env python3 +import os + + +def get_env(key, default=None): + """Retrieve environment variables, ensuring they are loaded from the GitHub Actions runner.""" + val = os.environ.get(key, default) + if val is None: + raise RuntimeError(f"Error: Environment variable {key} is not set.") + return val diff --git a/dev-support/integration-test/read-replica/python/src/hbase_docker_client.py b/dev-support/integration-test/read-replica/python/src/hbase_docker_client.py new file mode 100644 index 000000000000..890ca8f44b7a --- /dev/null +++ b/dev-support/integration-test/read-replica/python/src/hbase_docker_client.py @@ -0,0 +1,529 @@ +#!/usr/bin/env python3 +import ast +import logging +import re +import requests +import subprocess +import time +import xml.etree.ElementTree as ET + +from .logger_config import get_logger + +logger = get_logger(__name__) + + +class DockerExecCommandError(Exception): + pass + + +class HBaseShellCommandError(DockerExecCommandError): + pass + + +class DockerExecCommandTimeoutError(DockerExecCommandError): + pass + + +class HBaseDockerClient: + def __init__(self, container_name: str, local_conf: str, hbase_ui_port: int = 16010, + cluster_name: str = "HBase Cluster", max_retries: int = 12, sleep_time: int = 5) -> None: + self._container_name = container_name + self._local_conf = local_conf + self._hbase_ui_port = hbase_ui_port + self._cluster_name = cluster_name + self._max_retries = max_retries + self._sleep_time = sleep_time + + @property + def name(self) -> str: + return self._cluster_name + + def run_docker_exec_command(self, bash_cmd: str, timeout: int | None = None) -> str: + """ + Uses 'docker exec' to run the provided Bash command in the object's Docker container. + The command looks like: docker exec bash -c + Note: In the Terminal, we usually put double quotes around everything after "-c", + but doing that with subprocess.run() results in a failure. + """ + cmd = ["docker", "exec", self._container_name, "bash", "-c", f'''{bash_cmd}'''] + cmd_str = ' '.join(cmd) + logger.debug(f"Running command on {self._cluster_name}: {cmd_str}") + try: + process = subprocess.run(cmd, capture_output=True, timeout=timeout) + except subprocess.TimeoutExpired: + raise DockerExecCommandTimeoutError( + f"Command timed out after {timeout}s on {self._cluster_name} " + f"({self._container_name}): {bash_cmd}\n" + f"The command used to run this was: {cmd_str}\n" + ) + stdout = process.stdout.decode('utf-8') + if process.returncode != 0: + raise DockerExecCommandError( + f"The following command failed on {self._cluster_name} ({self._container_name}): {bash_cmd}\n" + f"The command used to run this was: {cmd_str}\n" + f"The command's STDERR was:\n{process.stderr.decode('utf-8')}\n" + f"The command's STDOUT was:\n{stdout}\n" + ) + return stdout + + def run_hbase_shell_command(self, hbase_cmd: str, timeout: int | None = None) -> str: + """ + Uses 'docker exec' to run the provided HBase shell command in the object's Docker container. + The command looks like: docker exec bash -c hbase shell -n <<< "" + """ + hbase_shell_cmd = f'''hbase shell -n <<< "{hbase_cmd}"''' + try: + return self.run_docker_exec_command(hbase_shell_cmd, timeout=timeout) + except DockerExecCommandTimeoutError: + # DockerExecCommandTimeoutError is a subclass of DockerExecCommandError, so we need to make sure + # it's specifically caught and re-raised. Otherwise, it's swallowed when catching DockerExecCommandError + raise + except DockerExecCommandError as e: + raise HBaseShellCommandError(e) + + def wait_for_hbase_ui(self) -> bool: + """Checks for a 200 OK on the HBase Master UI.""" + url = f"http://localhost:{self._hbase_ui_port}" + logger.info(f"Waiting for HBase UI: {self._cluster_name} on {url}") + last_exception = None + for attempt in range(1, self._max_retries + 1): + try: + response = requests.get(url) + if response.status_code == 200: + logger.info(f"SUCCESS: {self._cluster_name} UI is up.") + return True + except requests.exceptions.ConnectionError as e: + last_exception = e + logging.info(f"Waiting {self._sleep_time} seconds before requesting HBase UI again") + time.sleep(self._sleep_time) + + raise RuntimeError(f"\nTIMEOUT: {self._cluster_name} UI failed to respond after " + f"{self._max_retries} attempts. " + f"Last raised exception was: {last_exception}") + + def check_server_status(self, desired_status: dict | None = None) -> bool: + """Runs 'status' inside the HBase shell and validates the output.""" + if desired_status is None: + desired_status = {'masters': '1', 'region_servers': '1', 'dead_servers': '0'} + logger.info(f"Validating Cluster Status: {self._cluster_name} ({self._container_name})") + for attempt in range(1, self._max_retries + 1): + try: + output = self.get_hbase_status() + + # The cluster's status should have 1 active master, 1 region server, + # and no dead servers + validations = { + "Active Master": f"{desired_status['masters']} active master" in output, + "Region Server": f"{desired_status['region_servers']} servers" in output, + "No Dead Servers": f"{desired_status['dead_servers']} dead" in output + } + + if all(validations.values()): + for check, status in validations.items(): + logger.info(f" [PASS] {check}") + logger.info(f"SUCCESS: {self._cluster_name} is fully operational.") + return True + else: + logger.warning(f"{self._cluster_name} is responding, but not all " + f"components are ready...") + logger.info(f"HBase 'status' command output:\n{output}") + + except subprocess.CalledProcessError: + pass + + logging.info(f"Waiting {self._sleep_time} seconds before getting status on " + f"{self.name} again") + time.sleep(self._sleep_time) + + raise RuntimeError( + f"\nTIMEOUT: {self._cluster_name} shell check failed after {self._max_retries} attempts.") + + def get_hbase_status(self) -> str: + logger.debug(f"Getting status of {self.name}") + return self.run_hbase_shell_command("status") + + def wait_for_cluster_to_start(self) -> None: + """curls the cluster's HBase UI to make sure it is up and then makes sure all desired servers are up""" + self.wait_for_hbase_ui() + self.check_server_status() + + def create_table(self, table_name: str, column_family: str) -> bool: + logger.info(f"Creating table '{table_name}' on {self._cluster_name}") + create_cmd = f"create '{table_name}', '{column_family}'" + output = self.run_hbase_shell_command(create_cmd) + + if f"Created table {table_name}" not in output: + logger.error(f"Could not create table '{table_name}' on {self._cluster_name}") + return False + return True + + def disable_table(self, table_name: str) -> None: + logger.debug(f"Disabling table '{table_name}' on {self.name}") + self.run_hbase_shell_command(f"disable '{table_name}'") + + def drop_table(self, table_name: str) -> None: + logger.info(f"Dropping table '{table_name}' on {self.name}") + self.run_hbase_shell_command(f"drop '{table_name}'") + + def list_tables(self) -> list: + """Gets the list of HBase tables and returns it as a Python list""" + logger.debug(f"Getting the list of tables in HBase on {self.name}") + pattern = r'\[(.*?)\]' + output = self.run_hbase_shell_command("list") + output = output.replace('\n', ' ') + match = re.search(pattern, output) + return ast.literal_eval(match.group(0)) + + def list_regions(self, table_name: str) -> str: + """Gets list of regions and their info for the provided table""" + logger.info(f"Getting list of regions for table '{table_name}'") + return self.run_hbase_shell_command(f"list_regions '{table_name}'") + + def put(self, table_name: str, row: str, column: str, data: str, spec_map: str | None = None) -> None: + """ + Performs an HBase put command. + :param table_name: the table we are inserting data into + :param row: row of the table we are inserting data into + :param column: column of the table we are inserting data into + :param data: the actual data we are inserting (as a string) + :param spec_map: additional attributes input as a string + (e.g. "{ATTRIBUTES=>{'my-key'=>'my-value'}}") + """ + logger.info(f"Adding data to table '{table_name}' on {self.name}") + put_cmd = f"put '{table_name}', '{row}', '{column}', '{data}'" + if spec_map: + put_cmd += f", {spec_map}" + self.run_hbase_shell_command(put_cmd) + + def get(self, table_name: str, row: str, column: str | None = None, spec_map: str | None = None) -> str: + logger.info(f"Getting data from table '{table_name}' on {self.name}") + get_cmd = f"get '{table_name}', '{row}'" + if column: + get_cmd += f", '{column}'" + if spec_map: + get_cmd += f", {spec_map}" + output = self.run_hbase_shell_command(get_cmd) + logger.debug(f"Got data:\n{output}") + return output + + def delete(self, table_name: str, row: str, column: str, timestamp: int | None = None, + spec_map: str | None = None) -> None: + logger.info(f"Deleting data from table '{table_name}' on {self.name}") + delete_cmd = f"delete '{table_name}', '{row}', '{column}'" + if timestamp: + delete_cmd += f", {table_name}" + if spec_map: + delete_cmd += f", {spec_map}" + self.run_hbase_shell_command(delete_cmd) + + def scan(self, table_name: str, spec_map: str | None = None) -> str: + log_msg = f"Scanning table '{table_name}' on {self.name}" + scan_cmd = f"scan '{table_name}'" + if spec_map: + scan_cmd += f", {spec_map}" + log_msg += f" with spec_map {spec_map}" + logging.info(log_msg) + return self.run_hbase_shell_command(scan_cmd) + + def count(self, table_name: str, spec: str | None = None) -> str: + logger.info(f"Counting rows for table '{table_name}' on {self.name}") + count_cmd = f"count '{table_name}'" + if spec: + count_cmd += f"{spec}" + return self.run_hbase_shell_command(count_cmd) + + def flush(self, table_name: str, timeout: int | None = None) -> None: + logger.debug(f"Flushing table '{table_name}' on {self.name}") + self.run_hbase_shell_command(f"flush '{table_name}'", timeout=timeout) + + def split(self, thing_to_split: str, split_key: str | None = None) -> None: + log_msg = f"Splitting '{thing_to_split}'" + split_cmd = f"split '{thing_to_split}'" + + if split_key: + log_msg += f" on key '{split_key}'" + split_cmd += f", '{split_key}'" + + log_msg += f" on {self.name}" + + logger.info(log_msg) + self.run_hbase_shell_command(split_cmd) + + def flush_and_split(self, thing_to_split: str, split_key: str | None = None) -> None: + """ + Flushes the table and triggers an asynchronous region split. Split the entire table or pass a region to split an + individual region. With the second parameter, you can specify an explicit split key for the region. + + thing_to_split - TABLENAME, REGIONNAME, or ENCODED_REGIONNAME + split_key - where to have the region split + """ + self.flush(thing_to_split) + self.split(thing_to_split, split_key) + + def major_compact(self, table_or_region: str, column_family: str | None = None, mob: str | None = None) -> None: + log_msg = f"Running major_compact on '{table_or_region}'" + command = f"major_compact '{table_or_region}'" + + if column_family: + log_msg += f" for column family '{column_family}'" + command += f", '{column_family}'" + + if mob and column_family: + log_msg += " with MOB" + command += f", 'MOB'" + elif mob and not column_family: + log_msg += " with MOB" + command += ", nil, 'MOB'" + + self.run_hbase_shell_command(command) + + def major_compact_and_wait(self, table_or_region: str, column_family: str | None = None, mob: str | None = None, + timeout: int = 30, sleep_time: int = 1) -> bool: + """Triggers major compaction on a table and blocks until it completes.""" + logger.info(f"Triggering major compaction on '{table_or_region}' on {self.name}...") + self.major_compact(table_or_region, column_family, mob) + + start_time = time.time() + while time.time() - start_time < timeout: + output = self.run_hbase_shell_command(f"compaction_state '{table_or_region}'") + + # When all regions finish compacting, compaction_state returns NONE + if "NONE" in output: + logger.info(f"SUCCESS: Major compaction completed for '{table_or_region}'.") + return True + + logger.debug(f"Compaction still in progress for '{table_or_region}'... waiting {sleep_time}s") + time.sleep(sleep_time) + + raise RuntimeError( + f"TIMEOUT: Major compaction on table '{table_or_region}' failed to complete within {timeout} seconds." + ) + + def catalogjanitor_run(self) -> None: + """Forces the CatalogJanitor to immediately clean up split parent regions in hbase:meta.""" + logger.info(f"Running catalogjanitor_run on {self.name}") + self.run_hbase_shell_command("catalogjanitor_run") + + def refresh_meta(self) -> None: + logger.info(f"Refreshing meta on {self.name}") + self.run_hbase_shell_command("refresh_meta") + + def refresh_hfiles(self) -> None: + logger.info(f"Refreshing HFiles on {self.name}") + self.run_hbase_shell_command("refresh_hfiles") + + def refresh_meta_and_hfiles(self) -> None: + """Consecutively runs refresh_meta and refresh_hfiles in the HBase shell""" + self.refresh_meta() + self.refresh_hfiles() + + def enable_read_only_mode(self, run_update_all_config: bool = True) -> None: + """ + Sets hbase.global.readonly.enabled to 'true' in the local hbase-site.xml file and runs update_all_config + to dynamically update the configuration. This method assumes the hbase-site.xml file is a mounted volume + in the docker-compose file, which allows the config file within the docker container to be updated as well. + """ + self._set_read_only_mode(new_read_only_flag=True, run_update_all_config=run_update_all_config) + + def disable_read_only_mode(self, run_update_all_config: bool = True) -> None: + """ + Sets hbase.global.readonly.enabled to 'false' in the local hbase-site.xml file and runs update_all_config + to dynamically update the configuration. This method assumes the hbase-site.xml file is a mounted volume + in the docker-compose file, which allows the config file within the docker container to be updated as well. + """ + self._set_read_only_mode(new_read_only_flag=False, run_update_all_config=run_update_all_config) + + def _set_read_only_mode(self, new_read_only_flag: bool, run_update_all_config: bool = True) -> None: + action = "Enabling" if new_read_only_flag else "Disabling" + conjunction_adverb = "and then" if run_update_all_config else "but not" + logger.info(f"{action} read-only mode in conf for {self.name} " + f"{conjunction_adverb} running update_all_config after") + + new_read_only_flag = str(new_read_only_flag).lower() + self.set_hbase_conf_property_value('hbase.global.readonly.enabled', new_read_only_flag) + actual = self.get_hbase_conf_property_value('hbase.global.readonly.enabled') + assert actual == new_read_only_flag, ( + f"Expected hbase.global.readonly.enabled={new_read_only_flag} on {self.name}, but got '{actual}'" + ) + if run_update_all_config: + self.update_all_config() + + def update_all_config(self) -> None: + logger.debug(f"Running update_all_config on {self.name} to dynamically update the configuration") + self.run_hbase_shell_command("update_all_config") + + def get_hbase_conf_property_value(self, conf_prop: str) -> str | None: + tree = ET.parse(self._local_conf) + root = tree.getroot() + for prop in root.findall('property'): + name_elem = prop.find('name') + if name_elem is not None and name_elem.text == conf_prop: + return prop.find('value').text + + def set_hbase_conf_property_value(self, conf_prop: str, value: str) -> None: + """Sets hbase.global.readonly.enabled to a new value in a local hbase-site.xml file""" + tree = ET.parse(self._local_conf) + root = tree.getroot() + for prop in root.findall('property'): + name_elem = prop.find('name') + if name_elem is not None and name_elem.text == conf_prop: + value_elem = prop.find('value') + if value_elem is not None: + value_elem.text = str(value) + break + tree.write(self._local_conf, encoding='utf-8', xml_declaration=True) + # The conf file is a Docker volume - wait for the updated version to sync + time.sleep(1) + + def assert_read_only_error_occurs(self, cmd_type: str, table_name: str, column: str, + row: str | None = None, data: str | None = None) -> None: + """ + Runs a command on read-only cluster and expects an error to occur as a result. + """ + logger.info(f"Verifying we cannot perform a '{cmd_type}' on {self.name} " + f"since it is in read-only mode") + try: + # This should throw an exception + match cmd_type.lower(): + case 'create': + self.create_table(table_name, column) + case 'drop': + self.drop_table(table_name) + case 'put': + self.put(table_name, row, column, data) + case 'delete': + self.delete(table_name, row, column) + case _: + raise RuntimeError(f"Unexpected command type: {cmd_type}") + + # If we get here, then the command succeeded on the read-replica cluster, which should + # not have happened. + raise RuntimeError(f"Expected {cmd_type} attempt on {self.name} " + f"to result in an error") + except HBaseShellCommandError as e: + # Verify the command we ran on the read-replica cluster produced the expected exception + expected_error = ("org.apache.hadoop.hbase.WriteAttemptedOnReadOnlyClusterException: " + "Operation not allowed in Read-Only Mode") + assert expected_error in str(e), (f"Expected exception to contain the following: " + f"{expected_error}\n" + f"The actual exception was:\n{e}") + logger.info(f"{cmd_type.capitalize()} attempt on {self.name} failed as expected") + + def assert_table_does_not_exist(self, table_name: str) -> None: + logger.info(f"Verifying '{table_name}' is not in the list of tables on {self.name}") + assert table_name not in self.list_tables(), \ + f"Expected table '{table_name}' to not exist on {self.name}" + + def assert_table_exists(self, table_name: str) -> None: + logger.info(f"Verifying '{table_name}' is in the list of tables on {self.name}") + assert table_name in self.list_tables(), \ + f"Expected table '{table_name}' to exist on {self.name}" + + def assert_table_row_count(self, table_name: str, expected_row_count: int) -> None: + logger.info(f"Verifying table '{table_name}' on {self.name} has {expected_row_count} row(s)") + output = self.count(table_name) + match = re.search(r'^(\d+) row\(s\)$', output, re.MULTILINE) + actual_row_count = int(match.group(1)) if match else None + assert actual_row_count == expected_row_count, \ + (f"Expected table '{table_name}' on {self.name} to have {expected_row_count} row(s). " + f"Instead got {actual_row_count}") + + def assert_get_output(self, table: str, row: str, cf: str, expected_data: str) -> str: + output = self.get(table, row, cf) + assert f"value={expected_data}" in output, \ + f"Expected get command to retrieve a row with value={expected_data}. Output instead was:\n{output}" + return output + + def assert_region_count_for_table(self, table_name: str, expected_region_count: int) -> None: + logger.info(f"Verifying table '{table_name}' has {expected_region_count} region(s)") + output = self.list_regions(table_name) + match = re.search(r'^ (\d+) rows$', output, re.MULTILINE) + actual_region_count = int(match.group(1)) if match else None + assert actual_region_count == expected_region_count, \ + (f"Expected table '{table_name}' on {self.name} to have {expected_region_count} region(s). " + f"Instead got {actual_region_count}") + + @staticmethod + def __run_subprocess_command(command: list | str, error_msg: str, + shell: bool = False) -> subprocess.CompletedProcess: + if shell: + cmd_msg = command + else: + cmd_msg = f"{' '.join(command)}" + logger.info(f"Running: {cmd_msg}") + result = subprocess.run(command, capture_output=True, text=True, shell=shell) + if result.returncode != 0: + raise RuntimeError( + f"Command failed: {cmd_msg}\n" + f"{error_msg} (exit {result.returncode}):\n" + f"STDOUT: {result.stdout}\nSTDERR: {result.stderr}" + ) + return result + + @staticmethod + def wait_for_clusters_to_start(clusters: list) -> None: + for cluster in clusters: + cluster.wait_for_cluster_to_start() + logger.info("=" * 40) + logger.info("ALL CLUSTERS VERIFIED AND READY") + logger.info("=" * 40) + + @staticmethod + def are_containers_running(docker_compose_file: str | None = None) -> bool: + logger.info("Checking if docker containers are running") + command = ["docker", "compose"] + if docker_compose_file: + command += ["-f", docker_compose_file] + command += ["ps", "--status", "running", "-q"] + result = HBaseDockerClient.__run_subprocess_command(command, "Failed to get docker container status") + return bool(result.stdout.strip()) + + @staticmethod + def set_up_data_store_dir(data_store_root: str) -> None: + command = ["mkdir", "-p", f"{data_store_root}/data-store/hbase", f"{data_store_root}/data-store/run", + f"{data_store_root}/data-store/logs", f"{data_store_root}/data-store/zk"] + HBaseDockerClient.__run_subprocess_command(command, + error_msg=f"Failed to create {data_store_root} and its sub-dirs") + command = ["chmod", "-R", "777", f"{data_store_root}"] + HBaseDockerClient.__run_subprocess_command(command, + error_msg=f"Failed to give {data_store_root} " + f"and its sub-dirs full permissions") + + @staticmethod + def start_or_restart_containers(docker_compose_file: str | None = None, data_store_root: str | None = None) -> None: + if data_store_root: + HBaseDockerClient.set_up_data_store_dir(data_store_root) + + if HBaseDockerClient.are_containers_running(docker_compose_file): + logger.info("Restarting docker containers") + command = ["docker", "compose"] + if docker_compose_file: + command += ["-f", docker_compose_file] + command += ["restart"] + action = "restart" + else: + logger.info("Starting docker containers") + command = ["docker", "compose"] + if docker_compose_file: + command += ["-f", docker_compose_file] + command += ["up", "-d"] + action = "start" + + HBaseDockerClient.__run_subprocess_command(command, f"docker compose {action} failed") + logger.info(f"docker compose {action} completed successfully") + + @staticmethod + def stop_containers(docker_compose_file: str | None = None, data_dir: str | None = None, + sudo: bool = False) -> None: + command = "docker compose" + if docker_compose_file: + command += f" -f {docker_compose_file}" + command += " down" + log_msg = "Stopping docker containers" + if data_dir: + rm_cmd = "sudo rm -rf" if sudo else "rm -rf" + command += f" && {rm_cmd} {data_dir}" + log_msg += f" and deleting HBase data root dir at: {data_dir}" + logger.info(f"{log_msg}") + HBaseDockerClient.__run_subprocess_command(command, "stop_containers failed", shell=True) + logger.info("Successfully stopped docker containers") diff --git a/dev-support/integration-test/read-replica/python/src/logger_config.py b/dev-support/integration-test/read-replica/python/src/logger_config.py new file mode 100644 index 000000000000..0a3edec98fdc --- /dev/null +++ b/dev-support/integration-test/read-replica/python/src/logger_config.py @@ -0,0 +1,34 @@ +import logging +import sys + +from dotenv import load_dotenv +from .environment_loader import get_env + +LOG_FORMAT = '%(asctime)s %(levelname)-5s %(module)s.%(funcName)s(%(lineno)d): %(message)s' + +# Load settings from .env file +load_dotenv() + + +def configure_logging(level=get_env('LOG_LEVEL')): + """ + Centralized logging configuration for HBase testing scripts. + """ + logging.basicConfig( + format='%(asctime)s %(levelname)-5s %(module)s.%(funcName)s(%(lineno)d): %(message)s', + level=level, + handlers=[ + logging.StreamHandler(sys.stdout) # Ensures logs show up in GH Actions console + ] + ) + + +def get_logger(name): + """ + Helper to get a logger. This can be used to ensure the config + is applied whenever a logger is requested. + """ + # If the root logger has no handlers, configure it now + if not logging.getLogger().hasHandlers(): + configure_logging() + return logging.getLogger(name) diff --git a/dev-support/integration-test/read-replica/python/src/utils.py b/dev-support/integration-test/read-replica/python/src/utils.py new file mode 100644 index 000000000000..5675538bf874 --- /dev/null +++ b/dev-support/integration-test/read-replica/python/src/utils.py @@ -0,0 +1,228 @@ +#!/usr/bin/env python3 + +import argparse +import os +import time + +from dotenv import load_dotenv + +import python.proto.generated.ActiveClusterSuffix_pb2 as acs + +from python.src.environment_loader import get_env +from python.src.hbase_docker_client import HBaseDockerClient +from python.src.logger_config import get_logger + +logger = get_logger(__name__) + + +def log_script_start(file: str, script_logger=None): + (script_logger or logger).info(f"========== START {os.path.basename(file)} ==========") + + +def log_script_end(file: str, script_logger=None): + (script_logger or logger).info(f"========== END {os.path.basename(file)} ==========") + + +def add_common_skip_table_cleanup_arg(parser: argparse.ArgumentParser) -> argparse.ArgumentParser: + parser.add_argument('-t', '--skip-table-cleanup-on-start', action='store_true', + help='Skip cleaning up tables at the start of the test') + return parser + + +def add_common_skip_container_stop_or_restart_arg(parser: argparse.ArgumentParser) -> argparse.ArgumentParser: + parser.add_argument('-s', '--skip-container-start-or-restart', action='store_true', + help='Skip stopping, starting, and waiting for the Docker containers to be ready') + return parser + + +def load_env_and_set_up_clients(cluster1_name: str = "Cluster 1", + cluster2_name: str = "Cluster 2") -> tuple[HBaseDockerClient, HBaseDockerClient]: + load_dotenv() + container_name = get_env("HBASE_CONTAINER_NAME") + + active_cluster = HBaseDockerClient(container_name=container_name, + local_conf=f"{get_env('ACTIVE_CLUSTER_CONF_DIR')}/hbase-site.xml", + hbase_ui_port=get_env('ACTIVE_CLUSTER_PORT'), + cluster_name=cluster1_name) + replica_cluster = HBaseDockerClient(container_name=f'{container_name}-2', + local_conf=f"{get_env('REPLICA_CLUSTER_CONF_DIR')}/hbase-site.xml", + hbase_ui_port=get_env('REPLICA_CLUSTER_PORT'), + cluster_name=cluster2_name) + return active_cluster, replica_cluster + + +def run_put_and_get(cluster: HBaseDockerClient, table: str, row: str, cf: str, data: str): + cluster.put(table, row, cf, data) + cluster.assert_table_row_count(table, expected_row_count=1) + return cluster.assert_get_output(table, row, cf, expected_data=data) + + +def assert_crud_operations_work_on_active_cluster(cluster: HBaseDockerClient): + table = 'crud-test-table1' + cf = 'cf' + row = 'r1' + data = '1' + + # Create + cluster.create_table(table, cf) + cluster.assert_table_exists(table) + + # Retrieve + run_put_and_get(cluster, table, row, cf, data) + + # "Update" (there are no true updates in HBase) + data = '2' + run_put_and_get(cluster, table, row, cf, data) + + # Delete + # This row has two versions. This only deletes the first version + cluster.delete(table, row, column=f"{cf}:") + cluster.assert_table_row_count(table, expected_row_count=1) + cluster.assert_get_output(table, row, cf, expected_data='1') + + # Delete the final version + cluster.delete(table, row, column=f"{cf}:") + cluster.assert_table_row_count(table, expected_row_count=0) + + # Drop table + cluster.disable_table(table) + cluster.drop_table(table) + cluster.assert_table_does_not_exist(table) + + +def assert_correct_active_cluster_suffix(cluster: HBaseDockerClient, data_store_root: str): + logger.info(f"Verifying active cluster suffix file matches 'hbase.meta.table.suffix' " + f"in conf file for {cluster.name}") + active_cluster_file = f'{data_store_root}/data-store/hbase/active.cluster.suffix.id' + active_cluster_suffix = acs.ActiveClusterSuffix() + + # The active cluster suffix file may not get created right away + retries = 0 + while not os.path.exists(active_cluster_file): + if retries >= 5: + raise RuntimeError(f"Timed out waiting for active cluster file to exist: {active_cluster_file}") + logger.info(f"Waiting for active cluster file to exist: {active_cluster_file}") + time.sleep(1) + retries += 1 + + # Parse the active cluster suffix protobuf message file + with open(active_cluster_file, 'rb') as f: + data = f.read() + header = b'PBUF' + if data.startswith(header): + active_cluster_suffix.ParseFromString(data[len(header):]) + else: + active_cluster_suffix.ParseFromString(data) + actual_suffix = active_cluster_suffix.suffix + + # Assume the meta table suffix is blank if hbase.meta.table.suffix does not exist in HBase conf + expected_suffix = cluster.get_hbase_conf_property_value('hbase.meta.table.suffix') + if expected_suffix is None: + expected_suffix = '' + + # Verify the active cluster suffix file has the expected meta table suffix + assert actual_suffix == expected_suffix, (f"Expected {cluster.name} to have meta table suffix '{expected_suffix}', " + f"but got '{actual_suffix}' instead") + + +def reset_cluster_setup(active_cluster: HBaseDockerClient, replica_cluster: HBaseDockerClient, + skip_container_restart: bool, docker_compose_file: str, data_store_root: str, sudo=False): + """ + Resets the Read-Replica cluster setup where one cluster is the active cluster (read-write mode) and the other + cluster is the replica cluster (read-only mode). + """ + if not skip_container_restart: + HBaseDockerClient.stop_containers(docker_compose_file=docker_compose_file, data_dir=data_store_root, sudo=sudo) + + # If the containers are still running, then we need to run update_all_config in the HBase shell to update + # read-only mode on each cluster. Otherwise, we can just modify the conf files and the containers will be restarted + # in the desired read-only mode. + if skip_container_restart: + run_update_all_config = True + else: + run_update_all_config = False + + # First, make sure both clusters are read-only to prevent an error due to trying to have two active clusters + active_cluster.enable_read_only_mode(run_update_all_config=run_update_all_config) + replica_cluster.enable_read_only_mode(run_update_all_config=run_update_all_config) + + # Now activate read-write mode on our active cluster + active_cluster.disable_read_only_mode(run_update_all_config=run_update_all_config) + + if not skip_container_restart: + HBaseDockerClient.start_or_restart_containers(docker_compose_file=docker_compose_file, + data_store_root=f'{data_store_root}') + HBaseDockerClient.wait_for_clusters_to_start([active_cluster, replica_cluster]) + + +def clean_up_tables(active_cluster: HBaseDockerClient, replica_cluster: HBaseDockerClient) -> None: + """ + Drops all tables on the active cluster and then runs 'refresh_meta' on the + read-replica cluster to remove those tables + """ + tables = active_cluster.list_tables() + if tables: + logger.info(f"Removing all existing tables on {active_cluster.name}: {tables}") + for table in tables: + active_cluster.disable_table(table) + active_cluster.drop_table(table) + logger.info(f"Running 'refresh_meta' and 'refresh_hfiles' on {replica_cluster.name} to sync it with " + f"{active_cluster.name}") + replica_cluster.refresh_meta() + replica_cluster.refresh_hfiles() + + +def swap_cluster_roles(new_active_cluster, new_replica_cluster, run_update_all_config=True): + logger.info(f"Making {new_active_cluster.name} the active cluster and " + f"{new_replica_cluster.name} the replica cluster") + new_replica_cluster.enable_read_only_mode(run_update_all_config=run_update_all_config) + new_active_cluster.disable_read_only_mode(run_update_all_config=run_update_all_config) + + +def create_table_on_active_cluster(active_cluster: HBaseDockerClient, column_family: str): + """Create a new table on the active cluster and assert it exists""" + tables = active_cluster.list_tables() + new_table = f't{len(tables)+1}' + active_cluster.create_table(new_table, column_family) + active_cluster.assert_table_exists(new_table) + return new_table + + +def add_data_to_each_table_on_active_cluster(active_cluster: HBaseDockerClient, tables: list, column_family: str): + """Add data to each table in the active cluster""" + for i, table in enumerate(tables[::-1], 1): + active_cluster.put(table, f'r{i}', column_family, i) + active_cluster.flush(table) + + +def refresh_replica_and_verify_tables(replica_cluster: HBaseDockerClient, new_table: str, tables: list): + """ + Refresh meta and HFiles on the replica cluster, and verify the new table + exists and each table has the correct number of rows + """ + replica_cluster.refresh_meta_and_hfiles() + replica_cluster.assert_table_exists(new_table) + for i, table in enumerate(tables[::-1], 1): + replica_cluster.assert_table_row_count(table, i) + + +def create_table_and_test_active_and_replica_clusters(active_cluster: HBaseDockerClient, + replica_cluster: HBaseDockerClient, + column_family: str): + """ + Creates a new table and iteratively adds data to each existing table, including the new one. + Also verifies expected behavior for the replica cluster, such as verifying the new table is not + on the replica before refreshing meta, and then verify new table and data existence after + refreshing meta and HFiles. + """ + new_table = create_table_on_active_cluster(active_cluster, column_family) + + # The new table should not exist on the replica cluster before refreshing meta + replica_cluster.assert_table_does_not_exist(new_table) + + tables = active_cluster.list_tables() + # HBase sorts table list by string: ['t1', 't10', 't2, ..., 't9'] + # We want the list sorted by creation time, so we're sorting on the integer: ['t1', 't2, ..., 't9', 't10'] + tables.sort(key=lambda x: int(x[1:])) + add_data_to_each_table_on_active_cluster(active_cluster, tables, column_family) + refresh_replica_and_verify_tables(replica_cluster, new_table, tables) diff --git a/dev-support/integration-test/read-replica/requirements.txt b/dev-support/integration-test/read-replica/requirements.txt new file mode 100644 index 000000000000..baeeb3f039ab --- /dev/null +++ b/dev-support/integration-test/read-replica/requirements.txt @@ -0,0 +1,12 @@ +certifi==2026.2.25 +charset-normalizer==3.4.7 +dotenv==0.9.9 +grpcio==1.80.0 +grpcio-tools==1.80.0 +idna==3.11 +protobuf==6.33.6 +python-dotenv==1.2.2 +requests==2.33.1 +setuptools==82.0.1 +typing_extensions==4.15.0 +urllib3==2.6.3 diff --git a/dev-support/integration-test/read-replica/utils/bulkload.sh b/dev-support/integration-test/read-replica/utils/bulkload.sh new file mode 100755 index 000000000000..5aec97ab1d25 --- /dev/null +++ b/dev-support/integration-test/read-replica/utils/bulkload.sh @@ -0,0 +1,59 @@ +#!/usr/bin/env bash + +usage() { + echo "Usage: $0 [-n|--num-rows NUM_ROWS] [-i|--initial-row-value INITIAL_ROW_VALUE]" + exit 1 +} + +if [ "$#" -lt 2 ]; then + usage +fi + +TABLE_NAME=$1 +COLUMN_FAMILY=$2 +shift 2 + +NUM_ROWS="" +INITIAL_ROW_VALUE="" + +while [ "$#" -gt 0 ]; do + case "$1" in + -n|--num-rows) + NUM_ROWS="$2" + shift 2 + ;; + -i|--initial-row-value) + INITIAL_ROW_VALUE="$2" + shift 2 + ;; + *) + usage + ;; + esac +done + +TSV_GENERATOR_ARGS="" +if [ -n "$NUM_ROWS" ]; then + TSV_GENERATOR_ARGS="$TSV_GENERATOR_ARGS -n $NUM_ROWS" +fi +if [ -n "$INITIAL_ROW_VALUE" ]; then + TSV_GENERATOR_ARGS="$TSV_GENERATOR_ARGS -i $INITIAL_ROW_VALUE" +fi + +# Clean up any existing bulkload directories +rm -rf /tmp/bulkload + +# Re-create the necessary directory structure +mkdir -p /tmp/bulkload/tsvdata + +# Generate TSV data and save to the specified directory +python3 /opt/utils/tsv_generator.py /tmp/bulkload/tsvdata $TSV_GENERATOR_ARGS + +# Import TSV data to create HFiles for bulk loading +hbase org.apache.hadoop.hbase.mapreduce.ImportTsv \ + -Dimporttsv.columns=HBASE_ROW_KEY,$COLUMN_FAMILY:col0,$COLUMN_FAMILY:col1,$COLUMN_FAMILY:col2,$COLUMN_FAMILY:col3,$COLUMN_FAMILY:col4,$COLUMN_FAMILY:col5,$COLUMN_FAMILY:col6,$COLUMN_FAMILY:col7,$COLUMN_FAMILY:col8,$COLUMN_FAMILY:col9 \ + -Dimporttsv.bulk.output=/tmp/bulkload/HFiles \ + $TABLE_NAME /tmp/bulkload/tsvdata/output.tsv + +# Bulk load the generated HFiles into the HBase table +hbase completebulkload /tmp/bulkload/HFiles $TABLE_NAME diff --git a/dev-support/integration-test/read-replica/utils/tsv_generator.py b/dev-support/integration-test/read-replica/utils/tsv_generator.py new file mode 100644 index 000000000000..72a71d6dac90 --- /dev/null +++ b/dev-support/integration-test/read-replica/utils/tsv_generator.py @@ -0,0 +1,33 @@ +import argparse +import os +import random + +NUM_COLUMNS = 10 + + +def generate_data(row_key): + columns = [str(random.randint(1, 100)) for _ in range(NUM_COLUMNS)] + return f"{row_key}\t" + "\t".join(columns) + "\n" + + +def main(output_dir, num_rows, initial_row_value): + tsv_file = os.path.join(output_dir, "output.tsv") + + rows_written = 0 + with open(tsv_file, "w") as f: + for i in range(initial_row_value, num_rows+initial_row_value): + row_key = f"row{i}" + f.write(generate_data(row_key)) + rows_written += 1 + + print(f"TSV file generated at {tsv_file} with {rows_written} rows and {NUM_COLUMNS} columns.") + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="Generate a TSV file with random data for HBase bulk loading.") + parser.add_argument("output_directory", help="Directory to write the output TSV file") + parser.add_argument("-n", "--num-rows", type=int, default=500, help="Number of rows to generate (default: 500)") + parser.add_argument("-i", "--initial-row-value", type=int, default=0, help="Starting row number (default: 0)") + args = parser.parse_args() + + main(args.output_directory, args.num_rows, args.initial_row_value)