Merge branch 'stable-3.3' into stable-3.4 * stable-3.3: Specify cache.threads = 0 in the multi-site config documentation Change-Id: If053c76ec9f77029103d30c91888b309fa3ef221
diff --git a/BUILD b/BUILD index d7bd3d8..d95dc33 100644 --- a/BUILD +++ b/BUILD
@@ -18,9 +18,10 @@ ], resources = glob(["src/main/resources/**/*"]), deps = [ + ":events-broker-neverlink", + ":global-refdb-neverlink", + ":pull-replication-neverlink", ":replication-neverlink", - "@events-broker//jar", - "@global-refdb//jar", ], ) @@ -30,6 +31,24 @@ exports = ["//plugins/replication"], ) +java_library( + name = "pull-replication-neverlink", + neverlink = 1, + exports = ["//plugins/pull-replication"], +) + +java_library( + name = "events-broker-neverlink", + neverlink = 1, + exports = ["//plugins/events-broker"], +) + +java_library( + name = "global-refdb-neverlink", + neverlink = 1, + exports = ["//plugins/global-refdb"], +) + junit_tests( name = "multi_site_tests", srcs = glob(["src/test/java/**/*.java"]), @@ -49,8 +68,9 @@ visibility = ["//visibility:public"], exports = PLUGIN_DEPS + PLUGIN_TEST_DEPS + [ ":multi-site__plugin", - "@global-refdb//jar", - "@events-broker//jar", + "//plugins/events-broker", + "//plugins/global-refdb", + "//plugins/pull-replication", "//plugins/replication", ], ) @@ -74,12 +94,6 @@ srcs = [ "e2e-tests/test.sh", ], - data = [ - "//plugins/multi-site", - "//plugins/multi-site:e2e_multi_site_test_dir", - "//plugins/multi-site:e2e_multi_site_setup_local_env_dir", - "external_plugin_deps.bzl", - ] + glob(["setup_local_env/**/*"]) + glob(["e2e-tests/**/*"]), args = [ "--multisite-lib-file $(location //plugins/multi-site)", "--healthcheck-interval 5s", @@ -88,6 +102,11 @@ "--location '$(location //plugins/multi-site:e2e_multi_site_test_dir)'", "--local-env '$(location //plugins/multi-site:e2e_multi_site_setup_local_env_dir)'", ], + data = [ + "//plugins/multi-site", + "//plugins/multi-site:e2e_multi_site_test_dir", + "//plugins/multi-site:e2e_multi_site_setup_local_env_dir", + ] + glob(["setup_local_env/**/*"]) + glob(["e2e-tests/**/*"]), tags = [ "e2e-multi-site", ],
diff --git a/Jenkinsfile b/Jenkinsfile index f008749..6ef43af 100644 --- a/Jenkinsfile +++ b/Jenkinsfile
@@ -1,2 +1,4 @@ pluginPipeline(formatCheckId: 'gerritforge:multi-site-format-47168e90078b0b3f11401610930e82830e76bff7', - buildCheckId: 'gerritforge:multi-site-47168e90078b0b3f11401610930e82830e76bff7') + buildCheckId: 'gerritforge:multi-site-47168e90078b0b3f11401610930e82830e76bff7', + extraPlugins: [ 'pull-replication' ], + extraModules: [ 'events-broker', 'global-refdb' ])
diff --git a/README.md b/README.md index 79094d3..f683f4e 100644 --- a/README.md +++ b/README.md
@@ -40,8 +40,6 @@ cd gerrit/plugins ln -s ../../multi-site . -rm external_plugin_deps.bzl -ln -s multi-site/external_plugin_deps.bzl . ``` Example of building the multi-site plugin: @@ -65,6 +63,21 @@ daemon running (/var/run/docker.sock accessible) or a DOCKER_HOST pointing to a Docker server. +## Pre-requisites + +Each Gerrit server of the cluster must be identified with a globally unique +[instance-id](https://gerrit-documentation.storage.googleapis.com/Documentation/3.4.5/config-gerrit.html#gerrit.instanceId) +defined in `$GERRIT_SITE/etc/gerrit.config`. +When migrating from a multi-site configuration with Gerrit v3.3 or earlier, +you must reuse the instance-id value stored under `$GERRIT_SITE/data/multi-site`. + +Example: + +``` +[gerrit] + instanceId = 758fe5b7-1869-46e6-942a-3ae0ae7e3bd2 +``` + ## How to configure Install the multi-site plugin into the `$GERRIT_SITE/lib` directory of all
diff --git a/e2e-tests/test.sh b/e2e-tests/test.sh index 19c8b04..3ecb9fd 100755 --- a/e2e-tests/test.sh +++ b/e2e-tests/test.sh
@@ -16,11 +16,11 @@ LOCATION="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )" LOCAL_ENV="$( cd "${LOCATION}/../setup_local_env" >/dev/null 2>&1 && pwd )" -GERRIT_BRANCH=stable-3.3 +GERRIT_BRANCH=stable-3.4 GERRIT_CI=https://archive-ci.gerritforge.com/view/Plugins-$GERRIT_BRANCH/job LAST_BUILD=lastSuccessfulBuild/artifact/bazel-bin/plugins DEF_MULTISITE_LOCATION=${LOCATION}/../../../bazel-bin/plugins/multi-site/multi-site.jar -DEF_GERRIT_IMAGE=3.3.6-centos8 +DEF_GERRIT_IMAGE=3.4.0-centos8 DEF_GERRIT_HEALTHCHECK_START_PERIOD=60s DEF_GERRIT_HEALTHCHECK_INTERVAL=5s DEF_GERRIT_HEALTHCHECK_TIMEOUT=5s @@ -64,7 +64,7 @@ export BROKER_HOST=$2 export BROKER_PORT=$3 - export GROUP_ID=$4 + export INSTANCE_ID=$4 export SSH_ADVERTISED_PORT=$5 export LOCATION_TEST_SITE=/var/gerrit export REMOTE_DEBUG_PORT=5005 @@ -196,8 +196,6 @@ done # Defaults -EVENTS_BROKER_VER=`grep 'com.gerritforge:events-broker' ${LOCATION}/../external_plugin_deps.bzl | cut -d '"' -f 2 | cut -d ':' -f 3` -GLOBAL_REFDB_VER=`grep 'com.gerritforge:global-refdb' ${LOCATION}/../external_plugin_deps.bzl | cut -d '"' -f 2 | cut -d ':' -f 3` DEPLOYMENT_LOCATION=$(mktemp -d || $(echo >&2 "Could not create temp dir" && exit 1)) MULTISITE_LIB_LOCATION=${MULTISITE_LIB_LOCATION:-${DEF_MULTISITE_LOCATION}} BROKER_TYPE=${BROKER_TYPE:-"kafka"} @@ -261,13 +259,11 @@ docker cp ${CONTAINER_NAME}:/var/gerrit/plugins/replication.jar $COMMON_LIBS/ docker rm -fv ${CONTAINER_NAME} -echo "Downloading global-refdb library $GERRIT_BRANCH" -wget https://repo1.maven.org/maven2/com/gerritforge/global-refdb/$GLOBAL_REFDB_VER/global-refdb-$GLOBAL_REFDB_VER.jar \ - -O $COMMON_LIBS/global-refdb.jar || { echo >&2 "Cannot download global-refdb library: Check internet connection. Aborting"; exit 1; } +echo "Copying global-refdb library $GERRIT_BRANCH" +cp bazel-bin/plugins/global-refdb/global-refdb.jar $COMMON_LIBS/global-refdb.jar echo "Downloading events-broker library $GERRIT_BRANCH" -wget https://repo1.maven.org/maven2/com/gerritforge/events-broker/$EVENTS_BROKER_VER/events-broker-$EVENTS_BROKER_VER.jar \ - -O $COMMON_LIBS/events-broker.jar || { echo >&2 "Cannot download events-broker library: Check internet connection. Aborting"; exit 1; } +cp bazel-bin/plugins/events-broker/events-broker.jar $COMMON_LIBS/events-broker.jar echo "Setting up directories" mkdir -p ${GERRIT_1_ETC} ${GERRIT_1_PLUGINS} ${GERRIT_1_LIBS} ${GERRIT_2_ETC} ${GERRIT_2_PLUGINS} ${GERRIT_2_LIBS}
diff --git a/external_plugin_deps.bzl b/external_plugin_deps.bzl deleted file mode 100644 index 223778e..0000000 --- a/external_plugin_deps.bzl +++ /dev/null
@@ -1,14 +0,0 @@ -load("//tools/bzl:maven_jar.bzl", "maven_jar") - -def external_plugin_deps(): - maven_jar( - name = "global-refdb", - artifact = "com.gerritforge:global-refdb:3.3.1", - sha1 = "5df9dddad2fc67c922406f41549186b210cd957e", - ) - - maven_jar( - name = "events-broker", - artifact = "com.gerritforge:events-broker:3.3.2", - sha1 = "d8bcb77047cc12dd7c623b5b4de70a25499d3d6c", - )
diff --git a/setup_local_env/README.md b/setup_local_env/README.md index 88b40d5..e6ec9e7 100644 --- a/setup_local_env/README.md +++ b/setup_local_env/README.md
@@ -74,8 +74,6 @@ [--gerrit2-httpd-port] Gerrit Instance 2 http port; default 18081 [--gerrit2-sshd-port] Gerrit Instance 2 sshd port; default 49418 -[--replication-type] Options [file,ssh]; default file -[--replication-ssh-user] SSH user for the replication plugin; default $(whoami) [--replication-delay] Replication delay across the two instances in seconds [--just-cleanup-env] Cleans up previous deployment; default false
diff --git a/setup_local_env/configs/gerrit.config b/setup_local_env/configs/gerrit.config index 2cd8c9d..8a176f2 100644 --- a/setup_local_env/configs/gerrit.config +++ b/setup_local_env/configs/gerrit.config
@@ -5,6 +5,7 @@ installModule = com.gerritforge.gerrit.eventbroker.BrokerApiModule # events-broker module to setup BrokerApi dynamic item installModule = com.googlesource.gerrit.plugins.multisite.Module # multi-site needs to be a gerrit lib installDbModule = com.googlesource.gerrit.plugins.multisite.GitModule + instanceId = $INSTANCE_ID [database] type = h2 database = $LOCATION_TEST_SITE/db/ReviewDB @@ -43,7 +44,7 @@ [plugin "events-kafka"] sendAsync = true bootstrapServers = $BROKER_HOST:$BROKER_PORT - groupId = $GROUP_ID + groupId = $INSTANCE_ID numberOfSubscribers = 6 securityProtocol = PLAINTEXT pollingIntervalMs = 1000 @@ -55,12 +56,12 @@ pollingIntervalMs = 1000 region = us-east-1 endpoint = http://localhost:$BROKER_PORT - applicationName = $GROUP_ID + applicationName = $INSTANCE_ID initialPosition = trim_horizon [plugin "events-gcloud-pubsub"] numberOfSubscribers = 6 gcloudProject="test-project" - subscriptionId=$GROUP_ID + subscriptionId=$INSTANCE_ID privateKeyLocation="not used in local mode" [plugin "metrics-reporter-prometheus"] prometheusBearerToken = token
diff --git a/setup_local_env/configs/replication.config b/setup_local_env/configs/replication.config index ece1b3e..1c2aac4 100644 --- a/setup_local_env/configs/replication.config +++ b/setup_local_env/configs/replication.config
@@ -1,6 +1,7 @@ -[remote "Replication"] - $REPLICATION_URL - push = +refs/*:refs/* +[remote "$REPLICA_INSTANCE_ID"] + $PULL_REPLICATION_URL + $PULL_REPLICATION_API_URL + fetch = +refs/*:refs/* mirror = true timeout = 600 rescheduleDelay = 15 @@ -13,4 +14,8 @@ replicateOnStartup = false [replication] lockErrorMaxRetries = 5 - maxRetries = 5 \ No newline at end of file + maxRetries = 5 + useCGitClient = false + consumeStreamEvents = false + syncRefs="ALL REFS ASYNC" + maxApiPayloadSize=40000
diff --git a/setup_local_env/configs/secure.config b/setup_local_env/configs/secure.config new file mode 100644 index 0000000..faeb659 --- /dev/null +++ b/setup_local_env/configs/secure.config
@@ -0,0 +1,3 @@ +[auth] + registerEmailPrivateKey = A3D00Ny0bLW2z7xdInvIf35Q5mDCi5u9o/E= + bearertoken = some-bearer-token
diff --git a/setup_local_env/setup.sh b/setup_local_env/setup.sh index 11ae04f..247a721 100755 --- a/setup_local_env/setup.sh +++ b/setup_local_env/setup.sh
@@ -16,17 +16,15 @@ SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" >/dev/null 2>&1 && pwd )" -GERRIT_BRANCH=stable-3.3 +GERRIT_BRANCH=stable-3.4 GERRIT_CI=https://archive-ci.gerritforge.com/view/Plugins-$GERRIT_BRANCH/job LAST_BUILD=lastSuccessfulBuild/artifact/bazel-bin/plugins -EVENTS_BROKER_VER=`grep 'com.gerritforge:events-broker' $(dirname $0)/../external_plugin_deps.bzl | cut -d '"' -f 2 | cut -d ':' -f 3` -GLOBAL_REFDB_VER=`grep 'com.gerritforge:global-refdb' $(dirname $0)/../external_plugin_deps.bzl | cut -d '"' -f 2 | cut -d ':' -f 3` function check_application_requirements { type haproxy >/dev/null 2>&1 || { echo >&2 "Require haproxy but it's not installed. Aborting."; exit 1; } type java >/dev/null 2>&1 || { echo >&2 "Require java but it's not installed. Aborting."; exit 1; } type docker >/dev/null 2>&1 || { echo >&2 "Require docker but it's not installed. Aborting."; exit 1; } - type docker-compose >/dev/null 2>&1 || { echo >&2 "Require docker-compose but it's not installed. Aborting."; exit 1; } + [ $($SUDO docker --version | awk '{print $3}' | cut -d '.' -f 1) -ge 20 ] || { echo >&2 "Require docker v20 or later. Aborting."; exit 1; } type wget >/dev/null 2>&1 || { echo >&2 "Require wget but it's not installed. Aborting."; exit 1; } type envsubst >/dev/null 2>&1 || { echo >&2 "Require envsubst but it's not installed. Aborting."; exit 1; } type openssl >/dev/null 2>&1 || { echo >&2 "Require openssl but it's not installed. Aborting."; exit 1; } @@ -35,16 +33,16 @@ fi } -function get_replication_url { - REPLICATION_LOCATION_TEST_SITE=$1 - REPLICATION_HOSTNAME=$2 - USER=$REPLICATION_SSH_USER +function get_pull_replication_api_url { + REPLICATION_HOSTNAME=$1 - if [ "$REPLICATION_TYPE" = "file" ];then - echo "url = file://$REPLICATION_LOCATION_TEST_SITE/git/#{name}#.git" - elif [ "$REPLICATION_TYPE" = "ssh" ];then - echo "url = ssh://$USER@$REPLICATION_HOSTNAME:$REPLICATION_LOCATION_TEST_SITE/git/#{name}#.git" - fi + echo "apiUrl = http://$REPLICATION_HOSTNAME:$REPLICATION_HTTPD_PORT" +} + +function get_pull_replication_url { + REPLICATION_HOSTNAME=$1 + + echo "url = http://$REPLICATION_HOSTNAME:$REPLICATION_HTTPD_PORT/#{name}#.git" } function deploy_tls_certificates { @@ -66,12 +64,13 @@ export LOCATION_TEST_SITE=$3 export GERRIT_SSHD_PORT=$4 export REPLICATION_HTTPD_PORT=$5 - export REPLICATION_LOCATION_TEST_SITE=$6 - export GERRIT_HOSTNAME=$7 - export REPLICATION_HOSTNAME=$8 - export REMOTE_DEBUG_PORT=$9 - export GROUP_ID=${10} - export REPLICATION_URL=$(get_replication_url $REPLICATION_LOCATION_TEST_SITE $REPLICATION_HOSTNAME) + export GERRIT_HOSTNAME=$6 + export REPLICATION_HOSTNAME=$7 + export REMOTE_DEBUG_PORT=$8 + export INSTANCE_ID=${9} + export REPLICA_INSTANCE_ID=${10} + export PULL_REPLICATION_URL=$(get_pull_replication_url $REPLICATION_HOSTNAME) + export PULL_REPLICATION_API_URL=$(get_pull_replication_api_url $REPLICATION_HOSTNAME) echo "Replacing variables for file $file and copying to $CONFIG_TEST_SITE/$file_name" @@ -126,25 +125,25 @@ GERRIT_SITE1_SSHD_PORT=$3 CONFIG_TEST_SITE_1=$LOCATION_TEST_SITE_1/etc GERRIT_SITE1_REMOTE_DEBUG_PORT="5005" - GERRIT_SITE1_GROUP_ID="instance-1" + GERRIT_SITE1_INSTANCE_ID="instance-1" # SITE 2 GERRIT_SITE2_HOSTNAME=$4 GERRIT_SITE2_HTTPD_PORT=$5 GERRIT_SITE2_SSHD_PORT=$6 CONFIG_TEST_SITE_2=$LOCATION_TEST_SITE_2/etc GERRIT_SITE2_REMOTE_DEBUG_PORT="5006" - GERRIT_SITE2_GROUP_ID="instance-2" + GERRIT_SITE2_INSTANCE_ID="instance-2" # Set config SITE1 - copy_config_files $CONFIG_TEST_SITE_1 $GERRIT_SITE1_HTTPD_PORT $LOCATION_TEST_SITE_1 $GERRIT_SITE1_SSHD_PORT $GERRIT_SITE2_HTTPD_PORT $LOCATION_TEST_SITE_2 $GERRIT_SITE1_HOSTNAME $GERRIT_SITE2_HOSTNAME $GERRIT_SITE1_REMOTE_DEBUG_PORT $GERRIT_SITE1_GROUP_ID + copy_config_files $CONFIG_TEST_SITE_1 $GERRIT_SITE1_HTTPD_PORT $LOCATION_TEST_SITE_1 $GERRIT_SITE1_SSHD_PORT $GERRIT_SITE2_HTTPD_PORT $GERRIT_SITE1_HOSTNAME $GERRIT_SITE2_HOSTNAME $GERRIT_SITE1_REMOTE_DEBUG_PORT $GERRIT_SITE1_INSTANCE_ID $GERRIT_SITE2_INSTANCE_ID # Set config SITE2 - copy_config_files $CONFIG_TEST_SITE_2 $GERRIT_SITE2_HTTPD_PORT $LOCATION_TEST_SITE_2 $GERRIT_SITE2_SSHD_PORT $GERRIT_SITE1_HTTPD_PORT $LOCATION_TEST_SITE_1 $GERRIT_SITE1_HOSTNAME $GERRIT_SITE2_HOSTNAME $GERRIT_SITE2_REMOTE_DEBUG_PORT $GERRIT_SITE2_GROUP_ID + copy_config_files $CONFIG_TEST_SITE_2 $GERRIT_SITE2_HTTPD_PORT $LOCATION_TEST_SITE_2 $GERRIT_SITE2_SSHD_PORT $GERRIT_SITE1_HTTPD_PORT $GERRIT_SITE1_HOSTNAME $GERRIT_SITE2_HOSTNAME $GERRIT_SITE2_REMOTE_DEBUG_PORT $GERRIT_SITE2_INSTANCE_ID $GERRIT_SITE1_INSTANCE_ID } function is_docker_desktop { - echo $(docker info | grep "Operating System: Docker Desktop" | wc -l) + echo $($SUDO docker info | grep "Operating System: Docker Desktop" | wc -l) } function docker_host_env { @@ -183,9 +182,10 @@ kill $(ps -ax | grep haproxy | grep "gerrit_setup/ha-proxy-config" | awk '{print $1}') 2> /dev/null echo "Stopping $BROKER_TYPE docker container" - docker-compose -f "${SCRIPT_DIR}/docker-compose-${BROKER_TYPE}.yaml" down 2> /dev/null + printenv > "${SCRIPT_DIR}/docker-compose-${BROKER_TYPE}.env" + $SUDO docker compose -f "${SCRIPT_DIR}/docker-compose-${BROKER_TYPE}.yaml" --env-file "${SCRIPT_DIR}/docker-compose-${BROKER_TYPE}.env" down 2> /dev/null echo "Stopping core docker containers" - docker-compose -f "${SCRIPT_DIR}/docker-compose-core.yaml" down 2> /dev/null + $SUDO docker compose -f "${SCRIPT_DIR}/docker-compose-core.yaml" --env-file "${SCRIPT_DIR}/docker-compose-${BROKER_TYPE}.env" down 2> /dev/null echo "Stopping GERRIT instances" $1/bin/gerrit.sh stop 2> /dev/null @@ -197,7 +197,7 @@ function check_if_container_is_running { local container=$1; - echo $(docker inspect "$container" 2> /dev/null | grep '"Running": true' | wc -l) + echo $($SUDO docker inspect "$container" 2> /dev/null | grep '"Running": true' | wc -l) } function ensure_docker_compose_is_up_and_running { @@ -207,8 +207,9 @@ local is_container_running=$(check_if_container_is_running "$container_name") if [ "$is_container_running" -lt 1 ];then + printenv > "${SCRIPT_DIR}/${docker_compose_file}.env" echo "[$log_label] Starting docker containers" - docker-compose -f "${SCRIPT_DIR}/${docker_compose_file}" up -d + $SUDO docker compose -f "${SCRIPT_DIR}/${docker_compose_file}" --env-file "${SCRIPT_DIR}/${docker_compose_file}.env" up -d echo "[$log_label] Waiting for docker containers to start..." while [[ $(check_if_container_is_running "$container_name") -lt 1 ]];do sleep 10s; done @@ -231,6 +232,8 @@ echo echo "[--release-war-file] Location to release.war file" echo "[--multisite-lib-file] Location to lib multi-site.jar file" + echo "[--eventsbroker-lib-file] Location to lib events-broker.jar file" + echo "[--globalrefdb-lib-file] Location to lib global-refdb.jar file" echo echo "[--new-deployment] Cleans up previous gerrit deployment and re-installs it. default true" echo "[--get-websession-plugin] Download websession-broker plugin from CI lastSuccessfulBuild; default true" @@ -247,8 +250,6 @@ echo "[--gerrit2-httpd-port] Gerrit Instance 2 http port; default 18081" echo "[--gerrit2-sshd-port] Gerrit Instance 2 sshd port; default 49418" echo - echo "[--replication-type] Options [file,ssh]; default ssh" - echo "[--replication-ssh-user] SSH user for the replication plugin; default $(whoami)" echo "[--replication-delay] Replication delay across the two instances in seconds" echo echo "[--just-cleanup-env] Cleans up previous deployment; default false" @@ -257,6 +258,8 @@ echo echo "[--broker-type] events broker type; 'kafka', 'kinesis' or 'gcloud-pubsub'. Default 'kafka'" echo + echo "[--sudo] run docker commands with sudo" + echo exit 0 ;; "--new-deployment") @@ -284,6 +287,16 @@ shift shift ;; + "--eventsbroker-lib-file" ) + EVENTS_BROKER_LIB_LOCATION=$2 + shift + shift + ;; + "--globalrefdb-lib-file" ) + GLOBAL_REFDB_LIB_LOCATION=$2 + shift + shift + ;; "--gerrit-canonical-host" ) export GERRIT_CANONICAL_HOSTNAME=$2 shift @@ -319,16 +332,6 @@ shift shift ;; - "--replication-ssh-user" ) - export REPLICATION_SSH_USER=$2 - shift - shift - ;; - "--replication-type") - export REPLICATION_TYPE=$2 - shift - shift - ;; "--replication-delay") export REPLICATION_DELAY_SEC=$2 shift @@ -353,6 +356,11 @@ exit 1 fi ;; + "--sudo" ) + SUDO=sudo + shift + shift + ;; * ) echo "Unknown option argument: $1" shift @@ -376,8 +384,6 @@ GERRIT_2_HTTPD_PORT=${GERRIT_2_HTTPD_PORT:-"18081"} GERRIT_1_SSHD_PORT=${GERRIT_1_SSHD_PORT:-"39418"} GERRIT_2_SSHD_PORT=${GERRIT_2_SSHD_PORT:-"49418"} -REPLICATION_TYPE=${REPLICATION_TYPE:-"file"} -REPLICATION_SSH_USER=${REPLICATION_SSH_USER:-$(whoami)} export REPLICATION_DELAY_SEC=${REPLICATION_DELAY_SEC:-"5"} export SSH_ADVERTISED_PORT=${SSH_ADVERTISED_PORT:-"29418"} HTTPS_ENABLED=${HTTPS_ENABLED:-"false"} @@ -392,7 +398,8 @@ RELEASE_WAR_FILE_LOCATION=${RELEASE_WAR_FILE_LOCATION:-bazel-bin/release.war} MULTISITE_LIB_LOCATION=${MULTISITE_LIB_LOCATION:-bazel-bin/plugins/multi-site/multi-site.jar} - +EVENTS_BROKER_LIB_LOCATION=${EVENTS_BROKER_LIB_LOCATION:-bazel-bin/plugins/events-broker/events-broker.jar} +GLOBAL_REFDB_LIB_LOCATION=${GLOBAL_REFDB_LIB_LOCATION:-bazel-bin/plugins/global-refdb/global-refdb.jar} export FAKE_NFS=$COMMON_LOCATION/fake_nfs @@ -413,10 +420,20 @@ else cp -f $MULTISITE_LIB_LOCATION $DEPLOYMENT_LOCATION/multi-site.jar >/dev/null 2>&1 || { echo >&2 "$MULTISITE_LIB_LOCATION: Not able to copy the file. Aborting"; exit 1; } fi + +echo "Copying events-broker library" + cp -f $EVENTS_BROKER_LIB_LOCATION $DEPLOYMENT_LOCATION/events-broker.jar >/dev/null 2>&1 || { echo >&2 "$EVENTS_BROKER_LIB_LOCATION: Not able to copy the file. Aborting"; exit 1; } + +echo "Copying global-refdb library" + cp -f $GLOBAL_REFDB_LIB_LOCATION $DEPLOYMENT_LOCATION/global-refdb.jar >/dev/null 2>&1 || { echo >&2 "$GLOBAL_REFDB_LIB_LOCATION: Not able to copy the file. Aborting"; exit 1; } + if [ $DOWNLOAD_WEBSESSION_PLUGIN = "true" ];then echo "Downloading websession-broker plugin $GERRIT_BRANCH" wget $GERRIT_CI/plugin-websession-broker-bazel-$GERRIT_BRANCH/$LAST_BUILD/websession-broker/websession-broker.jar \ - -O $DEPLOYMENT_LOCATION/websession-broker.jar || { echo >&2 "Cannot download websession-broker plugin: Check internet connection. Abort\ + -O $DEPLOYMENT_LOCATION/websession-broker.jar || \ + wget $GERRIT_CI/plugin-websession-broker-bazel-master-$GERRIT_BRANCH/$LAST_BUILD/websession-broker/websession-broker.jar \ + -O $DEPLOYMENT_LOCATION/websession-broker.jar || \ + { echo >&2 "Cannot download websession-broker plugin: Check internet connection. Abort\ ing"; exit 1; } wget $GERRIT_CI/plugin-healthcheck-bazel-$GERRIT_BRANCH/$LAST_BUILD/healthcheck/healthcheck.jar \ -O $DEPLOYMENT_LOCATION/healthcheck.jar || { echo >&2 "Cannot download healthcheck plugin: Check internet connection. Abort\ @@ -427,30 +444,29 @@ echo "Downloading zookeeper plugin $GERRIT_BRANCH" wget $GERRIT_CI/plugin-zookeeper-refdb-bazel-$GERRIT_BRANCH/$LAST_BUILD/zookeeper-refdb/zookeeper-refdb.jar \ - -O $DEPLOYMENT_LOCATION/zookeeper-refdb.jar || { echo >&2 "Cannot download zookeeper plugin: Check internet connection. Abort\ -ing"; exit 1; } - -echo "Downloading global-refdb library $GERRIT_BRANCH" - wget https://repo1.maven.org/maven2/com/gerritforge/global-refdb/$GLOBAL_REFDB_VER/global-refdb-$GLOBAL_REFDB_VER.jar \ - -O $DEPLOYMENT_LOCATION/global-refdb.jar || { echo >&2 "Cannot download global-refdb library: Check internet connection. Abort\ -ing"; exit 1; } - -echo "Downloading events-broker library $GERRIT_BRANCH" - wget https://repo1.maven.org/maven2/com/gerritforge/events-broker/$EVENTS_BROKER_VER/events-broker-$EVENTS_BROKER_VER.jar \ - -O $DEPLOYMENT_LOCATION/events-broker.jar || { echo >&2 "Cannot download events-broker library: Check internet connection. Abort\ + -O $DEPLOYMENT_LOCATION/zookeeper-refdb.jar || \ + wget $GERRIT_CI/plugin-zookeeper-refdb-bazel-master-$GERRIT_BRANCH/$LAST_BUILD/zookeeper-refdb/zookeeper-refdb.jar \ + -O $DEPLOYMENT_LOCATION/zookeeper-refdb.jar || \ + { echo >&2 "Cannot download zookeeper plugin: Check internet connection. Abort\ ing"; exit 1; } if [ "$BROKER_TYPE" = "kafka" ]; then echo "Downloading events-kafka plugin $GERRIT_BRANCH" wget $GERRIT_CI/plugin-events-kafka-bazel-$GERRIT_BRANCH/$LAST_BUILD/events-kafka/events-kafka.jar \ - -O $DEPLOYMENT_LOCATION/events-kafka.jar || { echo >&2 "Cannot download events-kafka plugin: Check internet connection. Abort\ + -O $DEPLOYMENT_LOCATION/events-kafka.jar || \ + wget $GERRIT_CI/plugin-events-kafka-bazel-master-$GERRIT_BRANCH/$LAST_BUILD/events-kafka/events-kafka.jar \ + -O $DEPLOYMENT_LOCATION/events-kafka.jar || \ + { echo >&2 "Cannot download events-kafka plugin: Check internet connection. Abort\ ing"; exit 1; } fi if [ "$BROKER_TYPE" = "kinesis" ]; then echo "Downloading events-aws-kinesis plugin $GERRIT_BRANCH" wget $GERRIT_CI/plugin-events-aws-kinesis-bazel-$GERRIT_BRANCH/$LAST_BUILD/events-aws-kinesis/events-aws-kinesis.jar \ - -O $DEPLOYMENT_LOCATION/events-aws-kinesis.jar || { echo >&2 "Cannot download events-aws-kinesis plugin: Check internet connection. Abort\ + -O $DEPLOYMENT_LOCATION/events-aws-kinesis.jar || \ + wget $GERRIT_CI/plugin-events-aws-kinesis-bazel-master-$GERRIT_BRANCH/$LAST_BUILD/events-aws-kinesis/events-aws-kinesis.jar \ + -O $DEPLOYMENT_LOCATION/events-aws-kinesis.jar || \ + { echo >&2 "Cannot download events-aws-kinesis plugin: Check internet connection. Abort\ ing"; exit 1; } fi @@ -458,19 +474,25 @@ if [ "$BROKER_TYPE" = "gcloud-pubsub" ]; then echo "Downloading events-gcloud-pubsub plugin $GERRIT_BRANCH" wget $GERRIT_CI/plugin-events-gcloud-pubsub-bazel-$GERRIT_BRANCH/$LAST_BUILD/events-gcloud-pubsub/events-gcloud-pubsub.jar \ - -O $DEPLOYMENT_LOCATION/events-gcloud-pubsub.jar || { echo >&2 "Cannot download events-gcloud-pubsub plugin: Check internet connection. Abort\ + -O $DEPLOYMENT_LOCATION/events-gcloud-pubsub.jar || \ + wget $GERRIT_CI/plugin-events-gcloud-pubsub-bazel-master-$GERRIT_BRANCH/$LAST_BUILD/events-gcloud-pubsub/events-gcloud-pubsub.jar \ + -O $DEPLOYMENT_LOCATION/events-gcloud-pubsub.jar || \ + { echo >&2 "Cannot download events-gcloud-pubsub plugin: Check internet connection. Abort\ ing"; exit 1; } fi echo "Downloading metrics-reporter-prometheus plugin $GERRIT_BRANCH" + wget $GERRIT_CI/plugin-metrics-reporter-prometheus-bazel-$GERRIT_BRANCH/$LAST_BUILD/metrics-reporter-prometheus/metrics-reporter-prometheus.jar \ + -O $DEPLOYMENT_LOCATION/metrics-reporter-prometheus.jar || \ wget $GERRIT_CI/plugin-metrics-reporter-prometheus-bazel-master-$GERRIT_BRANCH/$LAST_BUILD/metrics-reporter-prometheus/metrics-reporter-prometheus.jar \ - -O $DEPLOYMENT_LOCATION/metrics-reporter-prometheus.jar || { echo >&2 "Cannot download metrics-reporter-prometheus plugin: Check internet connection. Abort\ + -O $DEPLOYMENT_LOCATION/metrics-reporter-prometheus.jar || \ + { echo >&2 "Cannot download metrics-reporter-prometheus plugin: Check internet connection. Abort\ ing"; exit 1; } -if [ "$REPLICATION_TYPE" = "ssh" ];then - echo "Using 'SSH' replication type" - echo "Make sure ~/.ssh/authorized_keys and ~/.ssh/known_hosts are configured correctly" -fi +echo "Downloading pull-replication plugin $GERRIT_BRANCH" + wget $GERRIT_CI/plugin-pull-replication-bazel-$GERRIT_BRANCH/$LAST_BUILD/pull-replication/pull-replication.jar \ + -O $DEPLOYMENT_LOCATION/pull-replication.jar || { echo >&2 "Cannot download pull-replication plugin: Check internet connection. Abort\ +ing"; exit 1; } if [ "$HTTPS_ENABLED" = "true" ];then export HTTP_PROTOCOL="https" @@ -530,13 +552,18 @@ echo "Replicating environment" cp -fR $LOCATION_TEST_SITE_1/* $LOCATION_TEST_SITE_2 - echo "Link replication plugin" - ln -s $LOCATION_TEST_SITE_1/plugins/replication.jar $LOCATION_TEST_SITE_1/lib/replication.jar - ln -s $LOCATION_TEST_SITE_2/plugins/replication.jar $LOCATION_TEST_SITE_2/lib/replication.jar + echo "Link pullreplication plugin" + ln -s $LOCATION_TEST_SITE_1/plugins/pull-replication.jar $LOCATION_TEST_SITE_1/lib/pull-replication.jar + ln -s $LOCATION_TEST_SITE_2/plugins/pull-replication.jar $LOCATION_TEST_SITE_2/lib/pull-replication.jar echo "Link multi-site library to plugin directory" ln -s $LOCATION_TEST_SITE_1/lib/multi-site.jar $LOCATION_TEST_SITE_1/plugins/multi-site.jar ln -s $LOCATION_TEST_SITE_2/lib/multi-site.jar $LOCATION_TEST_SITE_2/plugins/multi-site.jar + + echo "Copy pull-replication plugin" + cp -f $DEPLOYMENT_LOCATION/pull-replication.jar $LOCATION_TEST_SITE_1/plugins/pull-replication.jar + cp -f $DEPLOYMENT_LOCATION/pull-replication.jar $LOCATION_TEST_SITE_2/plugins/pull-replication.jar + fi DOCKER_HOST_ENV=$(docker_host_env) @@ -558,8 +585,12 @@ echo "Re-deploying configuration files" deploy_config_files $GERRIT_1_HOSTNAME $GERRIT_1_HTTPD_PORT $GERRIT_1_SSHD_PORT $GERRIT_2_HOSTNAME $GERRIT_2_HTTPD_PORT $GERRIT_2_SSHD_PORT +echo "Remove replication plugin from gerrit site 1" +rm $LOCATION_TEST_SITE_1/plugins/replication.jar echo "Starting gerrit site 1" $LOCATION_TEST_SITE_1/bin/gerrit.sh restart +echo "Remove replication plugin from gerrit site 2" +rm $LOCATION_TEST_SITE_2/plugins/replication.jar echo "Starting gerrit site 2" $LOCATION_TEST_SITE_2/bin/gerrit.sh restart @@ -574,8 +605,6 @@ echo "===============================" echo "The admin password is 'secret'" echo "deployment-location=$DEPLOYMENT_LOCATION" -echo "replication-type=$REPLICATION_TYPE" -echo "replication-ssh-user=$REPLICATION_SSH_USER" echo "replication-delay=$REPLICATION_DELAY_SEC" echo "enable-https=$HTTPS_ENABLED" echo
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java index 6b77713..2d67eab 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/Configuration.java
@@ -23,15 +23,19 @@ import com.google.common.base.Strings; import com.google.common.base.Supplier; import com.google.common.collect.ImmutableList; +import com.google.gerrit.server.config.ConfigUtil; import com.google.gerrit.server.config.SitePaths; import com.google.inject.Inject; import com.google.inject.Singleton; import com.google.inject.spi.Message; import java.io.IOException; +import java.time.Duration; import java.util.Arrays; import java.util.Collection; import java.util.Collections; import java.util.List; +import java.util.Random; +import java.util.concurrent.TimeUnit; import org.eclipse.jgit.errors.ConfigInvalidException; import org.eclipse.jgit.lib.Config; import org.eclipse.jgit.storage.file.FileBasedConfig; @@ -50,6 +54,10 @@ static final String THREAD_POOL_SIZE_KEY = "threadPoolSize"; static final int DEFAULT_THREAD_POOL_SIZE = 4; + private static final String REF_DATABASE = "ref-database"; + private static final String REPLICATION_LAG_REFRESH_INTERVAL = "replicationLagRefreshInterval"; + private static final Duration REPLICATION_LAG_REFRESH_INTERVAL_DEFAULT = Duration.ofSeconds(60); + private static final String REPLICATION_CONFIG = "replication.config"; // common parameters to cache and index sections private static final int DEFAULT_INDEX_MAX_TRIES = 2; @@ -64,7 +72,9 @@ private final Supplier<SharedRefDbConfiguration> sharedRefDb; private final Supplier<Collection<Message>> replicationConfigValidation; private final Supplier<Broker> broker; + private final Supplier<ReplicationFilter> replicationFilter; private final Config multiSiteConfig; + private final Supplier<Duration> replicationLagRefreshInterval; @Inject Configuration(SitePaths sitePaths) { @@ -86,6 +96,18 @@ new SharedRefDbConfiguration( enableSharedRefDbByDefault(lazyMultiSiteCfg.get()), PLUGIN_NAME)); broker = memoize(() -> new Broker(lazyMultiSiteCfg)); + replicationFilter = memoize(() -> new ReplicationFilter(lazyMultiSiteCfg)); + replicationLagRefreshInterval = + memoize( + () -> + Duration.ofMillis( + ConfigUtil.getTimeUnit( + lazyMultiSiteCfg.get(), + REF_DATABASE, + null, + REPLICATION_LAG_REFRESH_INTERVAL, + REPLICATION_LAG_REFRESH_INTERVAL_DEFAULT.toMillis(), + TimeUnit.MILLISECONDS))); } public Config getMultiSiteConfig() { @@ -116,6 +138,14 @@ return projects.get(); } + public ReplicationFilter replicationFilter() { + return replicationFilter.get(); + } + + public Duration replicationLagRefreshInterval() { + return replicationLagRefreshInterval.get(); + } + public Collection<Message> validate() { return replicationConfigValidation.get(); } @@ -311,6 +341,81 @@ } } + public static class ReplicationFilter { + static final String REPLICATION_FILTER_SECTION = "replication"; + static final String REPLICATION_PUSH_FILTER_SUBSECTION = "push-filter"; + static final String REPLICATION_FETCH_FILTER_SUBSECTION = "fetch-filter"; + + private static final String MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS = + "minWaitBeforeReloadLocalVersionMs"; + private static final String RANDOM_WAIT_MAX_BOUND_BEFORE_RELOAD_LOCAL_VERSION_MS = + "maxRandomWaitBeforeReloadLocalVersionMs"; + + private static final int MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS_DEFAULT = 1000; + private static final int RANDOM_WAIT_MAX_BOUND_BEFORE_RELOAD_LOCAL_VERSION_MS_DEFAULT = 1000; + private final Supplier<Integer> fetchMinWaitBeforeReloadLocalVersionMs; + private final Supplier<Integer> fetchWaitBeforeReloadLocalVersionMs; + private final Supplier<Integer> pushMinWaitBeforeReloadLocalVersionMs; + private final Supplier<Integer> pushWaitBeforeReloadLocalVersionMs; + + public ReplicationFilter(Supplier<Config> cfg) { + fetchMinWaitBeforeReloadLocalVersionMs = + memoize( + () -> + cfg.get() + .getInt( + REPLICATION_FILTER_SECTION, + REPLICATION_FETCH_FILTER_SUBSECTION, + MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS, + MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS_DEFAULT)); + fetchWaitBeforeReloadLocalVersionMs = + memoize( + () -> + cfg.get() + .getInt( + REPLICATION_FILTER_SECTION, + REPLICATION_FETCH_FILTER_SUBSECTION, + RANDOM_WAIT_MAX_BOUND_BEFORE_RELOAD_LOCAL_VERSION_MS, + RANDOM_WAIT_MAX_BOUND_BEFORE_RELOAD_LOCAL_VERSION_MS_DEFAULT)); + pushMinWaitBeforeReloadLocalVersionMs = + memoize( + () -> + cfg.get() + .getInt( + REPLICATION_FILTER_SECTION, + REPLICATION_PUSH_FILTER_SUBSECTION, + MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS, + MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS_DEFAULT)); + pushWaitBeforeReloadLocalVersionMs = + memoize( + () -> + cfg.get() + .getInt( + REPLICATION_FILTER_SECTION, + REPLICATION_PUSH_FILTER_SUBSECTION, + RANDOM_WAIT_MAX_BOUND_BEFORE_RELOAD_LOCAL_VERSION_MS, + RANDOM_WAIT_MAX_BOUND_BEFORE_RELOAD_LOCAL_VERSION_MS_DEFAULT)); + } + + public boolean isFetchFilterRandomSleepEnabled() { + return fetchWaitBeforeReloadLocalVersionMs.get() != 0; + } + + public Integer fetchFilterRandomSleepTimeMs() { + return fetchMinWaitBeforeReloadLocalVersionMs.get() + + new Random().nextInt(fetchWaitBeforeReloadLocalVersionMs.get()); + } + + public boolean isPushFilterRandomSleepEnabled() { + return pushWaitBeforeReloadLocalVersionMs.get() != 0; + } + + public Integer pushFilterRandomSleepTimeMs() { + return pushMinWaitBeforeReloadLocalVersionMs.get() + + new Random().nextInt(pushWaitBeforeReloadLocalVersionMs.get()); + } + } + static boolean getBoolean( Supplier<Config> cfg, String section, String subsection, String name, boolean defaultValue) { try {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jMessageLogger.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jMessageLogger.java index 7c88655..95c417f 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jMessageLogger.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jMessageLogger.java
@@ -14,9 +14,9 @@ package com.googlesource.gerrit.plugins.multisite; -import com.gerritforge.gerrit.eventbroker.EventGsonProvider; -import com.gerritforge.gerrit.eventbroker.EventMessage; import com.google.gerrit.extensions.systemstatus.ServerInformation; +import com.google.gerrit.server.events.Event; +import com.google.gerrit.server.events.EventGsonProvider; import com.google.gerrit.server.util.PluginLogFile; import com.google.gerrit.server.util.SystemLog; import com.google.gson.Gson; @@ -41,7 +41,7 @@ } @Override - public void log(Direction direction, String topic, EventMessage event) { + public void log(Direction direction, String topic, Event event) { msgLog.info("{} {} {}", direction, topic, gson.toJson(event)); } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jProjectVersionLogger.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jProjectVersionLogger.java index c2c4b46..23c720a 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jProjectVersionLogger.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/Log4jProjectVersionLogger.java
@@ -45,4 +45,9 @@ verLog.info("{ \"project\":\"{}\", \"version\":{} }", projectName, currentVersion); } } + + @Override + public void logDeleted(Project.NameKey projectName) { + verLog.info("{ \"project\":\"{}\", \"status\":\"DELETED\" }", projectName); + } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/MessageLogger.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/MessageLogger.java index b1f3e79..cc64b02 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/MessageLogger.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/MessageLogger.java
@@ -14,7 +14,7 @@ package com.googlesource.gerrit.plugins.multisite; -import com.gerritforge.gerrit.eventbroker.EventMessage; +import com.google.gerrit.server.events.Event; public interface MessageLogger { @@ -23,5 +23,5 @@ CONSUME; } - public void log(Direction direction, String topic, EventMessage event); + public void log(Direction direction, String topic, Event event); }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/Module.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/Module.java index 91a927b..9392077 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/Module.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/Module.java
@@ -16,30 +16,17 @@ import com.gerritforge.gerrit.globalrefdb.validation.LibModule; import com.google.gerrit.lifecycle.LifecycleModule; -import com.google.gerrit.server.config.SitePaths; import com.google.inject.CreationException; import com.google.inject.Inject; -import com.google.inject.Provides; -import com.google.inject.Singleton; import com.google.inject.spi.Message; +import com.googlesource.gerrit.plugins.multisite.broker.BrokerModule; import com.googlesource.gerrit.plugins.multisite.cache.CacheModule; -import com.googlesource.gerrit.plugins.multisite.event.EventModule; import com.googlesource.gerrit.plugins.multisite.forwarder.ForwarderModule; import com.googlesource.gerrit.plugins.multisite.forwarder.router.RouterModule; import com.googlesource.gerrit.plugins.multisite.index.IndexModule; -import java.io.BufferedReader; -import java.io.BufferedWriter; -import java.io.IOException; -import java.nio.file.Files; -import java.nio.file.Path; -import java.nio.file.Paths; import java.util.Collection; -import java.util.UUID; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; public class Module extends LifecycleModule { - private static final Logger log = LoggerFactory.getLogger(Module.class); private Configuration config; @Inject @@ -49,6 +36,7 @@ @Override protected void configure() { + boolean brokerRouterNeeded = false; Collection<Message> validationErrors = config.validate(); if (!validationErrors.isEmpty()) { @@ -64,65 +52,19 @@ if (config.cache().synchronize()) { install(new CacheModule()); + brokerRouterNeeded = true; } if (config.event().synchronize()) { - install(new EventModule(config)); + brokerRouterNeeded = true; } if (config.index().synchronize()) { install(new IndexModule()); + brokerRouterNeeded = true; } - install(new RouterModule()); - } - - @Provides - @Singleton - @InstanceId - UUID getInstanceId(SitePaths sitePaths) throws IOException { - UUID instanceId = null; - Path dataDir = sitePaths.data_dir.resolve(Configuration.PLUGIN_NAME); - if (!dataDir.toFile().exists()) { - dataDir.toFile().mkdirs(); + if (brokerRouterNeeded) { + install(new BrokerModule()); + install(new RouterModule()); } - String serverIdFile = - dataDir.toAbsolutePath().toString() + "/" + Configuration.INSTANCE_ID_FILE; - - instanceId = tryToLoadSavedInstanceId(serverIdFile); - - if (instanceId == null) { - instanceId = UUID.randomUUID(); - Files.createFile(Paths.get(serverIdFile)); - try (BufferedWriter writer = Files.newBufferedWriter(Paths.get(serverIdFile))) { - writer.write(instanceId.toString()); - } catch (IOException e) { - log.warn( - String.format( - "Cannot write instance ID, a new one will be generated at instance restart. (%s)", - e.getMessage())); - } - } - return instanceId; - } - - private UUID tryToLoadSavedInstanceId(String serverIdFile) { - if (Files.exists(Paths.get(serverIdFile))) { - try (BufferedReader br = Files.newBufferedReader(Paths.get(serverIdFile))) { - return UUID.fromString(br.readLine()); - } catch (IOException e) { - log.warn( - String.format( - "Cannot read instance ID from path '%s', deleting the old file and generating a new ID: (%s)", - serverIdFile, e.getMessage())); - try { - Files.delete(Paths.get(serverIdFile)); - } catch (IOException e1) { - log.warn( - String.format( - "Cannot delete old instance ID file at path '%s' with instance ID while generating a new one: (%s)", - serverIdFile, e1.getMessage())); - } - } - } - return null; } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/PluginModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/PluginModule.java index 20aaef6..db06c9f 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/PluginModule.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/PluginModule.java
@@ -15,38 +15,99 @@ package com.googlesource.gerrit.plugins.multisite; import com.gerritforge.gerrit.globalrefdb.validation.ProjectDeletedSharedDbCleanup; +import com.google.common.collect.ImmutableList; +import com.google.common.flogger.FluentLogger; import com.google.gerrit.extensions.events.ProjectDeletedListener; import com.google.gerrit.extensions.registration.DynamicSet; import com.google.gerrit.lifecycle.LifecycleModule; +import com.google.gerrit.server.git.WorkQueue; +import com.google.inject.AbstractModule; import com.google.inject.Inject; +import com.google.inject.Injector; +import com.google.inject.ProvisionException; import com.google.inject.Scopes; import com.googlesource.gerrit.plugins.multisite.broker.BrokerApiWrapper; import com.googlesource.gerrit.plugins.multisite.consumer.MultiSiteConsumerRunner; import com.googlesource.gerrit.plugins.multisite.consumer.ReplicationStatusModule; import com.googlesource.gerrit.plugins.multisite.consumer.SubscriberModule; +import com.googlesource.gerrit.plugins.multisite.event.EventModule; import com.googlesource.gerrit.plugins.multisite.forwarder.broker.BrokerForwarderModule; public class PluginModule extends LifecycleModule { - private Configuration config; + private static final FluentLogger log = FluentLogger.forEnclosingClass(); + private static final String[] FILTER_MODULES_CLASS_NAMES = + new String[] { + /* Class names are defined as String for avoiding this class failing to load + * if either replication or pull-replication plugins are missing. + */ + "com.googlesource.gerrit.plugins.multisite.validation.PullReplicationFilterModule", + "com.googlesource.gerrit.plugins.multisite.validation.PushReplicationFilterModule" + }; + + private final Configuration config; + private final WorkQueue workQueue; + private final Injector parentInjector; @Inject - public PluginModule(Configuration config) { + public PluginModule(Configuration config, WorkQueue workQueue, Injector parentInjector) { this.config = config; + this.workQueue = workQueue; + this.parentInjector = parentInjector; } @Override protected void configure() { - bind(BrokerApiWrapper.class).in(Scopes.SINGLETON); - install(new SubscriberModule()); + if (config.index().synchronize() + || config.cache().synchronize() + || config.event().synchronize()) { + install(new EventModule(config)); + bind(BrokerApiWrapper.class).in(Scopes.SINGLETON); + install(new SubscriberModule()); - install(new BrokerForwarderModule()); - listener().to(MultiSiteConsumerRunner.class); + install(new BrokerForwarderModule()); + listener().to(MultiSiteConsumerRunner.class); - install(new ReplicationStatusModule()); + install(new ReplicationStatusModule(workQueue)); + } + if (config.getSharedRefDbConfiguration().getSharedRefDb().isEnabled()) { listener().to(PluginStartup.class); DynamicSet.bind(binder(), ProjectDeletedListener.class) .to(ProjectDeletedSharedDbCleanup.class); } + + detectFilterModules() + .forEach( + mod -> { + install(mod); + log.atInfo().log( + "Replication filter module %s installed successfully", + mod.getClass().getSimpleName()); + }); + } + + private Iterable<AbstractModule> detectFilterModules() { + ImmutableList.Builder<AbstractModule> filterModulesBuilder = ImmutableList.builder(); + + for (String filterClassName : FILTER_MODULES_CLASS_NAMES) { + try { + @SuppressWarnings("unchecked") + Class<AbstractModule> filterClass = (Class<AbstractModule>) Class.forName(filterClassName); + + AbstractModule filterModule = parentInjector.getInstance(filterClass); + // Check if the filterModule would be valid for creating a child Guice Injector + parentInjector.createChildInjector(filterModule); + + filterModulesBuilder.add(filterModule); + } catch (NoClassDefFoundError | ClassNotFoundException e) { + log.atFine().withCause(e).log( + "Not loading %s because of missing the associated replication plugin", filterClassName); + } catch (Exception e) { + throw new ProvisionException( + "Unable to instantiate replication filter " + filterClassName, e); + } + } + + return filterModulesBuilder.build(); } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/ProjectVersionLogger.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/ProjectVersionLogger.java index 6ababb6..2ee2c13 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/ProjectVersionLogger.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/ProjectVersionLogger.java
@@ -19,4 +19,6 @@ public interface ProjectVersionLogger { public void log(Project.NameKey projectName, long currentVersion, long replicationLag); + + public void logDeleted(Project.NameKey projectName); }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapper.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapper.java index 71be5e6..f58efa7 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapper.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapper.java
@@ -15,62 +15,101 @@ package com.googlesource.gerrit.plugins.multisite.broker; import com.gerritforge.gerrit.eventbroker.BrokerApi; -import com.gerritforge.gerrit.eventbroker.EventMessage; import com.gerritforge.gerrit.eventbroker.TopicSubscriber; +import com.google.common.base.Strings; +import com.google.common.util.concurrent.FutureCallback; +import com.google.common.util.concurrent.Futures; +import com.google.common.util.concurrent.ListenableFuture; +import com.google.common.util.concurrent.SettableFuture; import com.google.gerrit.extensions.registration.DynamicItem; import com.google.gerrit.server.events.Event; import com.google.inject.Inject; -import com.googlesource.gerrit.plugins.multisite.InstanceId; import com.googlesource.gerrit.plugins.multisite.MessageLogger; import com.googlesource.gerrit.plugins.multisite.MessageLogger.Direction; import com.googlesource.gerrit.plugins.multisite.forwarder.Context; import java.util.Set; -import java.util.UUID; +import java.util.concurrent.Executor; import java.util.function.Consumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; public class BrokerApiWrapper implements BrokerApi { + private static final Logger log = LoggerFactory.getLogger(BrokerApiWrapper.class); + private final Executor executor; private final DynamicItem<BrokerApi> apiDelegate; private final BrokerMetrics metrics; private final MessageLogger msgLog; - private final UUID instanceId; @Inject public BrokerApiWrapper( + @BrokerExecutor Executor executor, DynamicItem<BrokerApi> apiDelegate, BrokerMetrics metrics, - MessageLogger msgLog, - @InstanceId UUID instanceId) { + MessageLogger msgLog) { this.apiDelegate = apiDelegate; + this.executor = executor; this.metrics = metrics; this.msgLog = msgLog; - this.instanceId = instanceId; } - public boolean send(String topic, Event event) { - return send(topic, apiDelegate.get().newMessage(instanceId, event)); - } - - @Override - public boolean send(String topic, EventMessage message) { - if (Context.isForwardedEvent()) { - return true; - } - boolean succeeded = false; + public boolean sendSync(String topic, Event event) { try { - succeeded = apiDelegate.get().send(topic, message); - } finally { - if (succeeded) { - msgLog.log(Direction.PUBLISH, topic, message); - metrics.incrementBrokerPublishedMessage(); - } else { - metrics.incrementBrokerFailedToPublishMessage(); - } + return send(topic, event).get(); + } catch (Throwable e) { + log.error( + "Failed to publish event '{}' to topic '{}' - error: {} - stack trace: {}", + event, + topic, + e.getMessage(), + e.getStackTrace()); + metrics.incrementBrokerFailedToPublishMessage(); + return false; } - return succeeded; } @Override - public void receiveAsync(String topic, Consumer<EventMessage> messageConsumer) { + public ListenableFuture<Boolean> send(String topic, Event message) { + SettableFuture<Boolean> resultFuture = SettableFuture.create(); + if (Context.isForwardedEvent()) { + resultFuture.set(true); + return resultFuture; + } + + if (Strings.isNullOrEmpty(message.instanceId)) { + log.warn( + "Dropping event '{}' because event instance id cannot be null or empty", + message.toString()); + resultFuture.set(true); + return resultFuture; + } + + ListenableFuture<Boolean> resfultF = apiDelegate.get().send(topic, message); + Futures.addCallback( + resfultF, + new FutureCallback<Boolean>() { + @Override + public void onSuccess(Boolean result) { + msgLog.log(Direction.PUBLISH, topic, message); + metrics.incrementBrokerPublishedMessage(); + } + + @Override + public void onFailure(Throwable throwable) { + log.error( + "Failed to publish message '{}' to topic '{}' - error: {}", + message.toString(), + topic, + throwable.getMessage()); + metrics.incrementBrokerFailedToPublishMessage(); + } + }, + executor); + + return resfultF; + } + + @Override + public void receiveAsync(String topic, Consumer<Event> messageConsumer) { apiDelegate.get().receiveAsync(topic, messageConsumer); }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/InstanceId.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerExecutor.java similarity index 83% rename from src/main/java/com/googlesource/gerrit/plugins/multisite/InstanceId.java rename to src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerExecutor.java index 87306a2..aa24eb1 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/InstanceId.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerExecutor.java
@@ -1,4 +1,4 @@ -// Copyright (C) 2019 The Android Open Source Project +// Copyright (C) 2021 The Android Open Source Project // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -package com.googlesource.gerrit.plugins.multisite; +package com.googlesource.gerrit.plugins.multisite.broker; import static java.lang.annotation.RetentionPolicy.RUNTIME; @@ -21,4 +21,4 @@ @Retention(RUNTIME) @BindingAnnotation -public @interface InstanceId {} +@interface BrokerExecutor {}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerExecutorProvider.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerExecutorProvider.java new file mode 100644 index 0000000..e843263 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerExecutorProvider.java
@@ -0,0 +1,29 @@ +// Copyright (C) 2021 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.broker; + +import com.google.gerrit.server.git.WorkQueue; +import com.google.inject.Inject; +import com.google.inject.Singleton; +import com.googlesource.gerrit.plugins.multisite.ExecutorProvider; + +@Singleton +class BrokerExecutorProvider extends ExecutorProvider { + + @Inject + BrokerExecutorProvider(WorkQueue workQueue) { + super(workQueue, 1, "Multi-Site-Broker"); + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerModule.java new file mode 100644 index 0000000..a5dac4a --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerModule.java
@@ -0,0 +1,28 @@ +// Copyright (C) 2021 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.broker; + +import com.google.gerrit.lifecycle.LifecycleModule; +import java.util.concurrent.Executor; + +public class BrokerModule extends LifecycleModule { + + @Override + protected void configure() { + bind(Executor.class) + .annotatedWith(BrokerExecutor.class) + .toProvider(BrokerExecutorProvider.class); + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandler.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandler.java index 2990264..e418da5 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandler.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandler.java
@@ -17,6 +17,7 @@ import com.google.common.cache.RemovalNotification; import com.google.gerrit.extensions.registration.DynamicSet; import com.google.gerrit.server.cache.CacheRemovalListener; +import com.google.gerrit.server.config.GerritInstanceId; import com.google.inject.Inject; import com.googlesource.gerrit.plugins.multisite.forwarder.CacheEvictionForwarder; import com.googlesource.gerrit.plugins.multisite.forwarder.Context; @@ -28,21 +29,25 @@ private final Executor executor; private final DynamicSet<CacheEvictionForwarder> forwarders; private final CachePatternMatcher matcher; + private final String instanceId; @Inject CacheEvictionHandler( DynamicSet<CacheEvictionForwarder> forwarders, @CacheExecutor Executor executor, - CachePatternMatcher matcher) { + CachePatternMatcher matcher, + @GerritInstanceId String instanceId) { this.forwarders = forwarders; this.executor = executor; this.matcher = matcher; + this.instanceId = instanceId; } @Override public void onRemoval(String plugin, String cache, RemovalNotification<K, V> notification) { if (!Context.isForwardedEvent() && !notification.wasEvicted() && matcher.matches(cache)) { - executor.execute(new CacheEvictionTask(new CacheEvictionEvent(cache, notification.getKey()))); + executor.execute( + new CacheEvictionTask(new CacheEvictionEvent(cache, notification.getKey(), instanceId))); } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CachePatternMatcher.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CachePatternMatcher.java index b8521a3..2dcb09a 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CachePatternMatcher.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/CachePatternMatcher.java
@@ -25,7 +25,7 @@ @Singleton class CachePatternMatcher { - private static final List<String> DEFAULT_PATTERNS = + private static final ImmutableList<String> DEFAULT_PATTERNS = ImmutableList.of("^groups.*", "ldap_groups", "ldap_usernames", "projects", "sshkeys"); private final Pattern pattern;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandler.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandler.java index fdc6fc3..44f8417 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandler.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandler.java
@@ -19,6 +19,7 @@ import com.google.gerrit.extensions.events.ProjectDeletedListener; import com.google.gerrit.extensions.events.ProjectEvent; import com.google.gerrit.extensions.registration.DynamicSet; +import com.google.gerrit.server.config.GerritInstanceId; import com.google.inject.Inject; import com.google.inject.Singleton; import com.googlesource.gerrit.plugins.multisite.forwarder.Context; @@ -33,15 +34,18 @@ private final DynamicSet<ProjectListUpdateForwarder> forwarders; private final Executor executor; private final ProjectsFilter projectsFilter; + private final String instanceId; @Inject public ProjectListUpdateHandler( DynamicSet<ProjectListUpdateForwarder> forwarders, @CacheExecutor Executor executor, - ProjectsFilter filter) { + ProjectsFilter filter, + @GerritInstanceId String instanceId) { this.forwarders = forwarders; this.executor = executor; this.projectsFilter = filter; + this.instanceId = instanceId; } @Override @@ -59,7 +63,8 @@ private void process(ProjectEvent event, boolean delete) { if (!Context.isForwardedEvent() && projectsFilter.matches(event.getProjectName())) { executor.execute( - new ProjectListUpdateTask(new ProjectListUpdateEvent(event.getProjectName(), delete))); + new ProjectListUpdateTask( + new ProjectListUpdateEvent(event.getProjectName(), delete, instanceId))); } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubcriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubcriber.java index b37f434..7107b87 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubcriber.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubcriber.java
@@ -14,20 +14,19 @@ package com.googlesource.gerrit.plugins.multisite.consumer; -import com.gerritforge.gerrit.eventbroker.EventMessage; import com.google.common.base.Strings; import com.google.common.flogger.FluentLogger; import com.google.gerrit.extensions.registration.DynamicSet; +import com.google.gerrit.server.config.GerritInstanceId; +import com.google.gerrit.server.events.Event; import com.google.gerrit.server.permissions.PermissionBackendException; import com.googlesource.gerrit.plugins.multisite.Configuration; -import com.googlesource.gerrit.plugins.multisite.InstanceId; import com.googlesource.gerrit.plugins.multisite.MessageLogger; import com.googlesource.gerrit.plugins.multisite.MessageLogger.Direction; import com.googlesource.gerrit.plugins.multisite.forwarder.CacheNotFoundException; import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventTopic; import com.googlesource.gerrit.plugins.multisite.forwarder.router.ForwardedEventRouter; import java.io.IOException; -import java.util.UUID; import java.util.function.Consumer; public abstract class AbstractSubcriber { @@ -44,13 +43,13 @@ public AbstractSubcriber( ForwardedEventRouter eventRouter, DynamicSet<DroppedEventListener> droppedEventListeners, - @InstanceId UUID instanceId, + @GerritInstanceId String gerritInstanceId, MessageLogger msgLog, SubscriberMetrics subscriberMetrics, Configuration cfg) { this.eventRouter = eventRouter; this.droppedEventListeners = droppedEventListeners; - this.instanceId = instanceId.toString(); + this.instanceId = gerritInstanceId; this.msgLog = msgLog; this.subscriberMetrics = subscriberMetrics; this.cfg = cfg; @@ -59,19 +58,22 @@ protected abstract EventTopic getTopic(); - public Consumer<EventMessage> getConsumer() { + protected abstract Boolean shouldConsumeEvent(Event event); + + public Consumer<Event> getConsumer() { return this::processRecord; } - private void processRecord(EventMessage event) { - String sourceInstanceId = event.getHeader().sourceInstanceId; + private void processRecord(Event event) { + String sourceInstanceId = event.instanceId; - if (Strings.isNullOrEmpty(sourceInstanceId) || instanceId.equals(sourceInstanceId)) { + if ((Strings.isNullOrEmpty(sourceInstanceId) || instanceId.equals(sourceInstanceId)) + || !shouldConsumeEvent(event)) { if (Strings.isNullOrEmpty(sourceInstanceId)) { logger.atWarning().log( String.format( "Dropping event %s because sourceInstanceId cannot be null", event.toString())); - } else { + } else if (instanceId.equals(sourceInstanceId)) { logger.atFiner().log( String.format( "Dropping event %s produced by our instanceId %s", event.toString(), instanceId)); @@ -80,16 +82,16 @@ } else { try { msgLog.log(Direction.CONSUME, topic, event); - eventRouter.route(event.getEvent()); + eventRouter.route(event); subscriberMetrics.incrementSubscriberConsumedMessage(); - subscriberMetrics.updateReplicationStatusMetrics(event); } catch (IOException e) { - logger.atSevere().withCause(e).log("Malformed event '%s'", event.getHeader()); + logger.atSevere().withCause(e).log("Malformed event '%s'", event); subscriberMetrics.incrementSubscriberFailedToConsumeMessage(); } catch (PermissionBackendException | CacheNotFoundException e) { - logger.atSevere().withCause(e).log("Cannot handle message '%s'", event.getHeader()); + logger.atSevere().withCause(e).log("Cannot handle message '%s'", event); subscriberMetrics.incrementSubscriberFailedToConsumeMessage(); } } + subscriberMetrics.updateReplicationStatusMetrics(event); } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriber.java index 5bbaee0..cdbf220 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriber.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriber.java
@@ -14,31 +14,49 @@ package com.googlesource.gerrit.plugins.multisite.consumer; +import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter; import com.google.gerrit.extensions.registration.DynamicSet; +import com.google.gerrit.server.config.GerritInstanceId; +import com.google.gerrit.server.events.Event; import com.google.inject.Inject; import com.google.inject.Singleton; import com.googlesource.gerrit.plugins.multisite.Configuration; -import com.googlesource.gerrit.plugins.multisite.InstanceId; import com.googlesource.gerrit.plugins.multisite.MessageLogger; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.ChangeIndexEvent; import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventTopic; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent; import com.googlesource.gerrit.plugins.multisite.forwarder.router.IndexEventRouter; -import java.util.UUID; @Singleton public class BatchIndexEventSubscriber extends AbstractSubcriber { + private final ProjectsFilter projectsFilter; + @Inject public BatchIndexEventSubscriber( IndexEventRouter eventRouter, DynamicSet<DroppedEventListener> droppedEventListeners, - @InstanceId UUID instanceId, + @GerritInstanceId String instanceId, MessageLogger msgLog, SubscriberMetrics subscriberMetrics, - Configuration cfg) { + Configuration cfg, + ProjectsFilter projectsFilter) { super(eventRouter, droppedEventListeners, instanceId, msgLog, subscriberMetrics, cfg); + this.projectsFilter = projectsFilter; } @Override protected EventTopic getTopic() { return EventTopic.BATCH_INDEX_TOPIC; } + + @Override + protected Boolean shouldConsumeEvent(Event event) { + if (event instanceof ChangeIndexEvent) { + return projectsFilter.matches(((ChangeIndexEvent) event).projectName); + } + if (event instanceof ProjectIndexEvent) { + return projectsFilter.matches(((ProjectIndexEvent) event).projectName); + } + return true; + } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/CacheEvictionEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/CacheEvictionEventSubscriber.java index eae66b4..d8342b0 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/CacheEvictionEventSubscriber.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/CacheEvictionEventSubscriber.java
@@ -15,14 +15,14 @@ package com.googlesource.gerrit.plugins.multisite.consumer; import com.google.gerrit.extensions.registration.DynamicSet; +import com.google.gerrit.server.config.GerritInstanceId; +import com.google.gerrit.server.events.Event; import com.google.inject.Inject; import com.google.inject.Singleton; import com.googlesource.gerrit.plugins.multisite.Configuration; -import com.googlesource.gerrit.plugins.multisite.InstanceId; import com.googlesource.gerrit.plugins.multisite.MessageLogger; import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventTopic; import com.googlesource.gerrit.plugins.multisite.forwarder.router.CacheEvictionEventRouter; -import java.util.UUID; @Singleton public class CacheEvictionEventSubscriber extends AbstractSubcriber { @@ -30,7 +30,7 @@ public CacheEvictionEventSubscriber( CacheEvictionEventRouter eventRouter, DynamicSet<DroppedEventListener> droppedEventListeners, - @InstanceId UUID instanceId, + @GerritInstanceId String instanceId, MessageLogger msgLog, SubscriberMetrics subscriberMetrics, Configuration cfg) { @@ -42,4 +42,9 @@ protected EventTopic getTopic() { return EventTopic.CACHE_TOPIC; } + + @Override + protected Boolean shouldConsumeEvent(Event event) { + return true; + } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/DroppedEventListener.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/DroppedEventListener.java index 6f4680c..b34e02c 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/DroppedEventListener.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/DroppedEventListener.java
@@ -14,7 +14,7 @@ package com.googlesource.gerrit.plugins.multisite.consumer; -import com.gerritforge.gerrit.eventbroker.EventMessage; +import com.google.gerrit.server.events.Event; public interface DroppedEventListener { /** @@ -22,5 +22,5 @@ * * @param event information about the event. */ - void onEventDropped(EventMessage event); + void onEventDropped(Event event); }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriber.java index 49d470a..480fc50 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriber.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriber.java
@@ -14,31 +14,68 @@ package com.googlesource.gerrit.plugins.multisite.consumer; +import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter; +import com.google.gerrit.entities.Change; import com.google.gerrit.extensions.registration.DynamicSet; +import com.google.gerrit.server.change.ChangeFinder; +import com.google.gerrit.server.config.GerritInstanceId; +import com.google.gerrit.server.events.Event; import com.google.inject.Inject; import com.google.inject.Singleton; import com.googlesource.gerrit.plugins.multisite.Configuration; -import com.googlesource.gerrit.plugins.multisite.InstanceId; import com.googlesource.gerrit.plugins.multisite.MessageLogger; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.ChangeIndexEvent; import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventTopic; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent; import com.googlesource.gerrit.plugins.multisite.forwarder.router.IndexEventRouter; -import java.util.UUID; +import java.util.Optional; @Singleton public class IndexEventSubscriber extends AbstractSubcriber { + private final ProjectsFilter projectsFilter; + private final ChangeFinder changeFinder; + @Inject public IndexEventSubscriber( IndexEventRouter eventRouter, DynamicSet<DroppedEventListener> droppedEventListeners, - @InstanceId UUID instanceId, + @GerritInstanceId String instanceId, MessageLogger msgLog, SubscriberMetrics subscriberMetrics, - Configuration cfg) { + Configuration cfg, + ProjectsFilter projectsFilter, + ChangeFinder changeFinder) { super(eventRouter, droppedEventListeners, instanceId, msgLog, subscriberMetrics, cfg); + this.projectsFilter = projectsFilter; + this.changeFinder = changeFinder; } @Override protected EventTopic getTopic() { return EventTopic.INDEX_TOPIC; } + + @Override + protected Boolean shouldConsumeEvent(Event event) { + if (event instanceof ChangeIndexEvent) { + ChangeIndexEvent changeIndexEvent = (ChangeIndexEvent) event; + String projectName = changeIndexEvent.projectName; + if (isDeletedChangeWithEmptyProject(changeIndexEvent)) { + projectName = findProjectFromChangeId(changeIndexEvent.changeId).orElse(projectName); + } + return projectsFilter.matches(projectName); + } + if (event instanceof ProjectIndexEvent) { + return projectsFilter.matches(((ProjectIndexEvent) event).projectName); + } + return true; + } + + private boolean isDeletedChangeWithEmptyProject(ChangeIndexEvent changeIndexEvent) { + return changeIndexEvent.deleted && changeIndexEvent.projectName.isEmpty(); + } + + private Optional<String> findProjectFromChangeId(int changeId) { + return changeFinder.findOne(Change.id(changeId)).map(c -> c.getChange().getProject().get()); + } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriber.java index 6ff0969..069a516 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriber.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriber.java
@@ -14,31 +14,42 @@ package com.googlesource.gerrit.plugins.multisite.consumer; +import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter; import com.google.gerrit.extensions.registration.DynamicSet; +import com.google.gerrit.server.config.GerritInstanceId; +import com.google.gerrit.server.events.Event; import com.google.inject.Inject; import com.google.inject.Singleton; import com.googlesource.gerrit.plugins.multisite.Configuration; -import com.googlesource.gerrit.plugins.multisite.InstanceId; import com.googlesource.gerrit.plugins.multisite.MessageLogger; import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventTopic; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectListUpdateEvent; import com.googlesource.gerrit.plugins.multisite.forwarder.router.ProjectListUpdateRouter; -import java.util.UUID; @Singleton public class ProjectUpdateEventSubscriber extends AbstractSubcriber { + private final ProjectsFilter projectsFilter; + @Inject public ProjectUpdateEventSubscriber( ProjectListUpdateRouter eventRouter, DynamicSet<DroppedEventListener> droppedEventListeners, - @InstanceId UUID instanceId, + @GerritInstanceId String instanceId, MessageLogger msgLog, SubscriberMetrics subscriberMetrics, - Configuration cfg) { + Configuration cfg, + ProjectsFilter projectsFilter) { super(eventRouter, droppedEventListeners, instanceId, msgLog, subscriberMetrics, cfg); + this.projectsFilter = projectsFilter; } @Override protected EventTopic getTopic() { return EventTopic.PROJECT_LIST_TOPIC; } + + @Override + protected Boolean shouldConsumeEvent(Event event) { + return projectsFilter.matches(((ProjectListUpdateEvent) event).projectName); + } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatus.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatus.java index 7360b8f..0253588 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatus.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatus.java
@@ -19,14 +19,20 @@ import com.google.common.flogger.FluentLogger; import com.google.gerrit.entities.Project; import com.google.gerrit.extensions.events.LifecycleListener; +import com.google.gerrit.extensions.events.ProjectDeletedListener; +import com.google.gerrit.metrics.CallbackMetric1; +import com.google.gerrit.metrics.MetricMaker; import com.google.gerrit.server.cache.CacheModule; import com.google.gerrit.server.cache.serialize.JavaCacheSerializer; import com.google.gerrit.server.cache.serialize.StringCacheSerializer; +import com.google.gerrit.server.git.WorkQueue; import com.google.gerrit.server.project.ProjectCache; import com.google.inject.Inject; import com.google.inject.Module; import com.google.inject.Singleton; import com.google.inject.name.Named; +import com.google.inject.name.Names; +import com.googlesource.gerrit.plugins.multisite.Configuration; import com.googlesource.gerrit.plugins.multisite.ProjectVersionLogger; import com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate; import java.util.Collection; @@ -36,23 +42,29 @@ import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; import java.util.stream.Collectors; @Singleton -public class ReplicationStatus implements LifecycleListener { +public class ReplicationStatus implements LifecycleListener, ProjectDeletedListener { private static final FluentLogger logger = FluentLogger.forEnclosingClass(); private final Map<String, Long> replicationStatusPerProject = new HashMap<>(); - static final String REPLICATION_STATUS_CACHE = "replication_status"; + static final String REPLICATION_STATUS = "replication_status"; - public static Module cacheModule() { + public static Module cacheModule(WorkQueue queue) { return new CacheModule() { @Override protected void configure() { - persist(REPLICATION_STATUS_CACHE, String.class, Long.class) + persist(REPLICATION_STATUS, String.class, Long.class) .version(1) .keySerializer(StringCacheSerializer.INSTANCE) .valueSerializer(new JavaCacheSerializer<>()); + + bind(ScheduledExecutorService.class) + .annotatedWith(Names.named(REPLICATION_STATUS)) + .toInstance(queue.createQueue(0, REPLICATION_STATUS)); } }; } @@ -62,20 +74,35 @@ private final Optional<ProjectVersionRefUpdate> projectVersionRefUpdate; private final ProjectVersionLogger verLogger; private final ProjectCache projectCache; + private final ScheduledExecutorService statusScheduler; + + private final Configuration config; + + private final MetricMaker metricMaker; @Inject public ReplicationStatus( - @Named(REPLICATION_STATUS_CACHE) Cache<String, Long> cache, + @Named(REPLICATION_STATUS) Cache<String, Long> cache, Optional<ProjectVersionRefUpdate> projectVersionRefUpdate, ProjectVersionLogger verLogger, - ProjectCache projectCache) { + ProjectCache projectCache, + @Named(REPLICATION_STATUS) ScheduledExecutorService statusScheduler, + Configuration config, + MetricMaker metricMaker) { this.cache = cache; this.projectVersionRefUpdate = projectVersionRefUpdate; this.verLogger = verLogger; this.projectCache = projectCache; + this.statusScheduler = statusScheduler; + this.config = config; + this.metricMaker = metricMaker; } public Long getMaxLag() { + return getMaxLagMillis() / 1000; + } + + public Long getMaxLagMillis() { Collection<Long> lags = replicationStatusPerProject.values(); if (lags.isEmpty()) { return 0L; @@ -121,15 +148,75 @@ } } + void removeProjectFromReplicationLagMetrics(Project.NameKey projectName) { + Optional<Long> localVersion = + projectVersionRefUpdate.get().getProjectLocalVersion(projectName.get()); + + if (!localVersion.isPresent() && replicationStatusPerProject.containsKey(projectName.get())) { + cache.invalidate(projectName.get()); + replicationStatusPerProject.remove(projectName.get()); + localVersionPerProject.remove(projectName.get()); + verLogger.logDeleted(projectName); + logger.atFine().log("Removed project '%s' from replication lag metrics", projectName); + } + } + + @VisibleForTesting + Runnable replicationLagMetricPerProject(CallbackMetric1<String, Long> metricCallback) { + return () -> { + if (replicationStatusPerProject.isEmpty()) { + metricCallback.forceCreate(""); + } else { + replicationStatusPerProject.entrySet().stream() + .filter(e -> e.getValue() > 0) + .forEach( + e -> + metricCallback.set( + SubscriberMetrics.sanitizeProjectName(e.getKey()), e.getValue())); + metricCallback.prune(); + } + }; + } + @VisibleForTesting public void doUpdateLag(Project.NameKey projectName, Long lag) { cache.put(projectName.get(), lag); replicationStatusPerProject.put(projectName.get(), lag); } + @VisibleForTesting + Long getReplicationStatus(String projectName) { + return replicationStatusPerProject.get(projectName); + } + + @VisibleForTesting + Long getLocalVersion(String projectName) { + return localVersionPerProject.get(projectName); + } + @Override public void start() { loadAllFromCache(); + + long replicationLagPollingInterval = config.replicationLagRefreshInterval().toMillis(); + + if (replicationLagPollingInterval > 0) { + statusScheduler.scheduleAtFixedRate( + this::refreshProjectsWithLag, + replicationLagPollingInterval, + replicationLagPollingInterval, + TimeUnit.MILLISECONDS); + } + } + + @VisibleForTesting + public void refreshProjectsWithLag() { + logger.atFine().log("Refreshing projects version lags triggered ..."); + replicationStatusPerProject.entrySet().stream() + .filter(entry -> entry.getValue() > 0) + .map(Map.Entry::getKey) + .map(Project::nameKey) + .forEach(this::updateReplicationLag); } @Override @@ -140,4 +227,9 @@ projectCache.all().stream().map(Project.NameKey::get).collect(Collectors.toSet()); replicationStatusPerProject.putAll(cache.getAllPresent(cachedProjects)); } + + @Override + public void onProjectDeleted(Event event) { + removeProjectFromReplicationLagMetrics(Project.nameKey(event.getProjectName())); + } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusModule.java index e954bc4..605e119 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusModule.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusModule.java
@@ -14,14 +14,27 @@ package com.googlesource.gerrit.plugins.multisite.consumer; +import com.google.gerrit.extensions.events.ProjectDeletedListener; +import com.google.gerrit.extensions.registration.DynamicSet; import com.google.gerrit.lifecycle.LifecycleModule; +import com.google.gerrit.server.git.WorkQueue; +import com.google.inject.Inject; import com.google.inject.Scopes; public class ReplicationStatusModule extends LifecycleModule { + + private final WorkQueue workQueue; + + @Inject + public ReplicationStatusModule(WorkQueue workQueue) { + this.workQueue = workQueue; + } + @Override protected void configure() { bind(ReplicationStatus.class).in(Scopes.SINGLETON); - install(ReplicationStatus.cacheModule()); + install(ReplicationStatus.cacheModule(workQueue)); listener().to(ReplicationStatus.class); + DynamicSet.bind(binder(), ProjectDeletedListener.class).to(ReplicationStatus.class); } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriber.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriber.java index 20c355e..ab64651 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriber.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriber.java
@@ -14,31 +14,45 @@ package com.googlesource.gerrit.plugins.multisite.consumer; +import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter; import com.google.gerrit.extensions.registration.DynamicSet; +import com.google.gerrit.server.config.GerritInstanceId; +import com.google.gerrit.server.events.Event; +import com.google.gerrit.server.events.ProjectEvent; import com.google.inject.Inject; import com.google.inject.Singleton; import com.googlesource.gerrit.plugins.multisite.Configuration; -import com.googlesource.gerrit.plugins.multisite.InstanceId; import com.googlesource.gerrit.plugins.multisite.MessageLogger; import com.googlesource.gerrit.plugins.multisite.forwarder.events.EventTopic; import com.googlesource.gerrit.plugins.multisite.forwarder.router.StreamEventRouter; -import java.util.UUID; @Singleton public class StreamEventSubscriber extends AbstractSubcriber { + private final ProjectsFilter projectsFilter; + @Inject public StreamEventSubscriber( StreamEventRouter eventRouter, DynamicSet<DroppedEventListener> droppedEventListeners, - @InstanceId UUID instanceId, + @GerritInstanceId String instanceId, MessageLogger msgLog, SubscriberMetrics subscriberMetrics, - Configuration cfg) { + Configuration cfg, + ProjectsFilter projectsFilter) { super(eventRouter, droppedEventListeners, instanceId, msgLog, subscriberMetrics, cfg); + this.projectsFilter = projectsFilter; } @Override protected EventTopic getTopic() { return EventTopic.STREAM_EVENT_TOPIC; } + + @Override + protected Boolean shouldConsumeEvent(Event event) { + if (event instanceof ProjectEvent) { + return projectsFilter.matches(((ProjectEvent) event).getProjectNameKey().get()); + } + return true; + } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetrics.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetrics.java index 899233e..bb971ff 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetrics.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetrics.java
@@ -14,32 +14,43 @@ package com.googlesource.gerrit.plugins.multisite.consumer; -import com.gerritforge.gerrit.eventbroker.EventMessage; -import com.google.common.flogger.FluentLogger; +import com.google.gerrit.metrics.CallbackMetric1; import com.google.gerrit.metrics.Counter1; import com.google.gerrit.metrics.Description; +import com.google.gerrit.metrics.Field; import com.google.gerrit.metrics.MetricMaker; import com.google.gerrit.server.events.Event; +import com.google.gerrit.server.events.ProjectEvent; import com.google.gerrit.server.events.RefUpdatedEvent; +import com.google.gerrit.server.logging.Metadata; import com.google.inject.Inject; import com.google.inject.Singleton; import com.googlesource.gerrit.plugins.multisite.MultiSiteMetrics; -import com.googlesource.gerrit.plugins.replication.RefReplicatedEvent; -import com.googlesource.gerrit.plugins.replication.RefReplicationDoneEvent; -import com.googlesource.gerrit.plugins.replication.ReplicationScheduledEvent; +import com.googlesource.gerrit.plugins.replication.events.ProjectDeletionReplicationSucceededEvent; +import com.googlesource.gerrit.plugins.replication.events.RefReplicatedEvent; +import com.googlesource.gerrit.plugins.replication.events.RefReplicationDoneEvent; +import com.googlesource.gerrit.plugins.replication.events.ReplicationScheduledEvent; +import java.util.regex.Matcher; +import java.util.regex.Pattern; @Singleton public class SubscriberMetrics extends MultiSiteMetrics { - private static final FluentLogger logger = FluentLogger.forEnclosingClass(); private static final String SUBSCRIBER_SUCCESS_COUNTER = "subscriber_msg_consumer_counter"; private static final String SUBSCRIBER_FAILURE_COUNTER = "subscriber_msg_consumer_failure_counter"; - private static final String REPLICATION_LAG_SEC = + public static final String REPLICATION_LAG_SEC = "multi_site/subscriber/subscriber_replication_status/sec_behind"; + private static final String REPLICATION_LAG_MSEC = + "multi_site/subscriber/subscriber_replication_status/msec_behind"; + private static final String REPLICATION_LAG_MSEC_PROJECT = + "multi_site/subscriber/subscriber_replication_status/msec_behind/per_project"; private final Counter1<String> subscriberSuccessCounter; private final Counter1<String> subscriberFailureCounter; private final ReplicationStatus replicationStatus; + private static final Pattern isValidMetricNamePattern = Pattern.compile("[a-zA-Z0-9_-]"); + private static final Field<String> PROJECT_NAME = + Field.ofString("project_name", Metadata.Builder::cacheName).build(); @Inject public SubscriberMetrics(MetricMaker metricMaker, ReplicationStatus replicationStatus) { @@ -64,6 +75,51 @@ Long.class, new Description("Replication lag (sec)").setGauge().setUnit(Description.Units.SECONDS), replicationStatus::getMaxLag); + metricMaker.newCallbackMetric( + REPLICATION_LAG_MSEC, + Long.class, + new Description("Replication lag (msec)") + .setGauge() + .setUnit(Description.Units.MILLISECONDS), + replicationStatus::getMaxLagMillis); + + CallbackMetric1<String, Long> metrics = + metricMaker.newCallbackMetric( + SubscriberMetrics.REPLICATION_LAG_MSEC_PROJECT, + Long.class, + new Description("Per-project replication lag (msec)") + .setGauge() + .setUnit(Description.Units.MILLISECONDS), + PROJECT_NAME); + metricMaker.newTrigger(metrics, replicationStatus.replicationLagMetricPerProject(metrics)); + } + + /** + * Ensures that the generated metric is compatible with prometheus metric names, as the set of + * values that represent a valid metric name are different from the ones that represent a valid + * project name. Main differences: - All _ are replaced with __ - All characters that aren't a + * letter(uppercase or lowercase) or a hyphen(-) are replaced with `_<hex_code>` This is to avoid + * all chances of name clashes when modifying a project name. + * + * @param name name of the metric to sanitize + * @return sanitized metric name + */ + public static String sanitizeProjectName(String name) { + StringBuilder sanitizedName = new StringBuilder(); + for (int i = 0; i < name.length(); i++) { + Character c = name.charAt(i); + Matcher matcher = isValidMetricNamePattern.matcher(String.valueOf(c)); + if (matcher.find()) { + if (c == '_') { + sanitizedName.append("__"); + } else { + sanitizedName.append(c); + } + } else { + sanitizedName.append("_").append(Integer.toHexString((int) c)); + } + } + return sanitizedName.toString(); } public void incrementSubscriberConsumedMessage() { @@ -74,20 +130,18 @@ subscriberFailureCounter.increment(SUBSCRIBER_FAILURE_COUNTER); } - public void updateReplicationStatusMetrics(EventMessage eventMessage) { - Event event = eventMessage.getEvent(); - if (event instanceof RefReplicationDoneEvent) { - RefReplicationDoneEvent replicationDone = (RefReplicationDoneEvent) event; - replicationStatus.updateReplicationLag(replicationDone.getProjectNameKey()); - } else if (event instanceof RefReplicatedEvent) { - RefReplicatedEvent replicated = (RefReplicatedEvent) event; - replicationStatus.updateReplicationLag(replicated.getProjectNameKey()); - } else if (event instanceof ReplicationScheduledEvent) { - ReplicationScheduledEvent updated = (ReplicationScheduledEvent) event; - replicationStatus.updateReplicationLag(updated.getProjectNameKey()); - } else if (event instanceof RefUpdatedEvent) { - RefUpdatedEvent updated = (RefUpdatedEvent) event; - replicationStatus.updateReplicationLag(updated.getProjectNameKey()); + public void updateReplicationStatusMetrics(Event event) { + + if (event instanceof RefReplicationDoneEvent + || event instanceof RefReplicatedEvent + || event instanceof ReplicationScheduledEvent + || event instanceof RefUpdatedEvent) { + ProjectEvent projectEvent = (ProjectEvent) event; + replicationStatus.updateReplicationLag(projectEvent.getProjectNameKey()); + } else if (event instanceof ProjectDeletionReplicationSucceededEvent) { + ProjectDeletionReplicationSucceededEvent projectDeletion = + (ProjectDeletionReplicationSucceededEvent) event; + replicationStatus.removeProjectFromReplicationLagMetrics(projectDeletion.getProjectNameKey()); } } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/Context.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/Context.java index 1c29d75..269388c 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/Context.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/Context.java
@@ -14,14 +14,22 @@ package com.googlesource.gerrit.plugins.multisite.forwarder; +import static com.googlesource.gerrit.plugins.replication.pull.api.PullReplicationEndpoints.APPLY_OBJECTS_API_ENDPOINT; +import static com.googlesource.gerrit.plugins.replication.pull.api.PullReplicationEndpoints.APPLY_OBJECT_API_ENDPOINT; + /** Allows to tag a forwarded event to avoid infinitely looping events. */ public class Context { + public static final String PULL_REPLICATION_PLUGIN_NAME = "pull-replication"; private static final ThreadLocal<Boolean> forwardedEvent = ThreadLocal.withInitial(() -> false); private Context() {} public static Boolean isForwardedEvent() { - return forwardedEvent.get(); + return forwardedEvent.get() + || + // When the event is a result of pull-replication event, is considered as + // "forwarded" action because did not happen on this node. + isPullReplicationApplyObjectIndexing(); } public static void setForwardedEvent(Boolean b) { @@ -31,4 +39,10 @@ public static void unsetForwardedEvent() { forwardedEvent.remove(); } + + public static boolean isPullReplicationApplyObjectIndexing() { + String threadName = Thread.currentThread().getName(); + return threadName.contains(PULL_REPLICATION_PLUGIN_NAME + "~" + APPLY_OBJECT_API_ENDPOINT) + || threadName.contains(PULL_REPLICATION_PLUGIN_NAME + "~" + APPLY_OBJECTS_API_ENDPOINT); + } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandler.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandler.java index 8d41500..522a78f 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandler.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandler.java
@@ -63,17 +63,27 @@ protected void attemptToIndex(String id, Optional<ChangeIndexEvent> indexEvent, int retryCount) { ChangeChecker checker = changeCheckerFactory.create(id); Optional<ChangeNotes> changeNotes = checker.getChangeNotes(); - if (changeNotes.isPresent()) { + boolean changeIsPresent = changeNotes.isPresent(); + boolean changeIsConsistent = checker.isChangeConsistent(); + if (changeIsPresent && changeIsConsistent) { reindexAndCheckIsUpToDate(id, indexEvent, checker, retryCount); } else { log.warn( - "Change {} not present yet in local Git repository (event={}) after {} attempt(s)", + "Change {} {} in local Git repository (event={}) after {} attempt(s)", id, + !changeIsPresent + ? "not present yet" + : (changeIsConsistent ? "is" : "is not") + " consistent", indexEvent, retryCount); if (!rescheduleIndex(id, indexEvent, retryCount + 1)) { log.error( - "Change {} could not be found in the local Git repository (event={})", id, indexEvent); + "Change {} {} in the local Git repository (event={})", + id, + !changeIsPresent + ? "could not be found" + : (changeIsConsistent ? "was" : "was not") + " consistent", + indexEvent); } } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerForwarder.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerForwarder.java index 975916a..19936ef 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerForwarder.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerForwarder.java
@@ -49,6 +49,6 @@ return true; } - return broker.send(eventTopic.topic(cfg), event); + return broker.sendSync(eventTopic.topic(cfg), event); } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerStreamEventForwarder.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerStreamEventForwarder.java index 9ff4688..8a020fc 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerStreamEventForwarder.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/broker/BrokerStreamEventForwarder.java
@@ -35,6 +35,6 @@ @Override public boolean send(Event event) { - return broker.send(EventTopic.STREAM_EVENT_TOPIC.topic(cfg), event); + return broker.sendSync(EventTopic.STREAM_EVENT_TOPIC.topic(cfg), event); } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/AccountIndexEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/AccountIndexEvent.java index 317eb76..065ef7e 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/AccountIndexEvent.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/AccountIndexEvent.java
@@ -21,8 +21,8 @@ public int accountId; - public AccountIndexEvent(int accountId) { - super(TYPE); + public AccountIndexEvent(int accountId, String instanceId) { + super(TYPE, instanceId); this.accountId = accountId; }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/CacheEvictionEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/CacheEvictionEvent.java index 1c2185f..4444756 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/CacheEvictionEvent.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/CacheEvictionEvent.java
@@ -22,8 +22,8 @@ public String cacheName; public Object key; - public CacheEvictionEvent(String cacheName, Object key) { - super(TYPE); + public CacheEvictionEvent(String cacheName, Object key, String instanceId) { + super(TYPE, instanceId); this.cacheName = cacheName; this.key = key; }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ChangeIndexEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ChangeIndexEvent.java index ab4ddf4..64fbdfb 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ChangeIndexEvent.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ChangeIndexEvent.java
@@ -28,8 +28,8 @@ public String targetSha; public boolean deleted; - public ChangeIndexEvent(String projectName, int changeId, boolean deleted) { - super(TYPE); + public ChangeIndexEvent(String projectName, int changeId, boolean deleted, String instanceId) { + super(TYPE, instanceId); this.projectName = projectName; this.changeId = changeId; this.deleted = deleted;
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/GroupIndexEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/GroupIndexEvent.java index 4981b2f..05ac198 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/GroupIndexEvent.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/GroupIndexEvent.java
@@ -24,8 +24,8 @@ public final String groupUUID; public final ObjectId sha1; - public GroupIndexEvent(String groupUUID, @Nullable ObjectId sha1) { - super(TYPE); + public GroupIndexEvent(String groupUUID, @Nullable ObjectId sha1, String instanceId) { + super(TYPE, instanceId); this.groupUUID = groupUUID; this.sha1 = sha1; }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/IndexEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/IndexEvent.java index ea2c3fb..2fdda72 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/IndexEvent.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/IndexEvent.java
@@ -15,7 +15,7 @@ package com.googlesource.gerrit.plugins.multisite.forwarder.events; public abstract class IndexEvent extends MultiSiteEvent { - protected IndexEvent(String type) { - super(type); + protected IndexEvent(String type, String instanceId) { + super(type, instanceId); } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/MultiSiteEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/MultiSiteEvent.java index 404d168..f29204b 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/MultiSiteEvent.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/MultiSiteEvent.java
@@ -29,7 +29,8 @@ register(ProjectListUpdateEvent.TYPE, ProjectListUpdateEvent.class); } - protected MultiSiteEvent(String type) { + protected MultiSiteEvent(String type, String instanceId) { super(type); + this.instanceId = instanceId; } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectIndexEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectIndexEvent.java index 8bdb7b5..954befb 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectIndexEvent.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectIndexEvent.java
@@ -21,8 +21,8 @@ public String projectName; - public ProjectIndexEvent(String projectName) { - super(TYPE); + public ProjectIndexEvent(String projectName, String instanceId) { + super(TYPE, instanceId); this.projectName = projectName; }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectListUpdateEvent.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectListUpdateEvent.java index d030e5b..0e18b27 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectListUpdateEvent.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/events/ProjectListUpdateEvent.java
@@ -22,8 +22,8 @@ public String projectName; public boolean remove; - public ProjectListUpdateEvent(String projectName, boolean remove) { - super(TYPE); + public ProjectListUpdateEvent(String projectName, boolean remove, String instanceId) { + super(TYPE, instanceId); this.projectName = projectName; this.remove = remove; }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/IndexEventRouter.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/IndexEventRouter.java index 202fb42..a647bce 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/IndexEventRouter.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/IndexEventRouter.java
@@ -21,6 +21,10 @@ import com.google.gerrit.entities.Account; import com.google.gerrit.extensions.events.LifecycleListener; import com.google.gerrit.server.config.AllUsersName; +import com.google.gerrit.server.config.GerritInstanceId; +import com.google.gerrit.server.events.Event; +import com.google.gerrit.server.events.EventListener; +import com.google.gerrit.server.events.RefEvent; import com.google.inject.Inject; import com.googlesource.gerrit.plugins.multisite.forwarder.ForwardedIndexAccountHandler; import com.googlesource.gerrit.plugins.multisite.forwarder.ForwardedIndexChangeHandler; @@ -32,12 +36,12 @@ import com.googlesource.gerrit.plugins.multisite.forwarder.events.GroupIndexEvent; import com.googlesource.gerrit.plugins.multisite.forwarder.events.IndexEvent; import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent; -import com.googlesource.gerrit.plugins.replication.RefReplicationDoneEvent; import java.io.IOException; import java.util.Optional; import java.util.Set; -public class IndexEventRouter implements ForwardedEventRouter<IndexEvent>, LifecycleListener { +public class IndexEventRouter + implements ForwardedEventRouter<IndexEvent>, EventListener, LifecycleListener { private static final FluentLogger logger = FluentLogger.forEnclosingClass(); private final ForwardedIndexAccountHandler indexAccountHandler; @@ -45,6 +49,7 @@ private final ForwardedIndexGroupHandler indexGroupHandler; private final ForwardedIndexProjectHandler indexProjectHandler; private final AllUsersName allUsersName; + private final String gerritInstanceId; @Inject public IndexEventRouter( @@ -52,12 +57,14 @@ ForwardedIndexChangeHandler indexChangeHandler, ForwardedIndexGroupHandler indexGroupHandler, ForwardedIndexProjectHandler indexProjectHandler, - AllUsersName allUsersName) { + AllUsersName allUsersName, + @GerritInstanceId String gerritInstanceId) { this.indexAccountHandler = indexAccountHandler; this.indexChangeHandler = indexChangeHandler; this.indexGroupHandler = indexGroupHandler; this.indexProjectHandler = indexProjectHandler; this.allUsersName = allUsersName; + this.gerritInstanceId = gerritInstanceId; } @Override @@ -85,7 +92,7 @@ } } - public void onRefReplicated(RefReplicationDoneEvent replicationEvent) throws IOException { + public void onRefReplicated(RefEvent replicationEvent) throws IOException { if (replicationEvent.getProjectNameKey().equals(allUsersName)) { Account.Id accountId = Account.Id.fromRef(replicationEvent.getRefName()); if (accountId != null) { @@ -97,6 +104,20 @@ } @Override + public void onEvent(Event event) { + if (event instanceof RefEvent + && (event.getType().contains("fetch-ref-replicated") + || event.getType().contains("fetch-ref-replication-done")) + && gerritInstanceId.equals(event.instanceId)) { + try { + onRefReplicated((RefEvent) event); + } catch (IOException e) { + logger.atSevere().withCause(e).log("Error while processing event %s", event); + } + } + } + + @Override public void start() {} @Override
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/RouterModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/RouterModule.java index bac907e..2193034 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/RouterModule.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/RouterModule.java
@@ -14,7 +14,9 @@ package com.googlesource.gerrit.plugins.multisite.forwarder.router; +import com.google.gerrit.extensions.registration.DynamicSet; import com.google.gerrit.lifecycle.LifecycleModule; +import com.google.gerrit.server.events.EventListener; import com.google.inject.Scopes; public class RouterModule extends LifecycleModule { @@ -22,6 +24,8 @@ protected void configure() { bind(IndexEventRouter.class).in(Scopes.SINGLETON); listener().to(IndexEventRouter.class).in(Scopes.SINGLETON); + DynamicSet.bind(binder(), EventListener.class).to(IndexEventRouter.class); + bind(CacheEvictionEventRouter.class).in(Scopes.SINGLETON); bind(ProjectListUpdateRouter.class).in(Scopes.SINGLETON); bind(StreamEventRouter.class).in(Scopes.SINGLETON);
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/StreamEventRouter.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/StreamEventRouter.java index 4ef3426..95a3e66 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/StreamEventRouter.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/forwarder/router/StreamEventRouter.java
@@ -18,7 +18,7 @@ import com.google.gerrit.server.permissions.PermissionBackendException; import com.google.inject.Inject; import com.googlesource.gerrit.plugins.multisite.forwarder.ForwardedEventHandler; -import com.googlesource.gerrit.plugins.replication.RefReplicationDoneEvent; +import com.googlesource.gerrit.plugins.replication.events.RefReplicationDoneEvent; import java.io.IOException; public class StreamEventRouter implements ForwardedEventRouter<Event> {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/http/HttpModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/http/HttpModule.java index 2ae2d9b..26bfb53 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/http/HttpModule.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/http/HttpModule.java
@@ -14,14 +14,25 @@ package com.googlesource.gerrit.plugins.multisite.http; +import com.google.inject.Inject; import com.google.inject.servlet.ServletModule; +import com.googlesource.gerrit.plugins.multisite.Configuration; public class HttpModule extends ServletModule { public static final String LAG_ENDPOINT_SEGMENT = "replication-lag"; + private final Configuration config; + + @Inject + public HttpModule(Configuration config) { + this.config = config; + } + @Override protected void configureServlets() { - serve(String.format("/%s", LAG_ENDPOINT_SEGMENT)).with(ReplicationStatusServlet.class); + if (config.event().synchronize()) { + serve(String.format("/%s", LAG_ENDPOINT_SEGMENT)).with(ReplicationStatusServlet.class); + } } }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeChecker.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeChecker.java index 9ee59eb..3277150 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeChecker.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeChecker.java
@@ -59,4 +59,11 @@ * @throws IOException if an I/O error occurred while reading the local Change */ public Optional<Long> getComputedChangeTs() throws IOException; + + /** + * Check if the local Change contains current patchset refs + * + * @return true if local change contains meta and current patchset refs + */ + public boolean isChangeConsistent(); }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeCheckerImpl.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeCheckerImpl.java index 08b26f7..4ef5b7f 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeCheckerImpl.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/index/ChangeCheckerImpl.java
@@ -19,6 +19,7 @@ import com.google.gerrit.exceptions.StorageException; import com.google.gerrit.server.CommentsUtil; import com.google.gerrit.server.change.ChangeFinder; +import com.google.gerrit.server.config.GerritInstanceId; import com.google.gerrit.server.git.GitRepositoryManager; import com.google.gerrit.server.notedb.ChangeNotes; import com.google.gerrit.server.util.ManualRequestContext; @@ -30,8 +31,11 @@ import java.sql.Timestamp; import java.util.Objects; import java.util.Optional; +import org.eclipse.jgit.errors.MissingObjectException; +import org.eclipse.jgit.lib.ObjectId; import org.eclipse.jgit.lib.Ref; import org.eclipse.jgit.lib.Repository; +import org.eclipse.jgit.revwalk.RevWalk; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -42,6 +46,7 @@ private final OneOffRequestContext oneOffReqCtx; private final String changeId; private final ChangeFinder changeFinder; + private final String instanceId; private Optional<Long> computedChangeTs = Optional.empty(); private Optional<ChangeNotes> changeNotes = Optional.empty(); @@ -55,12 +60,14 @@ CommentsUtil commentsUtil, ChangeFinder changeFinder, OneOffRequestContext oneOffReqCtx, + @GerritInstanceId String instanceId, @Assisted String changeId) { this.changeFinder = changeFinder; this.gitRepoMgr = gitRepoMgr; this.commentsUtil = commentsUtil; this.oneOffReqCtx = oneOffReqCtx; this.changeId = changeId; + this.instanceId = instanceId; } @Override @@ -69,7 +76,8 @@ return getComputedChangeTs() .map( ts -> { - ChangeIndexEvent event = new ChangeIndexEvent(projectName, changeId, deleted); + ChangeIndexEvent event = + new ChangeIndexEvent(projectName, changeId, deleted, instanceId); event.eventCreatedOn = ts; event.targetSha = getBranchTargetSha(); return event; @@ -140,6 +148,34 @@ } } + @Override + public boolean isChangeConsistent() { + Optional<ChangeNotes> notes = getChangeNotes(); + if (!notes.isPresent()) { + log.warn("Unable to compute change notes for change {}", changeId); + return true; + } + ObjectId currentPatchSetCommitId = notes.get().getCurrentPatchSet().commitId(); + try (Repository repo = gitRepoMgr.openRepository(changeNotes.get().getProjectName()); + RevWalk walk = new RevWalk(repo)) { + walk.parseCommit(currentPatchSetCommitId); + } catch (StorageException | MissingObjectException e) { + log.warn( + String.format( + "Consistency check failed for change %s, missing current patchset commit %s", + changeId, currentPatchSetCommitId.getName()), + e); + return false; + } catch (IOException e) { + log.warn( + String.format( + "Cannot check consistency for change %s, current patchset commit %s. Assuming change is consistent", + changeId, currentPatchSetCommitId.getName()), + e); + } + return true; + } + private Optional<Long> computeLastChangeTs() { return getChangeNotes().map(notes -> getTsFromChangeAndDraftComments(notes)); }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandler.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandler.java index eef3e4b..02f1b1c 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandler.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandler.java
@@ -21,6 +21,7 @@ import com.google.gerrit.extensions.events.GroupIndexedListener; import com.google.gerrit.extensions.events.ProjectIndexedListener; import com.google.gerrit.extensions.registration.DynamicSet; +import com.google.gerrit.server.config.GerritInstanceId; import com.google.inject.Inject; import com.googlesource.gerrit.plugins.multisite.forwarder.Context; import com.googlesource.gerrit.plugins.multisite.forwarder.ForwarderTask; @@ -48,6 +49,7 @@ private final ChangeCheckerImpl.Factory changeChecker; private final ProjectsFilter projectsFilter; private final GroupChecker groupChecker; + private final String instanceId; @Inject IndexEventHandler( @@ -55,18 +57,20 @@ DynamicSet<IndexEventForwarder> forwarders, ChangeCheckerImpl.Factory changeChecker, ProjectsFilter projectsFilter, - GroupChecker groupChecker) { + GroupChecker groupChecker, + @GerritInstanceId String instanceId) { this.forwarders = forwarders; this.executor = executor; this.changeChecker = changeChecker; this.projectsFilter = projectsFilter; this.groupChecker = groupChecker; + this.instanceId = instanceId; } @Override public void onAccountIndexed(int id) { if (!Context.isForwardedEvent()) { - IndexAccountTask task = new IndexAccountTask(new AccountIndexEvent(id)); + IndexAccountTask task = new IndexAccountTask(new AccountIndexEvent(id, instanceId)); if (queuedTasks.add(task)) { executor.execute(task); } @@ -87,7 +91,8 @@ public void onGroupIndexed(String groupUUID) { if (!Context.isForwardedEvent()) { IndexGroupTask task = - new IndexGroupTask(new GroupIndexEvent(groupUUID, groupChecker.getGroupHead(groupUUID))); + new IndexGroupTask( + new GroupIndexEvent(groupUUID, groupChecker.getGroupHead(groupUUID), instanceId)); if (queuedTasks.add(task)) { executor.execute(task); } @@ -97,7 +102,7 @@ @Override public void onProjectIndexed(String projectName) { if (!Context.isForwardedEvent() && projectsFilter.matches(projectName)) { - IndexProjectTask task = new IndexProjectTask(new ProjectIndexEvent(projectName)); + IndexProjectTask task = new IndexProjectTask(new ProjectIndexEvent(projectName, instanceId)); if (queuedTasks.add(task)) { executor.execute(task); } @@ -132,7 +137,7 @@ private void executeDeleteChangeTask(int id) { if (!Context.isForwardedEvent()) { - IndexChangeTask task = new IndexChangeTask(new ChangeIndexEvent("", id, true)); + IndexChangeTask task = new IndexChangeTask(new ChangeIndexEvent("", id, true, instanceId)); if (queuedTasks.add(task)) { executor.execute(task); }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilter.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilter.java new file mode 100644 index 0000000..d851420 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilter.java
@@ -0,0 +1,200 @@ +// Copyright (C) 2023 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.validation; + +import static com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF; +import static com.googlesource.gerrit.plugins.replication.pull.PullReplicationLogger.repLog; + +import com.gerritforge.gerrit.globalrefdb.GlobalRefDbLockException; +import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper; +import com.google.common.flogger.FluentLogger; +import com.google.gerrit.entities.Project; +import com.google.gerrit.entities.Project.NameKey; +import com.google.gerrit.server.git.GitRepositoryManager; +import com.google.inject.Inject; +import com.google.inject.Singleton; +import com.googlesource.gerrit.plugins.multisite.Configuration; +import com.googlesource.gerrit.plugins.replication.pull.ReplicationFetchFilter; +import java.io.IOException; +import java.util.Collections; +import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import org.eclipse.jgit.lib.ObjectId; +import org.eclipse.jgit.lib.Ref; +import org.eclipse.jgit.lib.RefDatabase; +import org.eclipse.jgit.lib.Repository; + +@Singleton +public class MultisiteReplicationFetchFilter implements ReplicationFetchFilter { + private static final String ZERO_ID_NAME = ObjectId.zeroId().name(); + private static final FluentLogger logger = FluentLogger.forEnclosingClass(); + + private final SharedRefDatabaseWrapper sharedRefDb; + private final GitRepositoryManager gitRepositoryManager; + private Configuration config; + + @Inject + public MultisiteReplicationFetchFilter( + SharedRefDatabaseWrapper sharedRefDb, + GitRepositoryManager gitRepositoryManager, + Configuration config) { + this.sharedRefDb = sharedRefDb; + this.gitRepositoryManager = gitRepositoryManager; + this.config = config; + } + + @Override + public Set<String> filter(String projectName, Set<String> refs) { + try (Repository repository = + gitRepositoryManager.openRepository(Project.nameKey(projectName))) { + RefDatabase refDb = repository.getRefDatabase(); + return refs.stream() + .filter(ref -> !hasBeenRemovedFromGlobalRefDb(projectName, ref)) + .filter( + ref -> { + if (shouldNotBeTrackedAnymoreOnGlobalRefDb(ref)) { + return true; + } + Optional<ObjectId> localRefOid = + getLocalSha1IfEqualsToExistingGlobalRefDb( + repository, projectName, refDb, ref, true); + localRefOid.ifPresent( + oid -> + repLog.info( + "{}:{}={} is already up-to-date with the shared-refdb and thus will NOT BE" + + " fetched", + projectName, + ref, + oid.getName())); + + return !localRefOid.isPresent(); + }) + .collect(Collectors.toSet()); + } catch (IOException ioe) { + String message = String.format("Error while opening project: '%s'", projectName); + repLog.error(message); + logger.atSevere().withCause(ioe).log(message); + return Collections.emptySet(); + } + } + + /* + * Since ac43a5f94c773c9db7a73d44035961d69d13fa53 the 'refs/multi-site/version' is + * not updated anymore on the global-refdb; however, the values stored already + * on the global-refdb could get in the way and prevent replication from happening + * as expected. + * + * Exclude the 'refs/multi-site/version' from local vs. global refdb checking + * pretending that the global-refdb for that ref did not exist. + */ + private boolean shouldNotBeTrackedAnymoreOnGlobalRefDb(String ref) { + return MULTI_SITE_VERSIONING_REF.equals(ref); + } + + /* If the ref to fetch has been set to all zeros on the global-refdb, it means + * that whatever is the situation locally, we do not need to fetch it: + * - If the remote still has it, fetching it will be useless because the global + * state is that the ref should be removed. + * - If the remote doesn't have it anymore, trying to fetch the ref won't do + * anything because you can't just remove local refs by fetching. + */ + private boolean hasBeenRemovedFromGlobalRefDb(String projectName, String ref) { + if (foundAsZeroInSharedRefDb(Project.nameKey(projectName), ref)) { + repLog.info( + "{}:{} is found as zeros (removed) in shared-refdb thus will NOT BE fetched", + projectName, + ref); + return true; + } + return false; + } + + private boolean foundAsZeroInSharedRefDb(NameKey projectName, String ref) { + return sharedRefDb + .get(projectName, ref, String.class) + .map(r -> ZERO_ID_NAME.equals(r)) + .orElse(false); + } + + private Optional<ObjectId> getLocalSha1IfEqualsToExistingGlobalRefDb( + Repository repository, + String projectName, + RefDatabase refDb, + String ref, + boolean retryWithRandomSleep) { + try { + Optional<ObjectId> localRefObjectId = + Optional.ofNullable(refDb.exactRef(ref)) + .filter( + r -> + sharedRefDb + .get(Project.nameKey(projectName), r.getName(), String.class) + .map(sharedRefObjId -> r.getObjectId().getName().equals(sharedRefObjId)) + .orElse(false)) + .map(Ref::getObjectId); + + if (!localRefObjectId.isPresent() && retryWithRandomSleep) { + randomSleepForMitigatingConditionWhereLocalRefHaveJustBeenChanged( + projectName, localRefObjectId, ref); + localRefObjectId = + getLocalSha1IfEqualsToExistingGlobalRefDb(repository, projectName, refDb, ref, false); + } + + return localRefObjectId; + } catch (GlobalRefDbLockException gle) { + String message = String.format("%s is locked on shared-refdb", ref); + repLog.error(message); + logger.atSevere().withCause(gle).log(message); + return Optional.empty(); + } catch (IOException ioe) { + String message = + String.format("Error while extracting ref '%s' for project '%s'", ref, projectName); + repLog.error(message); + logger.atSevere().withCause(ioe).log(message); + return Optional.empty(); + } + } + + private void randomSleepForMitigatingConditionWhereLocalRefHaveJustBeenChanged( + String projectName, Optional<ObjectId> refObjectId, String ref) { + if (!config.replicationFilter().isFetchFilterRandomSleepEnabled()) { + repLog.debug( + "'{}' is not up-to-date for project '{}' [local='{}']. Random sleep is disabled," + + " reload local ref without delay and re-check", + ref, + projectName, + refObjectId); + return; + } + + int randomSleepTimeMsec = config.replicationFilter().fetchFilterRandomSleepTimeMs(); + repLog.debug( + "'{}' is not up-to-date for project '{}' [local='{}']. Reload local ref in '{} ms' and" + + " re-check", + ref, + projectName, + refObjectId, + randomSleepTimeMsec); + try { + Thread.sleep(randomSleepTimeMsec); + } catch (InterruptedException ie) { + String message = + String.format("Error while waiting for next check for '%s', ref '%s'", projectName, ref); + repLog.error(message); + logger.atWarning().withCause(ie).log(message); + } + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilter.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilter.java index 98f4897..6f9c2e7 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilter.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilter.java
@@ -22,12 +22,13 @@ import com.google.gerrit.server.git.GitRepositoryManager; import com.google.inject.Inject; import com.google.inject.Singleton; +import com.googlesource.gerrit.plugins.multisite.Configuration; import com.googlesource.gerrit.plugins.replication.ReplicationPushFilter; import java.io.IOException; import java.util.Collections; import java.util.HashSet; import java.util.List; -import java.util.Random; +import java.util.Optional; import java.util.Set; import java.util.stream.Collectors; import org.eclipse.jgit.lib.ObjectId; @@ -42,20 +43,22 @@ public class MultisiteReplicationPushFilter implements ReplicationPushFilter { private static final FluentLogger logger = FluentLogger.forEnclosingClass(); private static final String REF_META_SUFFIX = "/meta"; - public static final int MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS = 1000; - public static final int RANDOM_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS = 1000; static final String REPLICATION_LOG_NAME = "replication_log"; static final Logger repLog = LoggerFactory.getLogger(REPLICATION_LOG_NAME); private final SharedRefDatabaseWrapper sharedRefDb; private final GitRepositoryManager gitRepositoryManager; + private Configuration config; @Inject public MultisiteReplicationPushFilter( - SharedRefDatabaseWrapper sharedRefDb, GitRepositoryManager gitRepositoryManager) { + SharedRefDatabaseWrapper sharedRefDb, + GitRepositoryManager gitRepositoryManager, + Configuration config) { this.sharedRefDb = sharedRefDb; this.gitRepositoryManager = gitRepositoryManager; + this.config = config; } @Override @@ -66,19 +69,23 @@ gitRepositoryManager.openRepository(Project.nameKey(projectName))) { List<RemoteRefUpdate> filteredRefUpdates = remoteUpdatesList.stream() - .filter( + .map( refUpdate -> { - boolean refUpToDate = isUpToDateWithRetry(projectName, repository, refUpdate); - if (!refUpToDate) { + Optional<RemoteRefUpdate> updatedRefUpdate = + isUpToDateWithRetry(projectName, repository, refUpdate); + if (!updatedRefUpdate.isPresent()) { repLog.warn( - "{} is not up-to-date with the shared-refdb and thus will NOT BE replicated", + "{} is not up-to-date with the shared-refdb and thus will NOT BE" + + " replicated", refUpdate); if (refUpdate.getSrcRef().endsWith(REF_META_SUFFIX)) { outdatedChanges.add(getRootChangeRefPrefix(refUpdate.getSrcRef())); } } - return refUpToDate; + return updatedRefUpdate; }) + .filter(Optional::isPresent) + .map(Optional::get) .collect(Collectors.toList()); return filteredRefUpdates.stream() @@ -102,45 +109,67 @@ } } - private boolean isUpToDateWithRetry( + private Optional<RemoteRefUpdate> isUpToDateWithRetry( String projectName, Repository repository, RemoteRefUpdate refUpdate) { String ref = refUpdate.getSrcRef(); try { if (sharedRefDb.isUpToDate( Project.nameKey(projectName), new ObjectIdRef.Unpeeled(Ref.Storage.NETWORK, ref, refUpdate.getNewObjectId()))) { - return true; + return Optional.of(refUpdate); } randomSleepForMitigatingConditionWhereLocalRefHaveJustBeenChanged( projectName, refUpdate, ref); + ObjectId reloadedNewObjectId = getNotNullExactRef(repository, ref); + RemoteRefUpdate refUpdateReloaded = + newRemoteRefUpdateWithObjectId(repository, refUpdate, reloadedNewObjectId); return sharedRefDb.isUpToDate( - Project.nameKey(projectName), - new ObjectIdRef.Unpeeled(Ref.Storage.NETWORK, ref, getNotNullExactRef(repository, ref))); + Project.nameKey(projectName), + new ObjectIdRef.Unpeeled( + Ref.Storage.NETWORK, ref, refUpdateReloaded.getNewObjectId())) + ? Optional.of(refUpdateReloaded) + : Optional.empty(); } catch (GlobalRefDbLockException gle) { String message = String.format("%s is locked on shared-refdb and thus will NOT BE replicated", ref); repLog.error(message); logger.atSevere().withCause(gle).log(message); - return false; + return Optional.empty(); } catch (IOException ioe) { String message = String.format("Error while extracting ref '%s' for project '%s'", ref, projectName); repLog.error(message); logger.atSevere().withCause(ioe).log(message); - return false; + return Optional.empty(); } } + private RemoteRefUpdate newRemoteRefUpdateWithObjectId( + Repository localDb, RemoteRefUpdate refUpdate, ObjectId reloadedNewObjectId) + throws IOException { + return new RemoteRefUpdate( + localDb, + refUpdate.getSrcRef(), + reloadedNewObjectId, + refUpdate.getRemoteName(), + refUpdate.isForceUpdate(), + null, + refUpdate.getExpectedOldObjectId()); + } + private void randomSleepForMitigatingConditionWhereLocalRefHaveJustBeenChanged( String projectName, RemoteRefUpdate refUpdate, String ref) { - int randomSleepTimeMsec = - MIN_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS - + new Random().nextInt(RANDOM_WAIT_BEFORE_RELOAD_LOCAL_VERSION_MS); + if (!config.replicationFilter().isPushFilterRandomSleepEnabled()) { + return; + } + + int randomSleepTimeMsec = config.replicationFilter().pushFilterRandomSleepTimeMs(); repLog.debug( String.format( - "'%s' is not up-to-date for project '%s' [local='%s']. Reload local ref in '%d ms' and re-check", + "'%s' is not up-to-date for project '%s' [local='%s']. Reload local ref in '%d ms' and" + + " re-check", ref, projectName, refUpdate.getNewObjectId(), randomSleepTimeMsec)); try { Thread.sleep(randomSleepTimeMsec);
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateImpl.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateImpl.java index 469122b..e07fef0 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateImpl.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateImpl.java
@@ -20,27 +20,27 @@ import com.gerritforge.gerrit.globalrefdb.GlobalRefDbSystemError; import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter; import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper; +import com.google.common.base.Predicates; import com.google.common.collect.ImmutableSet; import com.google.common.flogger.FluentLogger; import com.google.gerrit.entities.Project; +import com.google.gerrit.entities.Project.NameKey; import com.google.gerrit.entities.RefNames; import com.google.gerrit.server.events.Event; import com.google.gerrit.server.events.EventListener; import com.google.gerrit.server.events.RefUpdatedEvent; import com.google.gerrit.server.extensions.events.GitReferenceUpdated; import com.google.gerrit.server.git.GitRepositoryManager; -import com.google.gerrit.server.notedb.IntBlob; import com.google.inject.Inject; import com.googlesource.gerrit.plugins.multisite.ProjectVersionLogger; import com.googlesource.gerrit.plugins.multisite.forwarder.Context; import java.io.IOException; +import java.nio.charset.StandardCharsets; import java.util.Optional; import java.util.Set; import org.eclipse.jgit.errors.RepositoryNotFoundException; import org.eclipse.jgit.lib.ObjectId; -import org.eclipse.jgit.lib.ObjectIdRef; import org.eclipse.jgit.lib.ObjectInserter; -import org.eclipse.jgit.lib.Ref; import org.eclipse.jgit.lib.RefUpdate; import org.eclipse.jgit.lib.Repository; @@ -106,8 +106,7 @@ if (newProjectVersionRefUpdate.isPresent()) { verLogger.log(projectNameKey, newVersion, 0L); - if (updateSharedProjectVersion( - projectNameKey, newProjectVersionRefUpdate.get().getNewObjectId(), newVersion)) { + if (updateSharedProjectVersion(projectNameKey, newVersion)) { gitReferenceUpdated.fire(projectNameKey, newProjectVersionRefUpdate.get(), null); } } else { @@ -137,20 +136,9 @@ return newId; } - private boolean updateSharedProjectVersion( - Project.NameKey projectNameKey, ObjectId newObjectId, Long newVersion) + private boolean updateSharedProjectVersion(Project.NameKey projectNameKey, Long newVersion) throws SharedProjectVersionUpdateException { - Ref sharedRef = - sharedRefDb - .get(projectNameKey, MULTI_SITE_VERSIONING_REF, String.class) - .map( - (String objectId) -> - new ObjectIdRef.Unpeeled( - Ref.Storage.NEW, MULTI_SITE_VERSIONING_REF, ObjectId.fromString(objectId))) - .orElse( - new ObjectIdRef.Unpeeled( - Ref.Storage.NEW, MULTI_SITE_VERSIONING_REF, ObjectId.zeroId())); Optional<Long> sharedVersion = sharedRefDb .get(projectNameKey, MULTI_SITE_VERSIONING_VALUE_REF, String.class) @@ -160,59 +148,50 @@ if (sharedVersion.isPresent() && sharedVersion.get() >= newVersion) { logger.atWarning().log( String.format( - "NOT Updating project %s version %s (value=%d) in shared ref-db because is more recent than the local one %s (value=%d) ", - projectNameKey.get(), - newObjectId, - newVersion, - sharedRef.getObjectId().getName(), - sharedVersion.get())); + "NOT Updating project %s value=%d in shared ref-db because is more recent than the local value=%d", + projectNameKey.get(), newVersion, sharedVersion.get())); return false; } logger.atFine().log( String.format( - "Updating shared project %s version to %s (value=%d)", - projectNameKey.get(), newObjectId, newVersion)); + "Updating shared project %s value to %d", projectNameKey.get(), newVersion)); - boolean success = sharedRefDb.compareAndPut(projectNameKey, sharedRef, newObjectId); - if (!success) { - String message = - String.format( - "Project version blob update failed for %s. Current value %s, new value: %s", - projectNameKey.get(), safeGetObjectId(sharedRef), newObjectId); - logger.atSevere().log(message); - throw new SharedProjectVersionUpdateException(message); - } - - success = - sharedRefDb.compareAndPut( - projectNameKey, - MULTI_SITE_VERSIONING_VALUE_REF, - sharedVersion.map(Object::toString).orElse(null), - newVersion.toString()); - if (!success) { - String message = - String.format( - "Project version update failed for %s. Current value %s, new value: %s", - projectNameKey.get(), safeGetObjectId(sharedRef), newObjectId); - logger.atSevere().log(message); - throw new SharedProjectVersionUpdateException(message); - } - + updateProjectVersionValue(projectNameKey, newVersion, sharedVersion); return true; } catch (GlobalRefDbSystemError refDbSystemError) { String message = String.format( - "Error while updating shared project version for %s. Current value %s, new value: %s. Error: %s", + "Error while updating shared project value for %s. Current value %s, new value: %s. Error: %s", projectNameKey.get(), - sharedRef.getObjectId(), - newObjectId, + sharedVersion.map(Object::toString).orElse(null), + newVersion, refDbSystemError.getMessage()); logger.atSevere().withCause(refDbSystemError).log(message); throw new SharedProjectVersionUpdateException(message); } } + private void updateProjectVersionValue( + NameKey projectNameKey, Long newVersion, Optional<Long> sharedVersion) { + try { + if (sharedRefDb.isSetOperationSupported()) { + sharedRefDb.put(projectNameKey, MULTI_SITE_VERSIONING_VALUE_REF, newVersion.toString()); + return; + } + } catch (NoSuchMethodError e) { + logger.atSevere().log( + "Global-refdb library is outdated and is not supporting " + + "'put' method, update global-refdb to the newest version. Falling back to 'compareAndPut'"); + } + + sharedRefDb.compareAndPut( + projectNameKey, + MULTI_SITE_VERSIONING_VALUE_REF, + sharedVersion.map(Object::toString).orElse(null), + newVersion.toString()); + } + /* (non-Javadoc) * @see com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate#getProjectLocalVersion(java.lang.String) */ @@ -220,9 +199,9 @@ public Optional<Long> getProjectLocalVersion(String projectName) { try (Repository repository = gitRepositoryManager.openRepository(Project.NameKey.parse(projectName))) { - Optional<IntBlob> blob = IntBlob.parse(repository, MULTI_SITE_VERSIONING_REF); + Optional<Long> blob = longBlobParse(repository, MULTI_SITE_VERSIONING_REF); if (blob.isPresent()) { - Long repoVersion = Integer.toUnsignedLong(blob.get().value()); + Long repoVersion = blob.get(); logger.atFine().log("Local project '%s' has version %d", projectName, repoVersion); return Optional.of(repoVersion); } @@ -234,6 +213,22 @@ return Optional.empty(); } + private Optional<Long> longBlobParse(Repository repo, String refName) throws IOException { + return Optional.ofNullable(repo.exactRef(refName)) + .map( + (ref) -> { + try { + return Long.parseLong( + new String(repo.open(ref.getObjectId()).getBytes(), StandardCharsets.UTF_8)); + } catch (IOException e) { + logger.atSevere().withCause(e).log( + "Unable to extract long BLOB from %s:%s", repo.getDirectory(), ref); + return null; + } + }) + .filter(Predicates.notNull()); + } + /* (non-Javadoc) * @see com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate#getProjectRemoteVersion(java.lang.String) */ @@ -245,10 +240,6 @@ return globalVersion.flatMap(longString -> getLongValueOf(longString)); } - private Object safeGetObjectId(Ref currentRef) { - return currentRef == null ? "null" : currentRef.getObjectId(); - } - private Optional<Long> getLongValueOf(String longString) { try { return Optional.ofNullable(Long.parseLong(longString)); @@ -286,7 +277,7 @@ } private long getCurrentGlobalVersionNumber() { - return System.currentTimeMillis() / 1000; + return System.currentTimeMillis(); } private Boolean isSuccessful(RefUpdate.Result result) {
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/PullReplicationFilterModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/PullReplicationFilterModule.java new file mode 100644 index 0000000..b8ad99d --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/PullReplicationFilterModule.java
@@ -0,0 +1,28 @@ +// Copyright (C) 2023 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.validation; + +import com.google.gerrit.extensions.registration.DynamicItem; +import com.google.inject.AbstractModule; +import com.googlesource.gerrit.plugins.replication.pull.ReplicationFetchFilter; + +public class PullReplicationFilterModule extends AbstractModule { + + @Override + protected void configure() { + DynamicItem.bind(binder(), ReplicationFetchFilter.class) + .to(MultisiteReplicationFetchFilter.class); + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/PushReplicationFilterModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/PushReplicationFilterModule.java new file mode 100644 index 0000000..8a8e826 --- /dev/null +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/PushReplicationFilterModule.java
@@ -0,0 +1,28 @@ +// Copyright (C) 2023 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.validation; + +import com.google.gerrit.extensions.registration.DynamicItem; +import com.google.inject.AbstractModule; +import com.googlesource.gerrit.plugins.replication.ReplicationPushFilter; + +public class PushReplicationFilterModule extends AbstractModule { + + @Override + protected void configure() { + DynamicItem.bind(binder(), ReplicationPushFilter.class) + .to(MultisiteReplicationPushFilter.class); + } +}
diff --git a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ValidationModule.java b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ValidationModule.java index 8601cf2..fc4d505 100644 --- a/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ValidationModule.java +++ b/src/main/java/com/googlesource/gerrit/plugins/multisite/validation/ValidationModule.java
@@ -30,14 +30,11 @@ import com.gerritforge.gerrit.globalrefdb.validation.dfsrefdb.SharedRefEnforcement; import com.google.common.collect.ImmutableSet; import com.google.gerrit.extensions.config.FactoryModule; -import com.google.gerrit.extensions.registration.DynamicItem; import com.google.gerrit.server.config.RepositoryConfig; import com.google.inject.Scopes; import com.google.inject.TypeLiteral; import com.google.inject.name.Names; import com.googlesource.gerrit.plugins.multisite.Configuration; -import com.googlesource.gerrit.plugins.replication.ReplicationExtensionPointModule; -import com.googlesource.gerrit.plugins.replication.ReplicationPushFilter; public class ValidationModule extends FactoryModule { private final Configuration cfg; @@ -50,8 +47,6 @@ @Override protected void configure() { - install(new ReplicationExtensionPointModule()); - bind(SharedRefDatabaseWrapper.class).in(Scopes.SINGLETON); bind(SharedRefLogger.class).to(Log4jSharedRefLogger.class); factory(LockWrapper.Factory.class); @@ -70,8 +65,6 @@ ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF, ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_VALUE_REF)); install(new RepositoryManagerModule(repoConfig)); - DynamicItem.bind(binder(), ReplicationPushFilter.class) - .to(MultisiteReplicationPushFilter.class); if (cfg.getSharedRefDbConfiguration().getSharedRefDb().getEnforcementRules().isEmpty()) { bind(SharedRefEnforcement.class).to(DefaultSharedRefEnforcement.class).in(Scopes.SINGLETON);
diff --git a/src/main/resources/Documentation/about.md b/src/main/resources/Documentation/about.md index 623871b..d7c5648 100644 --- a/src/main/resources/Documentation/about.md +++ b/src/main/resources/Documentation/about.md
@@ -7,7 +7,8 @@ message broker for aligning with the other masters over different sites. The masters must be: - +* Gerrit instance id is mandatory for @PLUGIN@ plugin. All the master + must have [gerrit.instanceId](https://gerrit-review.googlesource.com/Documentation/config-gerrit.html#gerrit) populated. * events-broker library must be installed as a library module in the `$GERRIT_SITE/lib` directory of all the masters * global-refdb library must be installed as a library module in the @@ -98,3 +99,7 @@ * Subscriber replication lag (sec behind the producer) `metric=site/multi_site/subscriber/subscriber_replication_status/sec_behind, type=com.google.gerrit.metrics.dropwizard.CallbackMetricImpl` + +* Subscriber replication lag (millisec behind the producer) + +`metric=site/multi_site/subscriber/subscriber_replication_status/msec_behind, type=com.google.gerrit.metrics.dropwizard.CallbackMetricImpl`
diff --git a/src/main/resources/Documentation/build.md b/src/main/resources/Documentation/build.md index bb38288..146341e 100644 --- a/src/main/resources/Documentation/build.md +++ b/src/main/resources/Documentation/build.md
@@ -3,15 +3,29 @@ This plugin can be built with Bazel in the Gerrit tree. Clone or link this plugin to the plugins directory of Gerrit's -source tree. Put the external dependency Bazel build file into -the Gerrit /plugins directory, replacing the existing empty one. +source tree. + +Clone the [pull-replication](https://gerrit.googlesource.com/plugins/pull-replication) on +the same branch of the @PLUGIN@ plugin and link it to the `gerrit/plugins` directory. ``` + export BRANCH=$(git --git-dir=@PLUGIN@ branch) + git clone https://gerrit.googlesource.com/plugins/pull-replication cd gerrit/plugins rm external_plugin_deps.bzl ln -s @PLUGIN@/external_plugin_deps.bzl . ``` +Clone the [global-refdb](git clone "https://gerrit.googlesource.com/modules/global-refdb") on +the same branch of the @PLUGIN@ plugin and link it to the `gerrit/plugins` directory. + +``` + export BRANCH=$(git --git-dir=@PLUGIN@ branch) + git clone "https://gerrit.googlesource.com/modules/global-refdb" + cd gerrit/plugins + ln -s ../../global-refdb . +``` + From the Gerrit source tree issue the command: ```
diff --git a/src/main/resources/Documentation/config.md b/src/main/resources/Documentation/config.md index 02806a1..e451cb3 100644 --- a/src/main/resources/Documentation/config.md +++ b/src/main/resources/Documentation/config.md
@@ -3,15 +3,19 @@ ========================= The @PLUGIN@ plugin must be installed as a library module in the -`$GERRIT_SITE/lib` folder of all the instances. Configuration should -be specified in the `$site_path/etc/@PLUGIN@.config` file. +`$GERRIT_SITE/lib` folder of all the instances and linked to +`$GERRIT_SITE/plugins`, which enable to use it as both libModule +and plugin. +Configuration should be specified in the `$site_path/etc/@PLUGIN@.config` file. ## Configuration parameters ```cache.synchronize``` -: Whether to run cache evictions synchronously. It requires disabling the - background cache evictions notifications in `gerrit.config` by setting - `cache.threads = 0`. +: Whether to synchronize cache evictions. Set to false when relying on + low cache TTLs and therefore cache eviction is not strictly needed. + It requires disabling the background cache evictions notifications in + `gerrit.config` by setting `cache.threads = 0`. + Defaults to true. ```cache.threadPoolSize``` @@ -29,7 +33,8 @@ forwarded. ```event.synchronize``` -: Whether to synchronize stream events. +: Whether to synchronize stream events. Set to false when not using the SSH + stream events. Defaults to true. ```index.numStripedLocks``` @@ -37,7 +42,9 @@ Defaults to 10 ```index.synchronize``` -: Whether to synchronize secondary indexes. +: Whether to synchronize secondary indexes. Set to false when using multi-site + on Gerrit replicas that do not have an index, or when using an external + service such as ElasticSearch. Defaults to true. ```index.threadPoolSize``` @@ -72,10 +79,19 @@ : Name of the topic to use for publishing cache eviction events Defaults to GERRIT.EVENT.PROJECT.LIST +**NOTE**: All broker settings are ignored when all of the `cache`, +`index` or `event` synchronization is disabled. + ```ref-database.enabled``` : Enable the use of a shared ref-database Defaults: true +```ref-database.replicationLagRefreshInterval``` +: Enable the auto-refresh of the metrics to trace the auto-replication + lag by polling on a regular basis. Set to zero for disabling the polling + mechanism. + Defaults: 60 min + ```ref-database.enforcementRules.<policy>``` : Level of consistency enforcement across sites on a project:refs basis. Supports two values for enforcing the policy on multiple projects or refs. @@ -123,3 +139,80 @@ the project `foo/bar`, but no other project. By default, all projects are matched. + +```replication.push-filter.minWaitBeforeReloadLocalVersionMs``` +: Specifies the minimum amount of time in milliseconds replication plugin filter will + wait before retrying check for ref which is not up to date with global-refdb. + + By default: 1000 milliseconds + +```replication.push-filter.maxRandomWaitBeforeReloadLocalVersionMs``` +: Specifies the additional amount of time in milliseconds replication filter will + wait before retrying check for ref which is not up to date with global-refdb. + + If maxRandomWaitBeforeReloadLocalVersionMs is set to zero random sleep for not in sync + refs is disabled. + + By default: 1000 milliseconds + +```replication.fetch-filter.minWaitBeforeReloadLocalVersionMs``` +: Specifies the minimum amount of time in milliseconds pull-replication filter wait + before retrying check for ref which is not up to date with global-refdb. + + By default: 1000 milliseconds + +```replication.fetch-filter.maxRandomWaitBeforeReloadLocalVersionMs``` +: Specifies the additional amount of time in milliseconds pull-replication filter will + wait before retrying check for ref which is not up to date with global-refdb. + + If maxRandomWaitBeforeReloadLocalVersionMs is set to zero random sleep for not in sync + refs is disabled. + + By default: 1000 milliseconds + +## Replication filters + +The @PLUGIN@ plugin is also responsible for filtering out replication events that may +risk to create a split-brain situation. +It integrates the push and pull replication filtering extension points for validating +the refs to be replicated and dropping some of them. + +**Replication plugin** + +When using the Gerrit core replication plugin, also known as push-replication, link the +`replication.jar` to the `$GERRIT_SITE/lib` directory and add the following libModule +to `gerrit.config`: + +``` +[gerrit] + installModule = com.googlesource.gerrit.plugins.replication.ReplicationExtensionPointModule +``` + +The above configuration would be automatically detected by the @PLUGIN@ plugin which would then +install the PushReplicationFilterModule for filtering outgoing replication refs based +on their global-refdb status: + +- Outgoing replication of refs that are NOT up-to-date with the global-refdb will be + discarded, because they may cause split-brain on the remote replication endpoints. + +- All other refs will be pushed as normal to the remote replication ends. + +**Pull-replication plugin** + +When using the [pull-replication](https://gerrit.googlesource.com/plugins/pull-replication) +plugin, link the `pull-replication.jar` to the `$GERRIT_SITE/lib` directory and add the following +two libModules to `gerrit.config`: + +``` +[gerrit] + installModule = com.googlesource.gerrit.plugins.replication.pull.ReplicationExtensionPointModule +``` + +The above configuration would be automatically detected by the @PLUGIN@ plugin which would then +install the PullReplicationFilterModule for filtering incoming fetch replication refs based +on their global-refdb status. + +- Incoming replication of refs that locally are already up-to-date with the global-refdb will be + discarded, because they would not add anything more to the current status of the local refs. + +- All other refs will be fetched as normal from the replication sources.
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/ModuleTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/ModuleTest.java deleted file mode 100644 index 2df60dd..0000000 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/ModuleTest.java +++ /dev/null
@@ -1,65 +0,0 @@ -// Copyright (C) 2017 The Android Open Source Project -// -// Licensed under the Apache License, Version 2.0 (the "License"); -// you may not use this file except in compliance with the License. -// You may obtain a copy of the License at -// -// http://www.apache.org/licenses/LICENSE-2.0 -// -// Unless required by applicable law or agreed to in writing, software -// distributed under the License is distributed on an "AS IS" BASIS, -// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -// See the License for the specific language governing permissions and -// limitations under the License. - -package com.googlesource.gerrit.plugins.multisite; - -import static com.google.common.truth.Truth.assertThat; - -import com.google.gerrit.server.config.SitePaths; -import java.io.File; -import java.nio.file.Path; -import java.nio.file.Paths; -import java.util.UUID; -import org.junit.Before; -import org.junit.Rule; -import org.junit.Test; -import org.junit.rules.TemporaryFolder; -import org.junit.runner.RunWith; -import org.mockito.Answers; -import org.mockito.Mock; -import org.mockito.junit.MockitoJUnitRunner; - -@RunWith(MockitoJUnitRunner.class) -public class ModuleTest { - - @Mock(answer = Answers.RETURNS_DEEP_STUBS) - private Configuration configMock; - - @Rule public TemporaryFolder tempFolder = new TemporaryFolder(); - - private Module module; - - @Before - public void setup() { - module = new Module(configMock); - } - - @Test - public void shouldGetInstanceId() throws Exception { - File tmpSitePath = tempFolder.newFolder(); - File tmpPluginDataPath = - Paths.get(tmpSitePath.getPath(), "data", Configuration.PLUGIN_NAME).toFile(); - tmpPluginDataPath.mkdirs(); - Path path = Paths.get(tmpPluginDataPath.getPath(), Configuration.INSTANCE_ID_FILE); - SitePaths sitePaths = new SitePaths(Paths.get(tmpSitePath.getPath())); - assertThat(path.toFile().exists()).isFalse(); - - UUID gotUUID1 = module.getInstanceId(sitePaths); - assertThat(gotUUID1).isNotNull(); - assertThat(path.toFile().exists()).isTrue(); - - UUID gotUUID2 = module.getInstanceId(sitePaths); - assertThat(gotUUID1).isEqualTo(gotUUID2); - } -}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapperTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapperTest.java index 92fa101..7d1751c 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapperTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/broker/BrokerApiWrapperTest.java
@@ -1,15 +1,19 @@ package com.googlesource.gerrit.plugins.multisite.broker; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.only; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import com.gerritforge.gerrit.eventbroker.BrokerApi; +import com.google.common.util.concurrent.MoreExecutors; +import com.google.common.util.concurrent.SettableFuture; import com.google.gerrit.extensions.registration.DynamicItem; import com.google.gerrit.server.events.Event; +import com.google.gerrit.server.events.ProjectCreatedEvent; import com.googlesource.gerrit.plugins.multisite.MessageLogger; -import java.util.UUID; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -22,28 +26,35 @@ @Mock private BrokerApi brokerApi; @Mock Event event; @Mock MessageLogger msgLog; - private UUID instanceId = UUID.randomUUID(); private String topic = "index"; private BrokerApiWrapper objectUnderTest; @Before public void setUp() { + event.instanceId = "instance-id"; objectUnderTest = new BrokerApiWrapper( - DynamicItem.itemOf(BrokerApi.class, brokerApi), brokerMetrics, msgLog, instanceId); + MoreExecutors.directExecutor(), + DynamicItem.itemOf(BrokerApi.class, brokerApi), + brokerMetrics, + msgLog); } @Test public void shouldIncrementBrokerMetricCounterWhenMessagePublished() { - when(brokerApi.send(any(), any())).thenReturn(true); + SettableFuture<Boolean> resultF = SettableFuture.create(); + resultF.set(true); + when(brokerApi.send(any(), any())).thenReturn(resultF); objectUnderTest.send(topic, event); verify(brokerMetrics, only()).incrementBrokerPublishedMessage(); } @Test public void shouldIncrementBrokerFailedMetricCounterWhenMessagePublishingFailed() { - when(brokerApi.send(any(), any())).thenReturn(false); + SettableFuture<Boolean> resultF = SettableFuture.create(); + resultF.setException(new Exception("Force Future failure")); + when(brokerApi.send(any(), any())).thenReturn(resultF); objectUnderTest.send(topic, event); verify(brokerMetrics, only()).incrementBrokerFailedToPublishMessage(); } @@ -53,10 +64,26 @@ when(brokerApi.send(any(), any())) .thenThrow(new RuntimeException("Unexpected runtime exception")); try { - objectUnderTest.send(topic, event); + objectUnderTest.sendSync(topic, event); } catch (RuntimeException e) { // expected } verify(brokerMetrics, only()).incrementBrokerFailedToPublishMessage(); } + + @Test + public void shouldSkipMessageSendingWhenInstanceIdIsNull() { + ProjectCreatedEvent event = new ProjectCreatedEvent(); + event.instanceId = null; + objectUnderTest.send(topic, event); + verify(brokerApi, never()).send(any(), eq(event)); + } + + @Test + public void shouldSkipMessageSendingWhenInstanceIdIsEmpty() { + ProjectCreatedEvent event = new ProjectCreatedEvent(); + event.instanceId = ""; + objectUnderTest.send(topic, event); + verify(brokerApi, never()).send(any(), eq(event)); + } }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandlerTest.java index 67be583..ce222d0 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandlerTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/CacheEvictionHandlerTest.java
@@ -36,9 +36,10 @@ @Test public void shouldNotPublishAccountsCacheEvictions() { - + String instanceId = "instance-id"; final CacheEvictionHandler<String, String> handler = - new CacheEvictionHandler<>(DynamicSet.emptySet(), executorMock, defaultCacheMatcher); + new CacheEvictionHandler<>( + DynamicSet.emptySet(), executorMock, defaultCacheMatcher, instanceId); handler.onRemoval( "test", "accounts", RemovalNotification.create("test", "accounts", RemovalCause.EXPLICIT));
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandlerTest.java index 4263ddb..c8216bd 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandlerTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/cache/ProjectListUpdateHandlerTest.java
@@ -41,6 +41,8 @@ @RunWith(MockitoJUnitRunner.class) public class ProjectListUpdateHandlerTest { + private static final String INSTANCE_ID = "instance-id"; + private ProjectListUpdateHandler handler; @Mock private ProjectListUpdateForwarder forwarder; @@ -51,7 +53,7 @@ when(projectsFilter.matches(any(String.class))).thenReturn(true); handler = new ProjectListUpdateHandler( - asDynamicSet(forwarder), MoreExecutors.directExecutor(), projectsFilter); + asDynamicSet(forwarder), MoreExecutors.directExecutor(), projectsFilter, INSTANCE_ID); } private DynamicSet<ProjectListUpdateForwarder> asDynamicSet( @@ -69,7 +71,8 @@ handler.onNewProjectCreated(event); verify(forwarder) .updateProjectList( - any(ProjectListUpdateTask.class), eq(new ProjectListUpdateEvent(projectName, false))); + any(ProjectListUpdateTask.class), + eq(new ProjectListUpdateEvent(projectName, false, INSTANCE_ID))); } @Test @@ -80,7 +83,8 @@ handler.onProjectDeleted(event); verify(forwarder) .updateProjectList( - any(ProjectListUpdateTask.class), eq(new ProjectListUpdateEvent(projectName, true))); + any(ProjectListUpdateTask.class), + eq(new ProjectListUpdateEvent(projectName, true, INSTANCE_ID))); } @Test @@ -101,18 +105,22 @@ handler.onNewProjectCreated(event); verify(forwarder, never()) .updateProjectList( - any(ProjectListUpdateTask.class), eq(new ProjectListUpdateEvent(projectName, true))); + any(ProjectListUpdateTask.class), + eq(new ProjectListUpdateEvent(projectName, true, INSTANCE_ID))); } @Test public void testProjectUpdateTaskToString() throws Exception { String projectName = "someProjectName"; ProjectListUpdateTask task = - handler.new ProjectListUpdateTask(new ProjectListUpdateEvent(projectName, false)); + handler + .new ProjectListUpdateTask(new ProjectListUpdateEvent(projectName, false, INSTANCE_ID)); assertThat(task.toString()) .isEqualTo(String.format("Update project list in target instance: add '%s'", projectName)); - task = handler.new ProjectListUpdateTask(new ProjectListUpdateEvent(projectName, true)); + task = + handler + .new ProjectListUpdateTask(new ProjectListUpdateEvent(projectName, true, INSTANCE_ID)); assertThat(task.toString()) .isEqualTo( String.format("Update project list in target instance: remove '%s'", projectName));
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubscriberTestBase.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubscriberTestBase.java new file mode 100644 index 0000000..cb1f78e --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/AbstractSubscriberTestBase.java
@@ -0,0 +1,162 @@ +// Copyright (C) 2021 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.consumer; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter; +import com.google.gerrit.extensions.registration.DynamicSet; +import com.google.gerrit.server.events.Event; +import com.google.gerrit.server.permissions.PermissionBackendException; +import com.googlesource.gerrit.plugins.multisite.Configuration; +import com.googlesource.gerrit.plugins.multisite.Configuration.Broker; +import com.googlesource.gerrit.plugins.multisite.MessageLogger; +import com.googlesource.gerrit.plugins.multisite.forwarder.CacheNotFoundException; +import com.googlesource.gerrit.plugins.multisite.forwarder.router.ForwardedEventRouter; +import java.io.IOException; +import java.util.List; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.class) +@Ignore +public abstract class AbstractSubscriberTestBase { + protected static final String NODE_INSTANCE_ID = "node-instance-id"; + protected static final String INSTANCE_ID = "other-node-instance-id"; + protected static final String PROJECT_NAME = "project-name"; + + @Mock protected DroppedEventListener droppedEventListeners; + @Mock protected MessageLogger msgLog; + @Mock protected SubscriberMetrics subscriberMetrics; + @Mock protected Configuration cfg; + @Mock protected Broker brokerCfg; + @Mock protected ProjectsFilter projectsFilter; + + @SuppressWarnings("rawtypes") + protected ForwardedEventRouter eventRouter; + + protected AbstractSubcriber objectUnderTest; + + @Before + public void setup() { + when(cfg.broker()).thenReturn(brokerCfg); + when(brokerCfg.getTopic(any(), any())).thenReturn("test-topic"); + eventRouter = eventRouter(); + objectUnderTest = objectUnderTest(); + } + + @Test + public void shouldConsumeEventsWhenNotFilteredByProjectName() + throws IOException, PermissionBackendException, CacheNotFoundException { + for (Event event : events()) { + when(projectsFilter.matches(any(String.class))).thenReturn(true); + objectUnderTest.getConsumer().accept(event); + verifyConsumed(event); + } + } + + @Test + public void shouldSkipEventsWhenFilteredByProjectName() + throws IOException, PermissionBackendException, CacheNotFoundException { + for (Event event : events()) { + when(projectsFilter.matches(any(String.class))).thenReturn(false); + objectUnderTest.getConsumer().accept(event); + verifySkipped(event); + } + } + + @SuppressWarnings("unchecked") + @Test + public void shouldSkipLocalEvents() + throws IOException, PermissionBackendException, CacheNotFoundException { + for (Event event : events()) { + event.instanceId = NODE_INSTANCE_ID; + when(projectsFilter.matches(any(String.class))).thenReturn(true); + + objectUnderTest.getConsumer().accept(event); + + verify(projectsFilter, never()).matches(PROJECT_NAME); + verify(eventRouter, never()).route(event); + verify(droppedEventListeners, times(1)).onEventDropped(event); + reset(projectsFilter, eventRouter, droppedEventListeners); + } + } + + @Test + public void shouldUpdateReplicationMetricsWithLocalEvents() { + for (Event event : events()) { + event.instanceId = NODE_INSTANCE_ID; + when(projectsFilter.matches(any(String.class))).thenReturn(true); + + objectUnderTest.getConsumer().accept(event); + + verify(subscriberMetrics, times(1)).updateReplicationStatusMetrics(event); + reset(projectsFilter, eventRouter, droppedEventListeners, subscriberMetrics); + } + } + + @Test + public void shouldUpdateReplicationMetricsWithNonLocalEvents() { + for (Event event : events()) { + event.instanceId = INSTANCE_ID; + when(projectsFilter.matches(any(String.class))).thenReturn(true); + + objectUnderTest.getConsumer().accept(event); + + verify(subscriberMetrics, times(1)).updateReplicationStatusMetrics(event); + reset(projectsFilter, eventRouter, droppedEventListeners, subscriberMetrics); + } + } + + protected abstract AbstractSubcriber objectUnderTest(); + + protected abstract List<Event> events(); + + @SuppressWarnings("rawtypes") + protected abstract ForwardedEventRouter eventRouter(); + + @SuppressWarnings("unchecked") + protected void verifySkipped(Event event) + throws IOException, PermissionBackendException, CacheNotFoundException { + verify(projectsFilter, times(1)).matches(PROJECT_NAME); + verify(eventRouter, never()).route(event); + verify(droppedEventListeners, times(1)).onEventDropped(event); + reset(projectsFilter, eventRouter, droppedEventListeners); + } + + @SuppressWarnings("unchecked") + protected void verifyConsumed(Event event) + throws IOException, PermissionBackendException, CacheNotFoundException { + verify(projectsFilter, times(1)).matches(PROJECT_NAME); + verify(eventRouter, times(1)).route(event); + verify(droppedEventListeners, never()).onEventDropped(event); + reset(projectsFilter, eventRouter, droppedEventListeners); + } + + protected DynamicSet<DroppedEventListener> asDynamicSet(DroppedEventListener listener) { + DynamicSet<DroppedEventListener> result = new DynamicSet<>(); + result.add("multi-site", listener); + return result; + } +}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriberTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriberTest.java new file mode 100644 index 0000000..5031f36 --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/BatchIndexEventSubscriberTest.java
@@ -0,0 +1,77 @@ +// Copyright (C) 2021 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.consumer; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import com.google.common.collect.ImmutableList; +import com.google.gerrit.server.events.Event; +import com.google.gerrit.server.permissions.PermissionBackendException; +import com.googlesource.gerrit.plugins.multisite.forwarder.CacheNotFoundException; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.AccountIndexEvent; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.ChangeIndexEvent; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.IndexEvent; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent; +import com.googlesource.gerrit.plugins.multisite.forwarder.router.ForwardedEventRouter; +import com.googlesource.gerrit.plugins.multisite.forwarder.router.IndexEventRouter; +import java.io.IOException; +import java.util.List; +import org.junit.Test; + +public class BatchIndexEventSubscriberTest extends AbstractSubscriberTestBase { + private static final boolean DELETED = false; + private static final int CHANGE_ID = 1; + + @SuppressWarnings("unchecked") + @Test + public void shouldConsumeNonProjectAndNonChangeIndexingEventsTypes() + throws IOException, PermissionBackendException, CacheNotFoundException { + IndexEvent event = new AccountIndexEvent(1, INSTANCE_ID); + + objectUnderTest.getConsumer().accept(event); + + verify(projectsFilter, never()).matches(PROJECT_NAME); + verify(eventRouter, times(1)).route(event); + verify(droppedEventListeners, never()).onEventDropped(event); + } + + @SuppressWarnings("rawtypes") + @Override + protected ForwardedEventRouter eventRouter() { + return mock(IndexEventRouter.class); + } + + @Override + protected List<Event> events() { + return ImmutableList.of( + new ProjectIndexEvent(PROJECT_NAME, INSTANCE_ID), + new ChangeIndexEvent(PROJECT_NAME, CHANGE_ID, DELETED, INSTANCE_ID)); + } + + @Override + protected AbstractSubcriber objectUnderTest() { + return new BatchIndexEventSubscriber( + (IndexEventRouter) eventRouter, + asDynamicSet(droppedEventListeners), + NODE_INSTANCE_ID, + msgLog, + subscriberMetrics, + cfg, + projectsFilter); + } +}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/CallbackMetricMaker.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/CallbackMetricMaker.java new file mode 100644 index 0000000..a197c91 --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/CallbackMetricMaker.java
@@ -0,0 +1,26 @@ +package com.googlesource.gerrit.plugins.multisite.consumer; + +import com.google.gerrit.metrics.Counter1; +import com.google.gerrit.metrics.Description; +import com.google.gerrit.metrics.DisabledMetricMaker; +import com.google.gerrit.metrics.Field; +import org.junit.Ignore; + +@Ignore +public class CallbackMetricMaker extends DisabledMetricMaker { + private int callbackMetricCounter = 0; + + public int getCallbackMetricCounter() { + return callbackMetricCounter; + } + + @Override + public <F1> Counter1<F1> newCounter(String name, Description desc, Field<F1> field1) { + callbackMetricCounter += 1; + return super.newCounter(name, desc, field1); + } + + public void resetCallbackMetricCounter() { + callbackMetricCounter = 0; + } +}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriberTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriberTest.java new file mode 100644 index 0000000..0e63528 --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/IndexEventSubscriberTest.java
@@ -0,0 +1,133 @@ +// Copyright (C) 2021 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.consumer; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.common.collect.ImmutableList; +import com.google.gerrit.entities.Account; +import com.google.gerrit.entities.BranchNameKey; +import com.google.gerrit.entities.Change; +import com.google.gerrit.entities.Project; +import com.google.gerrit.server.change.ChangeFinder; +import com.google.gerrit.server.events.Event; +import com.google.gerrit.server.notedb.ChangeNotes; +import com.google.gerrit.server.permissions.PermissionBackendException; +import com.google.gerrit.server.util.time.TimeUtil; +import com.googlesource.gerrit.plugins.multisite.forwarder.CacheNotFoundException; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.AccountIndexEvent; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.ChangeIndexEvent; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.IndexEvent; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent; +import com.googlesource.gerrit.plugins.multisite.forwarder.router.ForwardedEventRouter; +import com.googlesource.gerrit.plugins.multisite.forwarder.router.IndexEventRouter; +import java.io.IOException; +import java.util.List; +import java.util.Optional; +import org.junit.Test; +import org.mockito.Mock; + +public class IndexEventSubscriberTest extends AbstractSubscriberTestBase { + private static final boolean DELETED = false; + private static final int CHANGE_ID = 1; + private static final String EMPTY_PROJECT_NAME = ""; + + @Mock protected ChangeFinder changeFinderMock; + + @SuppressWarnings("unchecked") + @Test + public void shouldConsumeNonProjectAndNonChangeIndexingEventsTypes() + throws IOException, PermissionBackendException, CacheNotFoundException { + IndexEvent event = new AccountIndexEvent(1, INSTANCE_ID); + + objectUnderTest.getConsumer().accept(event); + + verify(projectsFilter, never()).matches(PROJECT_NAME); + verify(eventRouter, times(1)).route(event); + verify(droppedEventListeners, never()).onEventDropped(event); + } + + @SuppressWarnings("unchecked") + @Test + public void shouldConsumeDeleteChangeIndexEventWithEmptyProjectNameWhenFound() + throws IOException, PermissionBackendException, CacheNotFoundException { + ChangeIndexEvent event = new ChangeIndexEvent(EMPTY_PROJECT_NAME, CHANGE_ID, true, INSTANCE_ID); + + ChangeNotes changeNotesMock = mock(ChangeNotes.class); + when(changeNotesMock.getChange()).thenReturn(newChange()); + when(changeFinderMock.findOne(any(Change.Id.class))).thenReturn(Optional.of(changeNotesMock)); + when(projectsFilter.matches(PROJECT_NAME)).thenReturn(true); + + objectUnderTest.getConsumer().accept(event); + + verify(projectsFilter, times(1)).matches(PROJECT_NAME); + verify(eventRouter, times(1)).route(event); + } + + @SuppressWarnings("unchecked") + @Test + public void shouldNOTConsumeDeleteChangeIndexEventWithEmptyProjectNameWhenNotFound() + throws IOException, PermissionBackendException, CacheNotFoundException { + ChangeIndexEvent event = new ChangeIndexEvent("", CHANGE_ID, true, INSTANCE_ID); + + when(changeFinderMock.findOne(any(Change.Id.class))).thenReturn(Optional.empty()); + when(projectsFilter.matches(EMPTY_PROJECT_NAME)).thenReturn(false); + + objectUnderTest.getConsumer().accept(event); + + verify(projectsFilter, times(1)).matches(EMPTY_PROJECT_NAME); + verify(eventRouter, never()).route(event); + } + + @SuppressWarnings("rawtypes") + @Override + protected ForwardedEventRouter eventRouter() { + return mock(IndexEventRouter.class); + } + + @Override + protected List<Event> events() { + return ImmutableList.of( + new ProjectIndexEvent(PROJECT_NAME, INSTANCE_ID), + new ChangeIndexEvent(PROJECT_NAME, CHANGE_ID, DELETED, INSTANCE_ID)); + } + + @Override + protected AbstractSubcriber objectUnderTest() { + return new IndexEventSubscriber( + (IndexEventRouter) eventRouter, + asDynamicSet(droppedEventListeners), + NODE_INSTANCE_ID, + msgLog, + subscriberMetrics, + cfg, + projectsFilter, + changeFinderMock); + } + + private Change newChange() { + return new Change( + Change.key(Integer.toString(CHANGE_ID)), + Change.id(CHANGE_ID), + Account.id(9999), + BranchNameKey.create(Project.nameKey(PROJECT_NAME), "refs/heads/master"), + TimeUtil.nowTs()); + } +}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriberTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriberTest.java new file mode 100644 index 0000000..201ee81 --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ProjectUpdateEventSubscriberTest.java
@@ -0,0 +1,50 @@ +// Copyright (C) 2021 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.consumer; + +import static org.mockito.Mockito.mock; + +import com.google.common.collect.ImmutableList; +import com.google.gerrit.server.events.Event; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectListUpdateEvent; +import com.googlesource.gerrit.plugins.multisite.forwarder.router.ForwardedEventRouter; +import com.googlesource.gerrit.plugins.multisite.forwarder.router.ProjectListUpdateRouter; +import java.util.List; + +public class ProjectUpdateEventSubscriberTest extends AbstractSubscriberTestBase { + + @SuppressWarnings("rawtypes") + @Override + protected ForwardedEventRouter eventRouter() { + return mock(ProjectListUpdateRouter.class); + } + + @Override + protected List<Event> events() { + return ImmutableList.of(new ProjectListUpdateEvent(PROJECT_NAME, false, INSTANCE_ID)); + } + + @Override + protected AbstractSubcriber objectUnderTest() { + return new ProjectUpdateEventSubscriber( + (ProjectListUpdateRouter) eventRouter, + asDynamicSet(droppedEventListeners), + NODE_INSTANCE_ID, + msgLog, + subscriberMetrics, + cfg, + projectsFilter); + } +}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusTest.java index ce82954..7828062 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/ReplicationStatusTest.java
@@ -15,16 +15,27 @@ package com.googlesource.gerrit.plugins.multisite.consumer; import static com.google.common.truth.Truth.assertThat; +import static org.mockito.Mockito.any; +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import com.google.common.cache.Cache; import com.google.common.cache.CacheBuilder; import com.google.common.collect.ImmutableSortedSet; import com.google.gerrit.entities.Project; +import com.google.gerrit.extensions.events.ProjectDeletedListener; +import com.google.gerrit.metrics.CallbackMetric1; +import com.google.gerrit.metrics.DisabledMetricMaker; import com.google.gerrit.server.project.ProjectCache; +import com.googlesource.gerrit.plugins.multisite.Configuration; import com.googlesource.gerrit.plugins.multisite.ProjectVersionLogger; import com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate; import java.util.Optional; +import java.util.concurrent.Executors; +import org.eclipse.jgit.lib.Config; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -37,6 +48,7 @@ @Mock private ProjectVersionLogger verLogger; @Mock private ProjectCache projectCache; @Mock private ProjectVersionRefUpdate projectVersionRefUpdate; + @Mock private CallbackMetric1<String, Long> perProjectReplicationLagMetricCallback; private ReplicationStatus objectUnderTest; private Cache<String, Long> replicationStatusCache; @@ -48,7 +60,13 @@ replicationStatusCache = CacheBuilder.newBuilder().build(); objectUnderTest = new ReplicationStatus( - replicationStatusCache, Optional.of(projectVersionRefUpdate), verLogger, projectCache); + replicationStatusCache, + Optional.of(projectVersionRefUpdate), + verLogger, + projectCache, + Executors.newScheduledThreadPool(1), + new Configuration(new Config(), new Config()), + new DisabledMetricMaker()); } @Test @@ -57,6 +75,14 @@ replicationStatusCache.put("projectB", 3L); objectUnderTest.start(); + assertThat(objectUnderTest.getMaxLagMillis()).isEqualTo(10L); + } + + @Test + public void shouldConvertMillisLagFromPersistedCacheOnStartToSecs() { + replicationStatusCache.put("projectA", 10000L); + + objectUnderTest.start(); assertThat(objectUnderTest.getMaxLag()).isEqualTo(10L); } @@ -67,7 +93,7 @@ objectUnderTest.start(); objectUnderTest.doUpdateLag(Project.nameKey("projectA"), 20L); - assertThat(objectUnderTest.getMaxLag()).isEqualTo(20L); + assertThat(objectUnderTest.getMaxLagMillis()).isEqualTo(20L); } @Test @@ -87,4 +113,127 @@ assertThat(replicationStatusCache.getIfPresent("projectA")).isEqualTo(20L); } + + @Test + public void shouldRemoveProjectFromPersistedCache() { + String projectName = "projectA"; + long lag = 100; + setupReplicationLag(projectName, lag); + when(projectVersionRefUpdate.getProjectLocalVersion(projectName)).thenReturn(Optional.empty()); + + objectUnderTest.onProjectDeleted(projectDeletedEvent(projectName)); + + assertThat(replicationStatusCache.getIfPresent(projectName)).isNull(); + } + + @Test + public void shouldRemoveProjectFromReplicationLags() { + String projectName = "projectA"; + long lag = 100; + setupReplicationLag(projectName, lag); + when(projectVersionRefUpdate.getProjectLocalVersion(projectName)).thenReturn(Optional.empty()); + + assertThat(objectUnderTest.getReplicationLags(1).keySet()).containsExactly(projectName); + + objectUnderTest.onProjectDeleted(projectDeletedEvent(projectName)); + + assertThat(objectUnderTest.getReplicationLags(1).keySet()).isEmpty(); + } + + @Test + public void shouldNotRemoveProjectFromReplicationLagsIfLocalVersionStillExists() { + String projectName = "projectA"; + long lag = 100; + setupReplicationLag(projectName, lag); + when(projectVersionRefUpdate.getProjectLocalVersion(projectName)) + .thenReturn(Optional.of(System.currentTimeMillis())); + + objectUnderTest.onProjectDeleted(projectDeletedEvent(projectName)); + + assertThat(objectUnderTest.getReplicationLags(1).keySet()).containsExactly(projectName); + } + + @Test + public void shouldNotEvictFromPersistentCacheIfLocalVersionStillExists() { + String projectName = "projectA"; + long lag = 100; + setupReplicationLag(projectName, lag); + when(projectVersionRefUpdate.getProjectLocalVersion(projectName)) + .thenReturn(Optional.of(System.currentTimeMillis())); + + objectUnderTest.onProjectDeleted(projectDeletedEvent(projectName)); + + assertThat(replicationStatusCache.getIfPresent(projectName)).isEqualTo(lag); + } + + @Test + public void shouldUpdateReplicationLagForProject() { + String projectName = "projectA"; + long projectLocalVersion = 10L; + long projectRemoteVersion = 20L; + + when(projectVersionRefUpdate.getProjectLocalVersion(eq(projectName))) + .thenReturn(Optional.of(projectLocalVersion)); + when(projectVersionRefUpdate.getProjectRemoteVersion(eq(projectName))) + .thenReturn(Optional.of(projectRemoteVersion)); + + objectUnderTest.updateReplicationLag(Project.nameKey(projectName)); + objectUnderTest.replicationLagMetricPerProject(perProjectReplicationLagMetricCallback).run(); + + assertThat(replicationStatusCache.getIfPresent(projectName)) + .isEqualTo(projectRemoteVersion - projectLocalVersion); + verify(perProjectReplicationLagMetricCallback) + .set(eq(projectName), eq(projectRemoteVersion - projectLocalVersion)); + } + + @Test + public void shouldNotGenerateCallbackMetricIfNoReplicationLag() { + String projectName = "projectA"; + long projectLatestVersion = 10L; + when(projectVersionRefUpdate.getProjectLocalVersion(eq(projectName))) + .thenReturn(Optional.of(projectLatestVersion)); + + objectUnderTest.updateReplicationLag(Project.nameKey(projectName)); + + assertThat(replicationStatusCache.getIfPresent(projectName)).isNull(); + verify(perProjectReplicationLagMetricCallback, never()).set(any(), any()); + } + + @SuppressWarnings("unchecked") + @Test + public void shouldAutoRefreshReplicationLagForProject() { + String projectName = "projectA"; + long projectLocalVersion = 10L; + long projectRemoteVersion = 20L; + + when(projectVersionRefUpdate.getProjectLocalVersion(eq(projectName))) + .thenReturn(Optional.of(projectLocalVersion), Optional.of(projectRemoteVersion)); + when(projectVersionRefUpdate.getProjectRemoteVersion(eq(projectName))) + .thenReturn(Optional.of(projectRemoteVersion)); + + objectUnderTest.updateReplicationLag(Project.nameKey(projectName)); + + assertThat(replicationStatusCache.getIfPresent(projectName)) + .isEqualTo(projectRemoteVersion - projectLocalVersion); + objectUnderTest.refreshProjectsWithLag(); + + assertThat(replicationStatusCache.getIfPresent(projectName)).isEqualTo(0); + } + + private void setupReplicationLag(String projectName, long lag) { + long currentVersion = System.currentTimeMillis(); + long newVersion = currentVersion + lag; + replicationStatusCache.put(projectName, 3L); + when(projectVersionRefUpdate.getProjectRemoteVersion(projectName)) + .thenReturn(Optional.of(newVersion)); + when(projectVersionRefUpdate.getProjectLocalVersion(projectName)) + .thenReturn(Optional.of(currentVersion)); + objectUnderTest.updateReplicationLag(Project.nameKey(projectName)); + } + + private ProjectDeletedListener.Event projectDeletedEvent(String projectName) { + ProjectDeletedListener.Event event = mock(ProjectDeletedListener.Event.class); + when(event.getProjectName()).thenReturn(projectName); + return event; + } }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriberTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriberTest.java new file mode 100644 index 0000000..71fd828 --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/StreamEventSubscriberTest.java
@@ -0,0 +1,122 @@ +// Copyright (C) 2021 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.consumer; + +import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.common.collect.ImmutableList; +import com.google.gerrit.entities.Project; +import com.google.gerrit.entities.Project.NameKey; +import com.google.gerrit.server.events.Event; +import com.google.gerrit.server.events.RefEvent; +import com.google.gerrit.server.events.RefUpdatedEvent; +import com.google.gerrit.server.permissions.PermissionBackendException; +import com.googlesource.gerrit.plugins.multisite.forwarder.CacheNotFoundException; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.AccountIndexEvent; +import com.googlesource.gerrit.plugins.multisite.forwarder.events.IndexEvent; +import com.googlesource.gerrit.plugins.multisite.forwarder.router.ForwardedEventRouter; +import com.googlesource.gerrit.plugins.multisite.forwarder.router.StreamEventRouter; +import com.googlesource.gerrit.plugins.replication.events.RefReplicatedEvent; +import com.googlesource.gerrit.plugins.replication.events.RefReplicationDoneEvent; +import com.googlesource.gerrit.plugins.replication.events.ReplicationScheduledEvent; +import java.io.IOException; +import java.util.List; +import org.junit.Test; +import org.mockito.Mock; + +public class StreamEventSubscriberTest extends AbstractSubscriberTestBase { + private static final NameKey PROJECT_NAME_KEY = NameKey.parse(PROJECT_NAME); + private @Mock RefUpdatedEvent refUpdatedEvent; + private @Mock RefReplicationDoneEvent refReplicationDoneEvent; + private @Mock ReplicationScheduledEvent replicationScheduledEvent; + private @Mock RefReplicatedEvent refReplicatedEvent; + + @SuppressWarnings("rawtypes") + @Override + protected ForwardedEventRouter eventRouter() { + return mock(StreamEventRouter.class); + } + + @Override + protected List<Event> events() { + when(refUpdatedEvent.getProjectNameKey()).thenReturn(PROJECT_NAME_KEY); + refUpdatedEvent.instanceId = INSTANCE_ID; + when(refReplicationDoneEvent.getProjectNameKey()).thenReturn(PROJECT_NAME_KEY); + refReplicationDoneEvent.instanceId = INSTANCE_ID; + when(replicationScheduledEvent.getProjectNameKey()).thenReturn(PROJECT_NAME_KEY); + replicationScheduledEvent.instanceId = INSTANCE_ID; + when(refReplicatedEvent.getProjectNameKey()).thenReturn(PROJECT_NAME_KEY); + refReplicatedEvent.instanceId = INSTANCE_ID; + + return ImmutableList.of( + refUpdatedEvent, refReplicationDoneEvent, replicationScheduledEvent, refReplicatedEvent); + } + + @SuppressWarnings("unchecked") + @Test + public void shouldNotConsumeNonProjectEventTypeEvents() + throws IOException, PermissionBackendException, CacheNotFoundException { + IndexEvent event = new AccountIndexEvent(1, INSTANCE_ID); + + objectUnderTest.getConsumer().accept(event); + + verify(projectsFilter, never()).matches(PROJECT_NAME); + verify(eventRouter, times(1)).route(event); + verify(droppedEventListeners, never()).onEventDropped(event); + } + + @Test + public void shouldUpdateReplicationMetricsWithRemoteRefEvent() { + Event event = + new RefEvent("ref-replicated") { + + @Override + public String getRefName() { + return "foo-ref"; + } + + @Override + public NameKey getProjectNameKey() { + return Project.nameKey(PROJECT_NAME); + } + }; + + event.instanceId = INSTANCE_ID; + when(projectsFilter.matches(eq(PROJECT_NAME))).thenReturn(true); + + objectUnderTest.getConsumer().accept(event); + + verify(subscriberMetrics, times(1)).updateReplicationStatusMetrics(event); + reset(projectsFilter, eventRouter, droppedEventListeners, subscriberMetrics); + } + + @Override + protected AbstractSubcriber objectUnderTest() { + return new StreamEventSubscriber( + (StreamEventRouter) eventRouter, + asDynamicSet(droppedEventListeners), + NODE_INSTANCE_ID, + msgLog, + subscriberMetrics, + cfg, + projectsFilter); + } +}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetricsTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetricsTest.java index 4c45e07..6413b18 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetricsTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/consumer/SubscriberMetricsTest.java
@@ -14,23 +14,28 @@ package com.googlesource.gerrit.plugins.multisite.consumer; +import static com.google.common.truth.Truth.assertThat; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyZeroInteractions; import static org.mockito.Mockito.when; -import com.gerritforge.gerrit.eventbroker.EventMessage; -import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper; import com.google.common.base.Suppliers; import com.google.common.cache.CacheBuilder; import com.google.gerrit.entities.Project; +import com.google.gerrit.metrics.DisabledMetricMaker; import com.google.gerrit.metrics.MetricMaker; import com.google.gerrit.server.data.RefUpdateAttribute; +import com.google.gerrit.server.events.Event; import com.google.gerrit.server.events.RefUpdatedEvent; -import com.google.gerrit.server.extensions.events.GitReferenceUpdated; import com.google.gerrit.server.project.ProjectCache; import com.googlesource.gerrit.plugins.multisite.ProjectVersionLogger; import com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate; +import com.googlesource.gerrit.plugins.replication.events.ProjectDeletionReplicationSucceededEvent; +import java.net.URISyntaxException; import java.util.Optional; -import java.util.UUID; +import java.util.concurrent.Executors; +import org.eclipse.jgit.lib.Config; +import org.eclipse.jgit.transport.URIish; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; @@ -43,37 +48,36 @@ private static final Project.NameKey A_TEST_PROJECT_NAME_KEY = Project.nameKey(A_TEST_PROJECT_NAME); - @Mock private SharedRefDatabaseWrapper sharedRefDb; - @Mock private GitReferenceUpdated gitReferenceUpdated; @Mock private MetricMaker metricMaker; @Mock private ProjectVersionLogger verLogger; @Mock private ProjectCache projectCache; @Mock private ProjectVersionRefUpdate projectVersionRefUpdate; private SubscriberMetrics metrics; - private EventMessage.Header msgHeader; + private ReplicationStatus replicationStatus; @Before public void setup() throws Exception { - msgHeader = new EventMessage.Header(UUID.randomUUID(), UUID.randomUUID()); - metrics = - new SubscriberMetrics( - metricMaker, - new ReplicationStatus( - CacheBuilder.newBuilder().build(), - Optional.of(projectVersionRefUpdate), - verLogger, - projectCache)); + replicationStatus = + new ReplicationStatus( + CacheBuilder.newBuilder().build(), + Optional.of(projectVersionRefUpdate), + verLogger, + projectCache, + Executors.newScheduledThreadPool(1), + new com.googlesource.gerrit.plugins.multisite.Configuration(new Config(), new Config()), + new DisabledMetricMaker()); + metrics = new SubscriberMetrics(metricMaker, replicationStatus); } @Test public void shouldLogProjectVersionWhenReceivingRefUpdatedEventWithoutLag() { - Optional<Long> globalRefDbVersion = Optional.of(System.currentTimeMillis() / 1000); + Optional<Long> globalRefDbVersion = Optional.of(System.currentTimeMillis()); when(projectVersionRefUpdate.getProjectRemoteVersion(A_TEST_PROJECT_NAME)) .thenReturn(globalRefDbVersion); when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME)) .thenReturn(globalRefDbVersion); - EventMessage eventMessage = new EventMessage(msgHeader, newRefUpdateEvent()); + Event eventMessage = newRefUpdateEvent(); metrics.updateReplicationStatusMetrics(eventMessage); @@ -82,20 +86,137 @@ @Test public void shouldLogProjectVersionWhenReceivingRefUpdatedEventWithALag() { - Optional<Long> globalRefDbVersion = Optional.of(System.currentTimeMillis() / 1000); + Optional<Long> globalRefDbVersion = Optional.of(System.currentTimeMillis()); long replicationLag = 60; when(projectVersionRefUpdate.getProjectRemoteVersion(A_TEST_PROJECT_NAME)) .thenReturn(globalRefDbVersion.map(ts -> ts + replicationLag)); when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME)) .thenReturn(globalRefDbVersion); - EventMessage eventMessage = new EventMessage(msgHeader, newRefUpdateEvent()); + Event eventMessage = newRefUpdateEvent(); metrics.updateReplicationStatusMetrics(eventMessage); verify(verLogger).log(A_TEST_PROJECT_NAME_KEY, globalRefDbVersion.get(), replicationLag); } + @Test + public void + shouldLogUponProjectDeletionSuccessWhenLocalVersionDoesNotExistAndSubscriberMetricsExist() + throws Exception { + long nowSecs = System.currentTimeMillis(); + long replicationLagSecs = 60; + Optional<Long> globalRefDbVersion = Optional.of(nowSecs); + when(projectVersionRefUpdate.getProjectRemoteVersion(A_TEST_PROJECT_NAME)) + .thenReturn(globalRefDbVersion.map(ts -> ts + replicationLagSecs)); + when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME)) + .thenReturn(globalRefDbVersion); + + Event refUpdateEventMessage = newRefUpdateEvent(); + metrics.updateReplicationStatusMetrics(refUpdateEventMessage); + + assertThat(replicationStatus.getReplicationStatus(A_TEST_PROJECT_NAME)) + .isEqualTo(replicationLagSecs); + assertThat(replicationStatus.getLocalVersion(A_TEST_PROJECT_NAME)).isEqualTo(nowSecs); + + when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME)) + .thenReturn(Optional.empty()); + + Event projectDeleteEventMessage = projectDeletionSuccess(); + metrics.updateReplicationStatusMetrics(projectDeleteEventMessage); + + verify(verLogger).logDeleted(A_TEST_PROJECT_NAME_KEY); + } + + @Test + public void shouldNotLogUponProjectDeletionSuccessWhenSubscriberMetricsDoNotExist() + throws Exception { + Event eventMessage = projectDeletionSuccess(); + when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME)) + .thenReturn(Optional.empty()); + + assertThat(replicationStatus.getReplicationStatus(A_TEST_PROJECT_NAME)).isNull(); + assertThat(replicationStatus.getLocalVersion(A_TEST_PROJECT_NAME)).isNull(); + + metrics.updateReplicationStatusMetrics(eventMessage); + + verifyZeroInteractions(verLogger); + } + + @Test + public void shouldNotLogUponProjectDeletionSuccessWhenLocalVersionStillExists() throws Exception { + Event eventMessage = projectDeletionSuccess(); + Optional<Long> anyRefVersionValue = Optional.of(System.currentTimeMillis()); + when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME)) + .thenReturn(anyRefVersionValue); + + metrics.updateReplicationStatusMetrics(eventMessage); + + verifyZeroInteractions(verLogger); + } + + @Test + public void shouldRemoveProjectMetricsUponProjectDeletionSuccess() throws Exception { + long nowSecs = System.currentTimeMillis(); + long replicationLagSecs = 60; + Optional<Long> globalRefDbVersion = Optional.of(nowSecs); + when(projectVersionRefUpdate.getProjectRemoteVersion(A_TEST_PROJECT_NAME)) + .thenReturn(globalRefDbVersion.map(ts -> ts + replicationLagSecs)); + when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME)) + .thenReturn(globalRefDbVersion); + + Event eventMessage = newRefUpdateEvent(); + + metrics.updateReplicationStatusMetrics(eventMessage); + + assertThat(replicationStatus.getReplicationStatus(A_TEST_PROJECT_NAME)) + .isEqualTo(replicationLagSecs); + assertThat(replicationStatus.getLocalVersion(A_TEST_PROJECT_NAME)).isEqualTo(nowSecs); + + when(projectVersionRefUpdate.getProjectLocalVersion(A_TEST_PROJECT_NAME)) + .thenReturn(Optional.empty()); + Event projectDeleteEvent = projectDeletionSuccess(); + + metrics.updateReplicationStatusMetrics(projectDeleteEvent); + + assertThat(replicationStatus.getReplicationStatus(A_TEST_PROJECT_NAME)).isNull(); + assertThat(replicationStatus.getLocalVersion(A_TEST_PROJECT_NAME)).isNull(); + } + + @Test + public void shouldDoNothingIfNameIsValid() { + String validProjectName = "aValidProject-Name123"; + + assertThat(metrics.sanitizeProjectName(validProjectName)).isEqualTo(validProjectName); + } + + @Test + public void shouldSanitizeNameWithDot() { + String validProjectName = "nameWithA.InTheMiddle"; + + assertThat(metrics.sanitizeProjectName(validProjectName)).isEqualTo("nameWithA_2eInTheMiddle"); + } + + @Test + public void shouldSanitizeNameWithSlash() { + String validProjectName = "nameWithA/InTheMiddle"; + + assertThat(metrics.sanitizeProjectName(validProjectName)).isEqualTo("nameWithA_2fInTheMiddle"); + } + + @Test + public void shouldDoubleUnderscoresInName() { + String validProjectName = "nameWithA_InTheMiddle"; + + assertThat(metrics.sanitizeProjectName(validProjectName)).isEqualTo("nameWithA__InTheMiddle"); + } + + private ProjectDeletionReplicationSucceededEvent projectDeletionSuccess() + throws URISyntaxException { + return new ProjectDeletionReplicationSucceededEvent( + A_TEST_PROJECT_NAME, new URIish("git://target")); + } + private RefUpdatedEvent newRefUpdateEvent() { RefUpdateAttribute refUpdate = new RefUpdateAttribute(); refUpdate.project = A_TEST_PROJECT_NAME;
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/event/CacheEvictionEventRouterTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/event/CacheEvictionEventRouterTest.java index 76c001f..6762ba2 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/event/CacheEvictionEventRouterTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/event/CacheEvictionEventRouterTest.java
@@ -34,6 +34,7 @@ @RunWith(MockitoJUnitRunner.class) public class CacheEvictionEventRouterTest { + private static final String INSTANCE_ID = "instance-id"; private static Gson gson = new EventGsonProvider().get(); private CacheEvictionEventRouter router; @Mock private ForwardedCacheEvictionHandler cacheEvictionHandler; @@ -45,7 +46,7 @@ @Test public void routerShouldSendEventsToTheAppropriateHandler_CacheEviction() throws Exception { - final CacheEvictionEvent event = new CacheEvictionEvent("cache", "key"); + final CacheEvictionEvent event = new CacheEvictionEvent("cache", "key", INSTANCE_ID); router.route(event); verify(cacheEvictionHandler).evict(CacheEntry.from(event.cacheName, event.key)); @@ -54,7 +55,7 @@ @Test public void routerShouldSendEventsToTheAppropriateHandler_CacheEvictionWithSlash() throws Exception { - final CacheEvictionEvent event = new CacheEvictionEvent("cache", "some/key"); + final CacheEvictionEvent event = new CacheEvictionEvent("cache", "some/key", INSTANCE_ID); router.route(event); verify(cacheEvictionHandler).evict(CacheEntry.from(event.cacheName, event.key)); @@ -63,7 +64,8 @@ @Test public void routerShouldSendEventsToTheAppropriateHandler_ProjectCacheEvictionWithSlash() throws Exception { - final CacheEvictionEvent event = new CacheEvictionEvent(Constants.PROJECTS, "some/project"); + final CacheEvictionEvent event = + new CacheEvictionEvent(Constants.PROJECTS, "some/project", INSTANCE_ID); router.route(event); verify(cacheEvictionHandler)
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/event/IndexEventRouterTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/event/IndexEventRouterTest.java index 1dc07bd..413b6c1 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/event/IndexEventRouterTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/event/IndexEventRouterTest.java
@@ -33,7 +33,7 @@ import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent; import com.googlesource.gerrit.plugins.multisite.forwarder.router.IndexEventRouter; import com.googlesource.gerrit.plugins.multisite.forwarder.router.StreamEventRouter; -import com.googlesource.gerrit.plugins.replication.RefReplicationDoneEvent; +import com.googlesource.gerrit.plugins.replication.events.RefReplicationDoneEvent; import java.util.Optional; import org.eclipse.jgit.lib.ObjectId; import org.junit.Before; @@ -44,7 +44,7 @@ @RunWith(MockitoJUnitRunner.class) public class IndexEventRouterTest { - + private static final String INSTANCE_ID = "instance-id"; private IndexEventRouter router; @Mock private ForwardedIndexAccountHandler indexAccountHandler; @Mock private ForwardedIndexChangeHandler indexChangeHandler; @@ -61,12 +61,13 @@ indexChangeHandler, indexGroupHandler, indexProjectHandler, - allUsersName); + allUsersName, + INSTANCE_ID); } @Test public void routerShouldSendEventsToTheAppropriateHandler_AccountIndex() throws Exception { - final AccountIndexEvent event = new AccountIndexEvent(1); + final AccountIndexEvent event = new AccountIndexEvent(1, INSTANCE_ID); router.route(event); verify(indexAccountHandler) @@ -80,7 +81,7 @@ StreamEventRouter streamEventRouter = new StreamEventRouter(forwardedEventHandler, router); - final AccountIndexEvent event = new AccountIndexEvent(1); + final AccountIndexEvent event = new AccountIndexEvent(1, INSTANCE_ID); router.route(event); verify(indexAccountHandler) @@ -96,7 +97,7 @@ @Test public void routerShouldSendEventsToTheAppropriateHandler_GroupIndex() throws Exception { final String groupId = "12"; - final GroupIndexEvent event = new GroupIndexEvent(groupId, ObjectId.zeroId()); + final GroupIndexEvent event = new GroupIndexEvent(groupId, ObjectId.zeroId(), INSTANCE_ID); router.route(event); verify(indexGroupHandler) @@ -108,7 +109,7 @@ @Test public void routerShouldSendEventsToTheAppropriateHandler_ProjectIndex() throws Exception { final String projectName = "projectName"; - final ProjectIndexEvent event = new ProjectIndexEvent(projectName); + final ProjectIndexEvent event = new ProjectIndexEvent(projectName, INSTANCE_ID); router.route(event); verify(indexProjectHandler) @@ -119,7 +120,7 @@ @Test public void routerShouldSendEventsToTheAppropriateHandler_ChangeIndex() throws Exception { - final ChangeIndexEvent event = new ChangeIndexEvent("projectName", 3, false); + final ChangeIndexEvent event = new ChangeIndexEvent("projectName", 3, false, INSTANCE_ID); router.route(event); verify(indexChangeHandler) @@ -133,7 +134,7 @@ @Test public void routerShouldSendEventsToTheAppropriateHandler_ChangeIndexDelete() throws Exception { - final ChangeIndexEvent event = new ChangeIndexEvent("projectName", 3, true); + final ChangeIndexEvent event = new ChangeIndexEvent("projectName", 3, true, INSTANCE_ID); router.route(event); verify(indexChangeHandler) @@ -147,7 +148,7 @@ @Test public void routerShouldFailForNotRecognisedEvents() throws Exception { - final IndexEvent newEventType = new IndexEvent("new-type") {}; + final IndexEvent newEventType = new IndexEvent("new-type", INSTANCE_ID) {}; assertThrows(UnsupportedOperationException.class, () -> router.route(newEventType)); verifyZeroInteractions(
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/event/ProjectListUpdateRouterTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/event/ProjectListUpdateRouterTest.java index 93daf92..8a21c39 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/event/ProjectListUpdateRouterTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/event/ProjectListUpdateRouterTest.java
@@ -38,7 +38,8 @@ @Test public void routerShouldSendEventsToTheAppropriateHandler_ProjectListUpdate() throws Exception { - final ProjectListUpdateEvent event = new ProjectListUpdateEvent("project", false); + String instanceId = "instance-id"; + final ProjectListUpdateEvent event = new ProjectListUpdateEvent("project", false, instanceId); router.route(event); verify(projectListUpdateHandler).update(event);
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/BrokerForwarderTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/BrokerForwarderTest.java index e3d1ae0..28bf56d 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/BrokerForwarderTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/BrokerForwarderTest.java
@@ -68,10 +68,10 @@ } } - public class TestEvent extends MultiSiteEvent { + public static class TestEvent extends MultiSiteEvent { protected TestEvent() { - super("test"); + super("test", "instance-id"); } } @@ -93,7 +93,7 @@ @Test public void shouldSendEventToBrokerFromGenericSourceThread() { brokerForwarder.send(newForwarderTask(), testTopic, testEvent); - verify(brokerMock).send(eq(testTopicName), eq(testEvent)); + verify(brokerMock).sendSync(eq(testTopicName), eq(testEvent)); } @Test
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerIT.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerIT.java index 826e154..75596ed 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerIT.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerIT.java
@@ -22,6 +22,7 @@ import com.google.common.collect.Sets; import com.google.gerrit.acceptance.LightweightPluginDaemonTest; import com.google.gerrit.acceptance.TestPlugin; +import com.google.gerrit.acceptance.config.GerritConfig; import com.google.gerrit.entities.Project; import com.google.gerrit.extensions.api.projects.ProjectInput; import com.google.gerrit.extensions.registration.DynamicSet; @@ -123,8 +124,10 @@ } @Test + @GerritConfig(name = "gerrit.instanceId", value = "testInstanceId") public void shouldEvictProjectCache() throws Exception { - objectUnderTest.route(new CacheEvictionEvent(ProjectCacheImpl.CACHE_NAME, project.get())); + objectUnderTest.route( + new CacheEvictionEvent(ProjectCacheImpl.CACHE_NAME, project.get(), "instance-id")); evictionsCacheTracker.waitForExpectedEvictions(); assertThat(evictionsCacheTracker.trackedEvictionsFor(ProjectCacheImpl.CACHE_NAME)) @@ -132,6 +135,7 @@ } @Test + @GerritConfig(name = "gerrit.instanceId", value = "instance-id") public void shouldEvictProjectCacheWithSlash() throws Exception { ProjectInput in = new ProjectInput(); in.name = name("my/project"); @@ -141,7 +145,7 @@ restartCacheEvictionsTracking(); objectUnderTest.route( - new CacheEvictionEvent(ProjectCacheImpl.CACHE_NAME, projectNameKey.get())); + new CacheEvictionEvent(ProjectCacheImpl.CACHE_NAME, projectNameKey.get(), "instance-id")); evictionsCacheTracker.waitForExpectedEvictions(); assertThat(evictionsCacheTracker.trackedEvictionsFor(ProjectCacheImpl.CACHE_NAME))
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerTest.java index 90ae8da..dbd358d 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedCacheEvictionHandlerTest.java
@@ -15,6 +15,7 @@ package com.googlesource.gerrit.plugins.multisite.forwarder; import static com.google.common.truth.Truth.assertThat; +import static com.google.gerrit.testing.GerritJUnit.assertThrows; import static org.mockito.Mockito.doReturn; import com.google.common.cache.Cache; @@ -48,10 +49,12 @@ public void shouldThrowAnExceptionWhenCacheNotFound() throws Exception { CacheEntry entry = new CacheEntry("somePlugin", "unexistingCache", null); - exception.expect(CacheNotFoundException.class); - exception.expectMessage( - String.format("cache %s.%s not found", entry.getPluginName(), entry.getCacheName())); - handler.evict(entry); + CacheNotFoundException thrown = + assertThrows(CacheNotFoundException.class, () -> handler.evict(entry)); + assertThat(thrown) + .hasMessageThat() + .isEqualTo( + String.format("cache %s.%s not found", entry.getPluginName(), entry.getCacheName())); } @Test
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexAccountHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexAccountHandlerTest.java index 32c6319..02efb77 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexAccountHandlerTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexAccountHandlerTest.java
@@ -61,9 +61,11 @@ @Test public void deleteIsNotSupported() throws Exception { - exception.expect(UnsupportedOperationException.class); - exception.expectMessage("Delete from account index not supported"); - handler.index(id, Operation.DELETE, Optional.empty()); + UnsupportedOperationException thrown = + assertThrows( + UnsupportedOperationException.class, + () -> handler.index(id, Operation.DELETE, Optional.empty())); + assertThat(thrown).hasMessageThat().isEqualTo("Delete from account index not supported"); } @Test
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandlerTest.java index 96470b6..b827cdf 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandlerTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexChangeHandlerTest.java
@@ -17,6 +17,7 @@ import static com.google.common.truth.Truth.assertThat; import static com.google.gerrit.testing.GerritJUnit.assertThrows; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.never; @@ -49,6 +50,7 @@ import org.mockito.Mock; import org.mockito.junit.MockitoJUnitRunner; import org.mockito.stubbing.Answer; +import org.mockito.stubbing.OngoingStubbing; @RunWith(MockitoJUnitRunner.class) public class ForwardedIndexChangeHandlerTest { @@ -62,6 +64,8 @@ private static final boolean THROW_STORAGE_EXCEPTION = true; private static final boolean CHANGE_UP_TO_DATE = true; private static final boolean CHANGE_OUTDATED = false; + private static final boolean CHANGE_CONSISTENT = true; + private static final boolean CHANGE_INCONSISTENT = false; @Rule public ExpectedException exception = ExpectedException.none(); @Mock private ChangeIndexer indexerMock; @@ -88,6 +92,7 @@ when(changeCheckerFactoryMock.create(any())).thenReturn(changeCheckerAbsentMock); when(configurationMock.index()).thenReturn(index); when(index.numStripedLocks()).thenReturn(10); + when(index.maxTries()).thenReturn(1); handler = new ForwardedIndexChangeHandler( indexerMock, configurationMock, indexExecutorMock, ctxMock, changeCheckerFactoryMock); @@ -95,16 +100,40 @@ @Test public void changeIsIndexedWhenUpToDate() throws Exception { - setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_UP_TO_DATE); + setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_UP_TO_DATE, CHANGE_CONSISTENT); handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty()); verify(indexerMock, times(1)).index(any(Change.class)); } @Test public void changeIsStillIndexedEvenWhenOutdated() throws Exception { - setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_OUTDATED); + setupChangeAccessRelatedMocks(CHANGE_EXISTS, CHANGE_OUTDATED, CHANGE_CONSISTENT); handler.index( - TEST_CHANGE_ID, Operation.INDEX, Optional.of(new ChangeIndexEvent("foo", 1, false))); + TEST_CHANGE_ID, + Operation.INDEX, + Optional.of(new ChangeIndexEvent("foo", 1, false, "instance-id"))); + verify(indexerMock, times(1)).index(any(Change.class)); + } + + @Test + public void changeIsIndexeAtFirstRetryWhenInitiallyInconsistent() throws Exception { + setupChangeAccessRelatedMocks( + CHANGE_EXISTS, + DO_NOT_THROW_STORAGE_EXCEPTION, + CHANGE_UP_TO_DATE, + CHANGE_INCONSISTENT, + CHANGE_CONSISTENT); + handler.index( + TEST_CHANGE_ID, + Operation.INDEX, + Optional.of(new ChangeIndexEvent("foo", 1, false, "instance-id"))); + verify(indexerMock, never()).index(any(Change.class)); + verify(indexExecutorMock, times(1)).schedule(any(Runnable.class), anyLong(), any()); + + handler.index( + TEST_CHANGE_ID, + Operation.INDEX, + Optional.of(new ChangeIndexEvent("foo", 1, false, "instance-id"))); verify(indexerMock, times(1)).index(any(Change.class)); } @@ -124,9 +153,11 @@ @Test public void indexerThrowsStorageExceptionTryingToIndexChange() throws Exception { - setupChangeAccessRelatedMocks(CHANGE_EXISTS, THROW_STORAGE_EXCEPTION, CHANGE_UP_TO_DATE); - exception.expect(StorageException.class); - handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty()); + setupChangeAccessRelatedMocks( + CHANGE_EXISTS, THROW_STORAGE_EXCEPTION, CHANGE_UP_TO_DATE, CHANGE_CONSISTENT); + assertThrows( + StorageException.class, + () -> handler.index(TEST_CHANGE_ID, Operation.INDEX, Optional.empty())); } @Test @@ -175,11 +206,21 @@ private void setupChangeAccessRelatedMocks(boolean changeExist, boolean changeUpToDate) throws Exception { - setupChangeAccessRelatedMocks(changeExist, DO_NOT_THROW_STORAGE_EXCEPTION, changeUpToDate); + setupChangeAccessRelatedMocks( + changeExist, DO_NOT_THROW_STORAGE_EXCEPTION, changeUpToDate, CHANGE_CONSISTENT); } private void setupChangeAccessRelatedMocks( - boolean changeExists, boolean storageException, boolean changeIsUpToDate) + boolean changeExist, boolean changeUpToDate, boolean changeConsistent) throws Exception { + setupChangeAccessRelatedMocks( + changeExist, DO_NOT_THROW_STORAGE_EXCEPTION, changeUpToDate, changeConsistent); + } + + private void setupChangeAccessRelatedMocks( + boolean changeExists, + boolean storageException, + boolean changeIsUpToDate, + boolean... changeConsistentReturnValues) throws StorageException { if (changeExists) { when(changeCheckerFactoryMock.create(TEST_CHANGE_ID)).thenReturn(changeCheckerPresentMock); @@ -190,5 +231,11 @@ } when(changeCheckerPresentMock.isUpToDate(any())).thenReturn(changeIsUpToDate); + + OngoingStubbing<Boolean> changeConsistentCall = + when(changeCheckerPresentMock.isChangeConsistent()); + for (boolean changeConsistent : changeConsistentReturnValues) { + changeConsistentCall = changeConsistentCall.thenReturn(changeConsistent); + } } }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexGroupHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexGroupHandlerTest.java index 982ac52..b49dbfa 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexGroupHandlerTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexGroupHandlerTest.java
@@ -74,9 +74,11 @@ @Test public void deleteIsNotSupported() throws Exception { - exception.expect(UnsupportedOperationException.class); - exception.expectMessage("Delete from group index not supported"); - handler.index(uuid, Operation.DELETE, Optional.empty()); + UnsupportedOperationException thrown = + assertThrows( + UnsupportedOperationException.class, + () -> handler.index(uuid, Operation.DELETE, Optional.empty())); + assertThat(thrown).hasMessageThat().isEqualTo("Delete from group index not supported"); } @Test @@ -141,6 +143,6 @@ } private Optional<GroupIndexEvent> groupIndexEvent(String uuid) { - return Optional.of(new GroupIndexEvent(uuid, null)); + return Optional.of(new GroupIndexEvent(uuid, null, "instance-id")); } }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexProjectHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexProjectHandlerTest.java index 3ce5e14..72b9427 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexProjectHandlerTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedIndexProjectHandlerTest.java
@@ -73,9 +73,11 @@ @Test public void deleteIsNotSupported() throws Exception { - exception.expect(UnsupportedOperationException.class); - exception.expectMessage("Delete from project index not supported"); - handler.index(nameKey, Operation.DELETE, Optional.empty()); + UnsupportedOperationException thrown = + assertThrows( + UnsupportedOperationException.class, + () -> handler.index(nameKey, Operation.DELETE, Optional.empty())); + assertThat(thrown).hasMessageThat().isEqualTo("Delete from project index not supported"); } @Test
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedProjectListUpdateHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedProjectListUpdateHandlerTest.java index 9893ce7..2412c79 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedProjectListUpdateHandlerTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/forwarder/ForwardedProjectListUpdateHandlerTest.java
@@ -34,6 +34,7 @@ @RunWith(MockitoJUnitRunner.class) public class ForwardedProjectListUpdateHandlerTest { + private static final String INSTANCE_ID = "instance-id"; private static final String PROJECT_NAME = "someProject"; private static final String SOME_MESSAGE = "someMessage"; private static final Project.NameKey PROJECT_KEY = Project.nameKey(PROJECT_NAME); @@ -48,13 +49,13 @@ @Test public void testSuccessfulAdd() throws Exception { - handler.update(new ProjectListUpdateEvent(PROJECT_NAME, false)); + handler.update(new ProjectListUpdateEvent(PROJECT_NAME, false, INSTANCE_ID)); verify(projectCacheMock).onCreateProject(PROJECT_KEY); } @Test public void testSuccessfulRemove() throws Exception { - handler.update(new ProjectListUpdateEvent(PROJECT_NAME, true)); + handler.update(new ProjectListUpdateEvent(PROJECT_NAME, true, INSTANCE_ID)); verify(projectCacheMock).remove(PROJECT_KEY); } @@ -72,7 +73,7 @@ .onCreateProject(PROJECT_KEY); assertThat(Context.isForwardedEvent()).isFalse(); - handler.update(new ProjectListUpdateEvent(PROJECT_NAME, false)); + handler.update(new ProjectListUpdateEvent(PROJECT_NAME, false, INSTANCE_ID)); assertThat(Context.isForwardedEvent()).isFalse(); verify(projectCacheMock).onCreateProject(PROJECT_KEY); @@ -92,7 +93,7 @@ .remove(PROJECT_KEY); assertThat(Context.isForwardedEvent()).isFalse(); - handler.update(new ProjectListUpdateEvent(PROJECT_NAME, true)); + handler.update(new ProjectListUpdateEvent(PROJECT_NAME, true, INSTANCE_ID)); assertThat(Context.isForwardedEvent()).isFalse(); verify(projectCacheMock).remove(PROJECT_KEY); @@ -113,7 +114,7 @@ RuntimeException thrown = assertThrows( RuntimeException.class, - () -> handler.update(new ProjectListUpdateEvent(PROJECT_NAME, false))); + () -> handler.update(new ProjectListUpdateEvent(PROJECT_NAME, false, INSTANCE_ID))); assertThat(thrown).hasMessageThat().isEqualTo(SOME_MESSAGE); assertThat(Context.isForwardedEvent()).isFalse(); @@ -135,7 +136,7 @@ RuntimeException thrown = assertThrows( RuntimeException.class, - () -> handler.update(new ProjectListUpdateEvent(PROJECT_NAME, true))); + () -> handler.update(new ProjectListUpdateEvent(PROJECT_NAME, true, INSTANCE_ID))); assertThat(thrown).hasMessageThat().isEqualTo(SOME_MESSAGE); assertThat(Context.isForwardedEvent()).isFalse();
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/http/ReplicationStatusServletIT.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/http/ReplicationStatusServletIT.java index 0739eab..f940fc4 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/http/ReplicationStatusServletIT.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/http/ReplicationStatusServletIT.java
@@ -24,9 +24,12 @@ import com.google.gerrit.acceptance.LightweightPluginDaemonTest; import com.google.gerrit.acceptance.RestResponse; import com.google.gerrit.acceptance.TestPlugin; +import com.google.gerrit.acceptance.config.GerritConfig; import com.google.gerrit.entities.Project; import com.google.gerrit.httpd.restapi.RestApiServlet; +import com.google.gerrit.server.git.WorkQueue; import com.google.inject.AbstractModule; +import com.google.inject.Inject; import com.google.inject.Scopes; import com.google.inject.multibindings.OptionalBinder; import com.googlesource.gerrit.plugins.multisite.Log4jProjectVersionLogger; @@ -56,13 +59,15 @@ private ReplicationStatus replicationStatus; public static class TestModule extends AbstractModule { + @Inject WorkQueue workQueue; + @Override protected void configure() { install(new ForwarderModule()); install(new CacheModule()); install(new RouterModule()); install(new IndexModule()); - install(new ReplicationStatusModule()); + install(new ReplicationStatusModule(workQueue)); SharedRefDbConfiguration sharedRefDbConfig = new SharedRefDbConfiguration(new Config(), "multi-site"); bind(SharedRefDbConfiguration.class).toInstance(sharedRefDbConfig); @@ -81,6 +86,7 @@ } @Test + @GerritConfig(name = "gerrit.instanceId", value = "testInstanceId") public void shouldSucceedForAdminUsers() throws Exception { RestResponse result = adminRestSession.get(LAG_ENDPOINT); @@ -89,6 +95,7 @@ } @Test + @GerritConfig(name = "gerrit.instanceId", value = "testInstanceId") public void shouldFailWhenUserHasNoAdminServerCapability() throws Exception { RestResponse result = userRestSession.get(LAG_ENDPOINT); result.assertForbidden(); @@ -96,6 +103,7 @@ } @Test + @GerritConfig(name = "gerrit.instanceId", value = "testInstanceId") public void shouldReturnCurrentProjectLag() throws Exception { replicationStatus.doUpdateLag(Project.nameKey("foo"), 123L); @@ -106,6 +114,7 @@ } @Test + @GerritConfig(name = "gerrit.instanceId", value = "testInstanceId") public void shouldReturnProjectsOrderedDescendinglyByLag() throws Exception { replicationStatus.doUpdateLag(Project.nameKey("bar"), 123L); replicationStatus.doUpdateLag(Project.nameKey("foo"), 3L); @@ -118,6 +127,7 @@ } @Test + @GerritConfig(name = "gerrit.instanceId", value = "testInstanceId") public void shouldHonourTheLimitParameter() throws Exception { replicationStatus.doUpdateLag(Project.nameKey("bar"), 1L); replicationStatus.doUpdateLag(Project.nameKey("foo"), 2L);
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupCheckerImplTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupCheckerImplTest.java index 6403cce..09f8f59 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupCheckerImplTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupCheckerImplTest.java
@@ -101,7 +101,7 @@ } private Optional<GroupIndexEvent> groupIndexEvent(String uuid, @Nullable ObjectId sha1) { - return Optional.of(new GroupIndexEvent(uuid, sha1)); + return Optional.of(new GroupIndexEvent(uuid, sha1, "instance-id")); } private void setCommitExistsInRepo(boolean commitExists) {
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupEventIndexTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupEventIndexTest.java index 93cec05..099f1dd 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupEventIndexTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/index/GroupEventIndexTest.java
@@ -16,7 +16,7 @@ import static com.google.common.truth.Truth.assertThat; -import com.gerritforge.gerrit.eventbroker.EventGsonProvider; +import com.google.gerrit.server.events.EventGsonProvider; import com.google.gson.Gson; import com.googlesource.gerrit.plugins.multisite.forwarder.events.GroupIndexEvent; import java.util.UUID; @@ -24,13 +24,14 @@ import org.junit.Test; public class GroupEventIndexTest { + private static final String INSTANCE_ID = "instance-id"; private static final Gson gson = new EventGsonProvider().get(); @Test public void groupEventIndexRoundTripWithSha1() { String aGroupUUID = UUID.randomUUID().toString(); ObjectId anObjectId = ObjectId.fromString("deadbeefdeadbeefdeadbeefdeadbeefdeadbeef"); - GroupIndexEvent original = new GroupIndexEvent(aGroupUUID, anObjectId); + GroupIndexEvent original = new GroupIndexEvent(aGroupUUID, anObjectId, INSTANCE_ID); assertThat(gson.fromJson(gson.toJson(original), GroupIndexEvent.class)).isEqualTo(original); } @@ -38,7 +39,7 @@ @Test public void groupEventIndexRoundTripWithoutSha1() { String aGroupUUID = UUID.randomUUID().toString(); - GroupIndexEvent original = new GroupIndexEvent(aGroupUUID, null); + GroupIndexEvent original = new GroupIndexEvent(aGroupUUID, null, INSTANCE_ID); assertThat(gson.fromJson(gson.toJson(original), GroupIndexEvent.class)).isEqualTo(original); }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandlerTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandlerTest.java index 660a302..8cf6ea9 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandlerTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/index/IndexEventHandlerTest.java
@@ -14,16 +14,21 @@ package com.googlesource.gerrit.plugins.multisite.index; +import static com.googlesource.gerrit.plugins.replication.pull.api.PullReplicationEndpoints.APPLY_OBJECT_API_ENDPOINT; import static org.mockito.Mockito.any; +import static org.mockito.Mockito.anyString; import static org.mockito.Mockito.eq; +import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.verifyZeroInteractions; import static org.mockito.Mockito.when; import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter; import com.google.common.util.concurrent.MoreExecutors; import com.google.gerrit.extensions.registration.DynamicSet; +import com.googlesource.gerrit.plugins.multisite.forwarder.Context; import com.googlesource.gerrit.plugins.multisite.forwarder.IndexEventForwarder; import com.googlesource.gerrit.plugins.multisite.forwarder.events.ProjectIndexEvent; import com.googlesource.gerrit.plugins.multisite.index.IndexEventHandler.IndexProjectTask; @@ -36,6 +41,8 @@ @RunWith(MockitoJUnitRunner.class) public class IndexEventHandlerTest { + private static final String INSTANCE_ID = "instance-id"; + private IndexEventHandler eventHandler; @Mock private ProjectsFilter projectsFilter; @@ -50,7 +57,8 @@ asDynamicSet(forwarder), changeChecker, projectsFilter, - new TestGroupChecker(true)); + new TestGroupChecker(true), + INSTANCE_ID); } private DynamicSet<IndexEventForwarder> asDynamicSet(IndexEventForwarder forwarder) { @@ -65,7 +73,7 @@ eventHandler.onProjectIndexed("test_project"); verify(forwarder, never()) - .index(any(IndexProjectTask.class), eq(new ProjectIndexEvent("test_project"))); + .index(any(IndexProjectTask.class), eq(new ProjectIndexEvent("test_project", INSTANCE_ID))); } @Test @@ -76,4 +84,25 @@ eventHandler.onChangeIndexed("test_project", changeId); verifyZeroInteractions(changeChecker); } + + @Test + public void shouldNotForwardIndexChangeIfCurrentThreadIsPullReplicationApplyObject() + throws Exception { + String currentThreadName = Thread.currentThread().getName(); + try { + Thread.currentThread().setName("pull-replication~" + APPLY_OBJECT_API_ENDPOINT); + int changeId = 1; + Context.setForwardedEvent(false); + lenient().when(projectsFilter.matches(any(String.class))).thenReturn(true); + lenient() + .when(changeChecker.create(anyString())) + .thenThrow( + new IllegalStateException("Change indexing event should have not been triggered")); + + eventHandler.onChangeIndexed("test_project", changeId); + verifyNoInteractions(changeChecker); + } finally { + Thread.currentThread().setName(currentThreadName); + } + } }
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/DisabledSharedRefLogger.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/DisabledSharedRefLogger.java new file mode 100644 index 0000000..fc2259f --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/DisabledSharedRefLogger.java
@@ -0,0 +1,42 @@ +// Copyright (C) 2022 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.validation; + +import com.gerritforge.gerrit.globalrefdb.validation.SharedRefLogger; +import org.eclipse.jgit.lib.ObjectId; +import org.eclipse.jgit.lib.Ref; +import org.junit.Ignore; + +@Ignore +public class DisabledSharedRefLogger implements SharedRefLogger { + + @Override + public void logRefUpdate(String project, Ref currRef, ObjectId newRefValue) {} + + @Override + public void logProjectDelete(String project) {} + + @Override + public void logLockAcquisition(String project, String refName) {} + + @Override + public void logLockRelease(String project, String refName) {} + + @Override + public <T> void logRefUpdate(String project, String refName, T currRef, T newRefValue) {} + + @Override + public <T> void logRefUpdate(String project, String refName, T newRefValue) {} +}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/FakeSharedRefDatabaseWrapper.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/FakeSharedRefDatabaseWrapper.java new file mode 100644 index 0000000..bf98a4b --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/FakeSharedRefDatabaseWrapper.java
@@ -0,0 +1,84 @@ +// Copyright (C) 2023 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.validation; + +import com.gerritforge.gerrit.globalrefdb.GlobalRefDatabase; +import com.gerritforge.gerrit.globalrefdb.GlobalRefDbLockException; +import com.gerritforge.gerrit.globalrefdb.GlobalRefDbSystemError; +import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDBMetrics; +import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper; +import com.google.gerrit.entities.Project; +import com.google.gerrit.extensions.registration.DynamicItem; +import com.google.gerrit.metrics.DisabledMetricMaker; +import java.util.Arrays; +import java.util.Optional; +import org.eclipse.jgit.lib.ObjectId; +import org.eclipse.jgit.lib.Ref; +import org.junit.Ignore; + +@Ignore +public class FakeSharedRefDatabaseWrapper extends SharedRefDatabaseWrapper { + + public FakeSharedRefDatabaseWrapper(String... rejectedRefs) { + + super( + DynamicItem.itemOf( + GlobalRefDatabase.class, + new GlobalRefDatabase() { + + @Override + public boolean isUpToDate(Project.NameKey project, Ref ref) + throws GlobalRefDbLockException { + return !Arrays.stream(rejectedRefs).anyMatch(r -> r.equals(ref.getName())); + } + + @Override + public boolean exists(Project.NameKey project, String refName) { + return true; + } + + @Override + public boolean compareAndPut( + Project.NameKey project, Ref currRef, ObjectId newRefValue) + throws GlobalRefDbSystemError { + return false; + } + + @Override + public <T> boolean compareAndPut( + Project.NameKey project, String refName, T currValue, T newValue) + throws GlobalRefDbSystemError { + return false; + } + + @Override + public AutoCloseable lockRef(Project.NameKey project, String refName) + throws GlobalRefDbLockException { + return null; + } + + @Override + public void remove(Project.NameKey project) throws GlobalRefDbSystemError {} + + @Override + public <T> Optional<T> get(Project.NameKey project, String refName, Class<T> clazz) + throws GlobalRefDbSystemError { + return Optional.empty(); + } + }), + new DisabledSharedRefLogger(), + new SharedRefDBMetrics(new DisabledMetricMaker())); + } +}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilterTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilterTest.java new file mode 100644 index 0000000..e09f6f6 --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationFetchFilterTest.java
@@ -0,0 +1,258 @@ +// Copyright (C) 2023 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.validation; + +import static com.google.common.truth.Truth.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper; +import com.google.gerrit.entities.Project; +import com.google.gerrit.testing.InMemoryRepositoryManager; +import com.google.gerrit.testing.InMemoryTestEnvironment; +import com.google.inject.Inject; +import com.googlesource.gerrit.plugins.multisite.Configuration; +import com.googlesource.gerrit.plugins.multisite.validation.dfsrefdb.RefFixture; +import java.util.Optional; +import java.util.Set; +import org.eclipse.jgit.internal.storage.dfs.InMemoryRepository; +import org.eclipse.jgit.junit.LocalDiskRepositoryTestCase; +import org.eclipse.jgit.junit.TestRepository; +import org.eclipse.jgit.lib.ObjectId; +import org.eclipse.jgit.revwalk.RevCommit; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.class) +public class MultisiteReplicationFetchFilterTest extends LocalDiskRepositoryTestCase + implements RefFixture { + + @Rule public InMemoryTestEnvironment testEnvironment = new InMemoryTestEnvironment(); + + @Mock SharedRefDatabaseWrapper sharedRefDatabaseMock; + @Mock Configuration config; + @Mock Configuration.ReplicationFilter replicationFilterConfig; + + @Inject private InMemoryRepositoryManager gitRepositoryManager; + + String project = A_TEST_PROJECT_NAME; + Project.NameKey projectName = A_TEST_PROJECT_NAME_KEY; + + private TestRepository<InMemoryRepository> repo; + + @Before + public void setupTestRepo() throws Exception { + InMemoryRepository inMemoryRepo = + gitRepositoryManager.createRepository(A_TEST_PROJECT_NAME_KEY); + repo = new TestRepository<>(inMemoryRepo); + doReturn(replicationFilterConfig).when(config).replicationFilter(); + } + + @Test + public void shouldReturnEmptyRefsWhenAllUpToDate() throws Exception { + String fooRefName = "refs/heads/foo"; + ObjectId fooObjectId = newRef(fooRefName).getId(); + String barRefName = "refs/heads/bar"; + ObjectId barObjectId = newRef(barRefName).getId(); + Set<String> refs = Set.of(fooRefName, barRefName); + + doReturn(Optional.of(fooObjectId.getName())) + .when(sharedRefDatabaseMock) + .get(eq(projectName), eq(fooRefName), eq(String.class)); + doReturn(Optional.of(barObjectId.getName())) + .when(sharedRefDatabaseMock) + .get(eq(projectName), eq(barRefName), eq(String.class)); + + MultisiteReplicationFetchFilter fetchFilter = + new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config); + Set<String> filteredRefs = fetchFilter.filter(project, refs); + + assertThat(filteredRefs).isEmpty(); + } + + @Test + public void shouldFilterOutOneUpToDateRef() throws Exception { + String refUpToDate = "refs/heads/uptodate"; + String outdatedRef = "refs/heads/outdated"; + ObjectId upToDateObjectId = newRef(refUpToDate).getId(); + newRef(outdatedRef).getId(); + Set<String> refsToFetch = Set.of(refUpToDate, outdatedRef); + + doReturn(Optional.of(upToDateObjectId.getName())) + .when(sharedRefDatabaseMock) + .get(eq(projectName), eq(refUpToDate), eq(String.class)); + doReturn(Optional.of(AN_OUTDATED_OBJECT_ID.getName())) + .when(sharedRefDatabaseMock) + .get(eq(projectName), eq(outdatedRef), eq(String.class)); + + MultisiteReplicationFetchFilter fetchFilter = + new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config); + Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch); + + assertThat(filteredRefsToFetch).containsExactly(outdatedRef); + } + + @Test + public void shouldLoadLocalVersionAndFilterOut() throws Exception { + String temporaryOutdated = "refs/heads/temporaryOutdated"; + RevCommit localRef = newRef(temporaryOutdated); + + Set<String> refsToFetch = Set.of(temporaryOutdated); + doReturn(Optional.empty()) + .doReturn(Optional.of(localRef.getId().getName())) + .when(sharedRefDatabaseMock) + .get(eq(projectName), eq(temporaryOutdated), eq(String.class)); + + MultisiteReplicationFetchFilter fetchFilter = + new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config); + Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch); + + assertThat(filteredRefsToFetch).isEmpty(); + + verify(sharedRefDatabaseMock, times(2)).get(any(), any(), any()); + } + + @Test + public void shouldLoadLocalVersionAndNotFilter() throws Exception { + String temporaryOutdated = "refs/heads/temporaryOutdated"; + newRef(temporaryOutdated); + + Set<String> refsToFetch = Set.of(temporaryOutdated); + doReturn(Optional.of(AN_OUTDATED_OBJECT_ID.getName())) + .doReturn(Optional.of(AN_OUTDATED_OBJECT_ID.getName())) + .when(sharedRefDatabaseMock) + .get(eq(projectName), eq(temporaryOutdated), eq(String.class)); + + MultisiteReplicationFetchFilter fetchFilter = + new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config); + Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch); + + assertThat(filteredRefsToFetch).hasSize(1); + verify(sharedRefDatabaseMock, times(3)) + .get(eq(projectName), eq(temporaryOutdated), eq(String.class)); + } + + @Test + public void shouldNotFilterOutWhenMissingInTheSharedRefDb() throws Exception { + String temporaryOutdated = "refs/heads/temporaryOutdated"; + newRef(temporaryOutdated); + + doReturn(Optional.empty()) + .when(sharedRefDatabaseMock) + .get(eq(projectName), eq(temporaryOutdated), eq(String.class)); + + Set<String> refsToFetch = Set.of(temporaryOutdated); + + MultisiteReplicationFetchFilter fetchFilter = + new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config); + Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch); + + assertThat(filteredRefsToFetch).hasSize(1); + } + + @Test + public void shouldNotFilterOutWhenRefsMultisiteVersionIsPresentInSharedRefDb() throws Exception { + String refsMultisiteVersionRef = ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF; + RevCommit multiSiteVersionRef = newRef(refsMultisiteVersionRef); + + doReturn(Optional.of(multiSiteVersionRef.getId().getName())) + .when(sharedRefDatabaseMock) + .get(eq(projectName), eq(refsMultisiteVersionRef), eq(String.class)); + + Set<String> refsToFetch = Set.of(refsMultisiteVersionRef); + + MultisiteReplicationFetchFilter fetchFilter = + new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config); + Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch); + + assertThat(filteredRefsToFetch).hasSize(1); + } + + @Test + public void shouldFilterOutWhenRefIsDeletedInTheSharedRefDb() throws Exception { + String temporaryOutdated = "refs/heads/temporaryOutdated"; + newRef(temporaryOutdated); + + Set<String> refsToFetch = Set.of(temporaryOutdated); + doReturn(Optional.of(ObjectId.zeroId().getName())) + .when(sharedRefDatabaseMock) + .get(eq(projectName), eq(temporaryOutdated), eq(String.class)); + + MultisiteReplicationFetchFilter fetchFilter = + new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config); + Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch); + + assertThat(filteredRefsToFetch).hasSize(0); + verify(sharedRefDatabaseMock).get(eq(projectName), any(), any()); + } + + @Test + public void shouldNotFilterOutWhenRefIsMissingOnlyInTheLocalRepository() throws Exception { + String refObjectId = "0000000000000000000000000000000000000001"; + String nonExistingLocalRef = "refs/heads/temporaryOutdated"; + + Set<String> refsToFetch = Set.of(nonExistingLocalRef); + doReturn(Optional.of(refObjectId)) + .when(sharedRefDatabaseMock) + .get(eq(projectName), any(), any()); + + MultisiteReplicationFetchFilter fetchFilter = + new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config); + Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch); + + assertThat(filteredRefsToFetch).hasSize(1); + } + + @Test + public void shouldNotFilterOutRefThatDoesntExistLocallyOrInSharedRefDb() throws Exception { + String nonExisting = "refs/heads/non-existing-ref"; + + Set<String> refsToFetch = Set.of(nonExisting); + + MultisiteReplicationFetchFilter fetchFilter = + new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config); + Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch); + + assertThat(filteredRefsToFetch).hasSize(1); + } + + @Test + public void shouldFilterOutRefMissingInTheLocalRepositoryAndDeletedInSharedRefDb() + throws Exception { + String nonExistingLocalRef = "refs/heads/temporaryOutdated"; + + Set<String> refsToFetch = Set.of(nonExistingLocalRef); + doReturn(Optional.of(ObjectId.zeroId().getName())) + .when(sharedRefDatabaseMock) + .get(eq(projectName), any(), any()); + + MultisiteReplicationFetchFilter fetchFilter = + new MultisiteReplicationFetchFilter(sharedRefDatabaseMock, gitRepositoryManager, config); + Set<String> filteredRefsToFetch = fetchFilter.filter(project, refsToFetch); + + assertThat(filteredRefsToFetch).hasSize(0); + } + + private RevCommit newRef(String refName) throws Exception { + return repo.branch(refName).commit().create(); + } +}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilterTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilterTest.java new file mode 100644 index 0000000..b4a5e5d --- /dev/null +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/MultisiteReplicationPushFilterTest.java
@@ -0,0 +1,159 @@ +// Copyright (C) 2022 The Android Open Source Project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package com.googlesource.gerrit.plugins.multisite.validation; + +import static com.google.common.truth.Truth.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; + +import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper; +import com.google.gerrit.entities.Project; +import com.google.gerrit.testing.InMemoryRepositoryManager; +import com.google.gerrit.testing.InMemoryTestEnvironment; +import com.google.inject.Inject; +import com.googlesource.gerrit.plugins.multisite.Configuration; +import com.googlesource.gerrit.plugins.multisite.validation.dfsrefdb.RefFixture; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import org.eclipse.jgit.internal.storage.dfs.InMemoryRepository; +import org.eclipse.jgit.junit.LocalDiskRepositoryTestCase; +import org.eclipse.jgit.junit.TestRepository; +import org.eclipse.jgit.lib.ObjectId; +import org.eclipse.jgit.lib.ObjectIdRef; +import org.eclipse.jgit.lib.Ref; +import org.eclipse.jgit.transport.RemoteRefUpdate; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.class) +public class MultisiteReplicationPushFilterTest extends LocalDiskRepositoryTestCase + implements RefFixture { + + @Rule public InMemoryTestEnvironment testEnvironment = new InMemoryTestEnvironment(); + + @Mock SharedRefDatabaseWrapper sharedRefDatabaseMock; + @Mock Configuration config; + @Mock Configuration.ReplicationFilter replicationFilterConfig; + + @Inject private InMemoryRepositoryManager gitRepositoryManager; + + String project = A_TEST_PROJECT_NAME; + Project.NameKey projectName = A_TEST_PROJECT_NAME_KEY; + + private TestRepository<InMemoryRepository> repo; + + @Before + public void setupTestRepo() throws Exception { + InMemoryRepository inMemoryRepo = + gitRepositoryManager.createRepository(A_TEST_PROJECT_NAME_KEY); + doReturn(replicationFilterConfig).when(config).replicationFilter(); + repo = new TestRepository<>(inMemoryRepo); + } + + @Test + public void shouldReturnAllRefUpdatesWhenAllUpToDate() throws Exception { + List<RemoteRefUpdate> refUpdates = + Arrays.asList(refUpdate("refs/heads/foo"), refUpdate("refs/heads/bar")); + doReturn(true).when(sharedRefDatabaseMock).isUpToDate(eq(projectName), any()); + + MultisiteReplicationPushFilter pushFilter = + new MultisiteReplicationPushFilter(sharedRefDatabaseMock, gitRepositoryManager, config); + List<RemoteRefUpdate> filteredRefUpdates = pushFilter.filter(project, refUpdates); + + assertThat(filteredRefUpdates).containsExactlyElementsIn(refUpdates); + } + + @Test + public void shouldFilterOutOneOutdatedRef() throws Exception { + RemoteRefUpdate refUpToDate = refUpdate("refs/heads/uptodate"); + RemoteRefUpdate outdatedRef = refUpdate("refs/heads/outdated"); + List<RemoteRefUpdate> refUpdates = Arrays.asList(refUpToDate, outdatedRef); + SharedRefDatabaseWrapper sharedRefDatabase = + new FakeSharedRefDatabaseWrapper(outdatedRef.getSrcRef()); + + MultisiteReplicationPushFilter pushFilter = + new MultisiteReplicationPushFilter(sharedRefDatabase, gitRepositoryManager, config); + List<RemoteRefUpdate> filteredRefUpdates = pushFilter.filter(project, refUpdates); + + assertThat(filteredRefUpdates).containsExactly(refUpToDate); + } + + @Test + public void shouldLoadLocalVersionAndNotFilter() throws Exception { + String refName = "refs/heads/temporaryOutdated"; + RemoteRefUpdate temporaryOutdated = refUpdate(refName); + ObjectId latestObjectId = repo.getRepository().exactRef(refName).getObjectId(); + + List<RemoteRefUpdate> refUpdates = Collections.singletonList(temporaryOutdated); + doReturn(false).doReturn(true).when(sharedRefDatabaseMock).isUpToDate(eq(projectName), any()); + + MultisiteReplicationPushFilter pushFilter = + new MultisiteReplicationPushFilter(sharedRefDatabaseMock, gitRepositoryManager, config); + List<RemoteRefUpdate> filteredRefUpdates = pushFilter.filter(project, refUpdates); + + assertThat(filteredRefUpdates).hasSize(1); + assertThat(filteredRefUpdates.get(0).getNewObjectId()).isEqualTo(latestObjectId); + + verify(sharedRefDatabaseMock, times(2)).isUpToDate(any(), any()); + } + + @Test + public void shouldLoadLocalVersionAndFilter() throws Exception { + RemoteRefUpdate temporaryOutdated = refUpdate("refs/heads/temporaryOutdated"); + repo.branch("refs/heads/temporaryOutdated").commit().create(); + List<RemoteRefUpdate> refUpdates = Collections.singletonList(temporaryOutdated); + doReturn(false).doReturn(false).when(sharedRefDatabaseMock).isUpToDate(eq(projectName), any()); + + MultisiteReplicationPushFilter pushFilter = + new MultisiteReplicationPushFilter(sharedRefDatabaseMock, gitRepositoryManager, config); + List<RemoteRefUpdate> filteredRefUpdates = pushFilter.filter(project, refUpdates); + + assertThat(filteredRefUpdates).isEmpty(); + verify(sharedRefDatabaseMock, times(2)).isUpToDate(any(), any()); + } + + @Test + public void shouldFilterOutAllOutdatedChangesRef() throws Exception { + RemoteRefUpdate refUpToDate = refUpdate("refs/heads/uptodate"); + RemoteRefUpdate refChangeUpToDate = refUpdate("refs/changes/25/1225/2"); + RemoteRefUpdate changeMetaRef = refUpdate("refs/changes/12/4512/meta"); + RemoteRefUpdate changeRef = refUpdate("refs/changes/12/4512/1"); + List<RemoteRefUpdate> refUpdates = + Arrays.asList(refUpToDate, refChangeUpToDate, changeMetaRef, changeRef); + SharedRefDatabaseWrapper sharedRefDatabase = + new FakeSharedRefDatabaseWrapper(changeMetaRef.getSrcRef()); + + MultisiteReplicationPushFilter pushFilter = + new MultisiteReplicationPushFilter(sharedRefDatabase, gitRepositoryManager, config); + List<RemoteRefUpdate> filteredRefUpdates = pushFilter.filter(project, refUpdates); + + assertThat(filteredRefUpdates).containsExactly(refUpToDate, refChangeUpToDate); + } + + private RemoteRefUpdate refUpdate(String refName) throws Exception { + ObjectId srcObjId = ObjectId.fromString("0000000000000000000000000000000000000001"); + Ref srcRef = new ObjectIdRef.Unpeeled(Ref.Storage.NEW, refName, srcObjId); + repo.branch(refName).commit().create(); + return new RemoteRefUpdate(null, srcRef, "origin", false, "origin", srcObjId); + } +}
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateTest.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateTest.java index 88d2270..d99fb4b 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateTest.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/ProjectVersionRefUpdateTest.java
@@ -18,16 +18,21 @@ import static com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF; import static com.googlesource.gerrit.plugins.multisite.validation.ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_VALUE_REF; import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.isNull; import static org.mockito.Mockito.atMost; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.verifyZeroInteractions; import static org.mockito.Mockito.when; import com.gerritforge.gerrit.globalrefdb.validation.ProjectsFilter; import com.gerritforge.gerrit.globalrefdb.validation.SharedRefDatabaseWrapper; +import com.google.common.base.Suppliers; import com.google.gerrit.entities.Project; import com.google.gerrit.entities.RefNames; +import com.google.gerrit.server.data.RefUpdateAttribute; import com.google.gerrit.server.events.Event; import com.google.gerrit.server.events.RefUpdatedEvent; import com.google.gerrit.server.extensions.events.GitReferenceUpdated; @@ -95,16 +100,9 @@ Context.setForwardedEvent(false); when(sharedRefDb.get( A_TEST_PROJECT_NAME_KEY, - ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF, - String.class)) - .thenReturn(Optional.of("26f7ee61bf0e470e8393c884526eec8a9b943a63")); - when(sharedRefDb.get( - A_TEST_PROJECT_NAME_KEY, ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_VALUE_REF, String.class)) .thenReturn(Optional.of("" + (masterCommit.getCommitTime() - 1))); - when(sharedRefDb.compareAndPut(any(Project.NameKey.class), any(Ref.class), any(ObjectId.class))) - .thenReturn(true); when(sharedRefDb.compareAndPut(any(Project.NameKey.class), any(String.class), any(), any())) .thenReturn(true); when(refUpdatedEvent.getProjectNameKey()).thenReturn(A_TEST_PROJECT_NAME_KEY); @@ -129,6 +127,35 @@ } @Test + public void producerShouldUsePutInsteadOfCompareAndPutWhenExtendedGlobalRefDb() + throws IOException { + when(sharedRefDb.isSetOperationSupported()).thenReturn(true); + RefUpdatedEvent refUpdatedEvent = new RefUpdatedEvent(); + RefUpdateAttribute refUpdatedAttribute = new RefUpdateAttribute(); + refUpdatedAttribute.project = A_TEST_PROJECT_NAME_KEY.get(); + refUpdatedAttribute.refName = A_TEST_REF_NAME; + refUpdatedEvent.refUpdate = Suppliers.memoize(() -> refUpdatedAttribute); + + new ProjectVersionRefUpdateImpl( + repoManager, sharedRefDb, gitReferenceUpdated, verLogger, projectsFilter) + .onEvent(refUpdatedEvent); + + Ref ref = repo.getRepository().findRef(MULTI_SITE_VERSIONING_REF); + + verify(sharedRefDb, never()) + .compareAndPut(any(Project.NameKey.class), anyString(), anyLong(), anyLong()); + + verify(sharedRefDb).put(any(Project.NameKey.class), any(String.class), any(String.class)); + assertThat(ref).isNotNull(); + + ObjectLoader loader = repo.getRepository().open(ref.getObjectId()); + long storedVersion = readLongObject(loader); + assertThat(storedVersion).isGreaterThan((long) masterCommit.getCommitTime()); + + verify(verLogger).log(A_TEST_PROJECT_NAME_KEY, storedVersion, 0); + } + + @Test public void producerShouldUpdateProjectVersionUponForcedPushRefUpdatedEvent() throws Exception { Context.setForwardedEvent(false); @@ -137,27 +164,20 @@ Thread.sleep(1000L); repo.branch("master").update(masterCommit); - - when(sharedRefDb.get( - A_TEST_PROJECT_NAME_KEY, - ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF, - String.class)) - .thenReturn(Optional.of("26f7ee61bf0e470e8393c884526eec8a9b943a63")); when(sharedRefDb.get( A_TEST_PROJECT_NAME_KEY, ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_VALUE_REF, String.class)) .thenReturn(Optional.of("" + (masterCommit.getCommitTime() - 1))); - when(sharedRefDb.compareAndPut(any(Project.NameKey.class), any(Ref.class), any(ObjectId.class))) - .thenReturn(true); when(sharedRefDb.compareAndPut(any(Project.NameKey.class), any(String.class), any(), any())) .thenReturn(true); when(refUpdatedEvent.getProjectNameKey()).thenReturn(A_TEST_PROJECT_NAME_KEY); when(refUpdatedEvent.getRefName()).thenReturn(A_TEST_REF_NAME); - new ProjectVersionRefUpdateImpl( - repoManager, sharedRefDb, gitReferenceUpdated, verLogger, projectsFilter) - .onEvent(refUpdatedEvent); + ProjectVersionRefUpdateImpl projectVersion = + new ProjectVersionRefUpdateImpl( + repoManager, sharedRefDb, gitReferenceUpdated, verLogger, projectsFilter); + projectVersion.onEvent(refUpdatedEvent); Ref ref = repo.getRepository().findRef(MULTI_SITE_VERSIONING_REF); @@ -168,6 +188,10 @@ ObjectLoader loader = repo.getRepository().open(ref.getObjectId()); long storedVersion = readLongObject(loader); + + Optional<Long> localStoredVersion = projectVersion.getProjectLocalVersion(A_TEST_PROJECT_NAME); + assertThat(localStoredVersion).isEqualTo(Optional.of(storedVersion)); + assertThat(storedVersion).isGreaterThan((long) masterPlusOneCommit.getCommitTime()); verify(verLogger).log(A_TEST_PROJECT_NAME_KEY, storedVersion, 0); @@ -179,17 +203,10 @@ Context.setForwardedEvent(false); when(sharedRefDb.get( A_TEST_PROJECT_NAME_KEY, - ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_REF, - String.class)) - .thenReturn(Optional.empty()); - when(sharedRefDb.get( - A_TEST_PROJECT_NAME_KEY, ProjectVersionRefUpdate.MULTI_SITE_VERSIONING_VALUE_REF, String.class)) .thenReturn(Optional.empty()); - when(sharedRefDb.compareAndPut(any(Project.NameKey.class), any(Ref.class), any(ObjectId.class))) - .thenReturn(true); when(sharedRefDb.compareAndPut(any(Project.NameKey.class), any(String.class), any(), any())) .thenReturn(true); when(refUpdatedEvent.getProjectNameKey()).thenReturn(A_TEST_PROJECT_NAME_KEY);
diff --git a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/dfsrefdb/RefFixture.java b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/dfsrefdb/RefFixture.java index baba94b..1885d27 100644 --- a/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/dfsrefdb/RefFixture.java +++ b/src/test/java/com/googlesource/gerrit/plugins/multisite/validation/dfsrefdb/RefFixture.java
@@ -33,6 +33,8 @@ static final ObjectId AN_OBJECT_ID_1 = new ObjectId(1, 2, 3, 4, 5); static final ObjectId AN_OBJECT_ID_2 = new ObjectId(1, 2, 3, 4, 6); static final ObjectId AN_OBJECT_ID_3 = new ObjectId(1, 2, 3, 4, 7); + static final ObjectId AN_OUTDATED_OBJECT_ID = + ObjectId.fromString("da37cb098dd7df4c8662ae8afc1acae5f7567775"); static final String A_TEST_REF_NAME = "refs/heads/master"; static final String A_REF_NAME_OF_A_PATCHSET = "refs/changes/01/1/1";