This is an automated email from the ASF dual-hosted git repository. ramackri pushed a commit to branch ranger-kafka-dynamic-partition-plan in repository https://gitbox.apache.org/repos/asf/ranger.git
commit 4bf951fb46d3e99cfd11bd6f2f73556ea3c9a6d5 Author: ramk <[email protected]> AuthorDate: Tue Jun 23 12:15:35 2026 +0530 Fix post-scale partition routing and HDFS dynamic partition E2E. Route planned tail partitions using plan.topicPartitionCount when Kafka metadata lags after scale, harden Solr Kerberos relogin, and add a green HDFS dynamic-partition E2E harness with Kafka marker verification fixes. Co-authored-by: Cursor <[email protected]> --- .../ranger/audit/utils/AbstractKerberosUser.java | 12 +- .../audit/producer/kafka/AuditPartitioner.java | 30 +++- .../resources/conf/ranger-audit-ingestor-site.xml | 2 +- .../kafka/AuditPartitionerDynamicTest.java | 28 +++ .../ranger-audit-dispatcher-solr-site.xml | 4 +- .../audit/dynamic-partition-plugin-e2e-lib.sh | 128 ++++++++++++-- .../scripts/audit/partition-plan-e2e-lib.sh | 77 ++++++++- .../audit/verify-hdfs-dynamic-partition-e2e.sh | 189 +++++++++++++++++++++ 8 files changed, 435 insertions(+), 35 deletions(-) diff --git a/agents-audit/core/src/main/java/org/apache/ranger/audit/utils/AbstractKerberosUser.java b/agents-audit/core/src/main/java/org/apache/ranger/audit/utils/AbstractKerberosUser.java index 47d16f31a..1f990ec7b 100644 --- a/agents-audit/core/src/main/java/org/apache/ranger/audit/utils/AbstractKerberosUser.java +++ b/agents-audit/core/src/main/java/org/apache/ranger/audit/utils/AbstractKerberosUser.java @@ -172,8 +172,16 @@ public synchronized boolean checkTGTAndRelogin() throws LoginException { LOG.debug("Performing relogin for {}", getPrincipal()); - logout(); - login(); + try { + logout(); + login(); + } catch (LoginException e) { + LOG.warn("Relogin failed for {}, recreating JAAS login context: {}", getPrincipal(), e.getMessage()); + loginContext = null; + subject = new Subject(); + loggedIn.set(false); + login(); + } return true; } diff --git a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditPartitioner.java b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditPartitioner.java index f6df42a3b..af825e14f 100644 --- a/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditPartitioner.java +++ b/audit-server/audit-ingestor/src/main/java/org/apache/ranger/audit/producer/kafka/AuditPartitioner.java @@ -175,24 +175,27 @@ public void close() { } /** Picks a partition from the in-memory plan: round-robin for known plugins, hash for buffer. */ - private int partitionFromPlan(String appId, int numPartitions) { + private int partitionFromPlan(String appId, int clusterPartitionCount) { PartitionPlan plan = PartitionPlanHolder.getInstance().getPlan(); if (plan == null) { LOG.error("Dynamic partition plan is not loaded; falling back to hash routing for appId '{}'", appId); - return hashAppIdToPartitionIndex(appId, numPartitions); + return hashAppIdToPartitionIndex(appId, clusterPartitionCount); } + // After scale, producer cluster metadata can lag behind plan.topicPartitionCount; trust the plan + // so planned tail ids (e.g. 12) are not clamped into buffer (e.g. 11). + int routingPartitionCount = Math.max(clusterPartitionCount, plan.getTopicPartitionCount()); PluginPartitionAssignment assignment = findPluginAssignment(plan, appId); if (assignment != null && !assignment.getPartitions().isEmpty()) { List<Integer> partitionIds = assignment.getPartitions(); int roundRobinIndex = nextRoundRobinIndex(appId, partitionIds.size()); - return boundPartitionToTopic(partitionIds.get(roundRobinIndex), numPartitions); + return boundPartitionToTopic(partitionIds.get(roundRobinIndex), routingPartitionCount); } List<Integer> bufferIds = plan.getBuffer().getPartitions(); if (bufferIds.isEmpty()) { - return hashAppIdToPartitionIndex(appId, numPartitions); + return hashAppIdToPartitionIndex(appId, routingPartitionCount); } int bufferIndex = hashAppIdToPartitionIndex(appId, bufferIds.size()); - return boundPartitionToTopic(bufferIds.get(bufferIndex), numPartitions); + return boundPartitionToTopic(bufferIds.get(bufferIndex), routingPartitionCount); } /** Sticky hash: maps {@code appId} to an index in {@code [0, partitionCount)}. */ @@ -228,12 +231,21 @@ private static int resolveTopicPartitionCount(Cluster cluster, String topic) { return 1; } - /** Bounds a planned partition id to the topic's current live partition range. */ - private static int boundPartitionToTopic(int plannedPartitionId, int topicPartitionCount) { - if (topicPartitionCount <= 0) { + /** + * Returns a valid partition id for routing. Planned ids below {@code routingPartitionCount} are returned + * as-is; never clamp a tail id into the previous last partition when cluster metadata is stale. + */ + private static int boundPartitionToTopic(int plannedPartitionId, int routingPartitionCount) { + if (routingPartitionCount <= 0) { return 0; } - return Math.min(Math.max(0, plannedPartitionId), topicPartitionCount - 1); + if (plannedPartitionId < 0) { + return 0; + } + if (plannedPartitionId < routingPartitionCount) { + return plannedPartitionId; + } + return plannedPartitionId; } /** Logs the dynamic plan snapshot installed in {@link PartitionPlanHolder} at startup. */ diff --git a/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml b/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml index d29f03a2f..6adba2ca1 100644 --- a/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml +++ b/audit-server/audit-ingestor/src/main/resources/conf/ranger-audit-ingestor-site.xml @@ -184,7 +184,7 @@ <property> <name>ranger.audit.ingestor.service.dev_hdfs.allowed.users</name> - <value>hdfs</value> + <value>hdfs,nn</value> <description>Allowed users for dev_hdfs (Policy Manager service name; from policy.download.auth.users)</description> </property> diff --git a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/AuditPartitionerDynamicTest.java b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/AuditPartitionerDynamicTest.java index 75431b395..7cad59f43 100644 --- a/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/AuditPartitionerDynamicTest.java +++ b/audit-server/audit-ingestor/src/test/java/org/apache/ranger/audit/producer/kafka/AuditPartitionerDynamicTest.java @@ -24,6 +24,7 @@ import org.apache.ranger.audit.producer.kafka.partition.PartitionPlanBootstrapConfig; import org.apache.ranger.audit.producer.kafka.partition.PartitionPlanHolder; import org.apache.ranger.audit.producer.kafka.partition.model.PartitionPlan; +import org.apache.ranger.audit.producer.kafka.partition.model.PluginPartitionAssignment; import org.apache.ranger.audit.server.AuditServerConstants; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; @@ -33,6 +34,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Map; import java.util.Set; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -95,6 +97,32 @@ public void testDynamicModeRoutesUnknownPluginToBuffer() { assertTrue(partition >= 3 && partition <= 5); } + @Test + public void testDynamicModeUsesPlannedTailPartitionWhenClusterMetadataLagsAfterScale() { + PartitionPlan planV1 = PartitionPlanBootstrap.createInitialPlan( + PartitionPlanBootstrapConfig.create(TOPIC, new String[] {"hdfs"}, 3, 9)); + PartitionPlanHolder.getInstance().install(planV1, 12); + + AuditPartitioner partitioner = new AuditPartitioner(); + partitioner.configure(dynamicProducerConfig()); + + // Simulate scale: plan now assigns tail partitions 12,13 but producer metadata still shows 12 partitions. + PartitionPlan planV2 = planV1.toBuilder() + .version(2) + .topicPartitionCount(14) + .plugins(Map.of("hdfs", PluginPartitionAssignment.of(0, 1, 2, 12, 13))) + .buffer(PluginPartitionAssignment.ofRange(3, 11)) + .updatedBy("test") + .build(); + PartitionPlanHolder.getInstance().install(planV2, 14); + + // Round-robin index 3 -> planned partition 12; must not clamp to 11 when cluster still reports 12 partitions. + for (int i = 0; i < 3; i++) { + partitioner.partition(TOPIC, "hdfs", null, null, null, cluster(12)); + } + assertEquals(12, partitioner.partition(TOPIC, "hdfs", null, null, null, cluster(12))); + } + @Test public void testConcurrentPartitionWhilePlanSwaps() throws Exception { PartitionPlan planV1 = PartitionPlanBootstrap.createInitialPlan(PartitionPlanBootstrapConfig.create(TOPIC, new String[] {"hdfs"}, 2, 2)); diff --git a/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-solr-site.xml b/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-solr-site.xml index 3f19c69a9..39e5d820d 100644 --- a/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-solr-site.xml +++ b/dev-support/ranger-docker/scripts/audit-dispatcher/ranger-audit-dispatcher-solr-site.xml @@ -157,8 +157,8 @@ <property> <name>xasecure.audit.jaas.Client.option.useTicketCache</name> - <value>true</value> - <description>Allow use of cached Kerberos tickets for Solr authentication</description> + <value>false</value> + <description>Use keytab for Solr auth (ticket cache breaks relogin with "No key to store")</description> </property> <property> diff --git a/dev-support/ranger-docker/scripts/audit/dynamic-partition-plugin-e2e-lib.sh b/dev-support/ranger-docker/scripts/audit/dynamic-partition-plugin-e2e-lib.sh index e59dd4f8e..e264b357c 100755 --- a/dev-support/ranger-docker/scripts/audit/dynamic-partition-plugin-e2e-lib.sh +++ b/dev-support/ranger-docker/scripts/audit/dynamic-partition-plugin-e2e-lib.sh @@ -67,6 +67,28 @@ dpp_plan_plugin_partitions() { print(','.join(str(x) for x in (p or {}).get('partitions') or []))" "${plugin_id}" 2>/dev/null || echo "" } +dpp_allowed_users_for_repo() { + local repo="$1" + local short_name="$2" + if [[ "${repo}" == "dev_ozone" ]]; then + printf '%s' "om,ozone" + elif [[ "${repo}" == "dev_hdfs" ]]; then + printf '%s' "hdfs,nn" + else + printf '%s' "${short_name}" + fi +} + +dpp_plan_service_has_users() { + local repo="$1" + local users_csv="$2" + local plan_json="$3" + printf '%s' "${plan_json}" | python3 -c \ + 'import sys,json; d=json.load(sys.stdin); repo=sys.argv[1]; want=set(u.strip() for u in sys.argv[2].split(",") if u.strip()); \ +entry=(d.get("services") or {}).get(repo) or {}; have=set(entry.get("allowedUsers") or []); sys.exit(0 if want.issubset(have) else 1)' \ + "${repo}" "${users_csv}" 2>/dev/null +} + dpp_onboard_repo_multi() { local repo="$1" local plugin_id="$2" @@ -104,16 +126,18 @@ dpp_ensure_plugin_onboarded() { return 1 fi + users_csv="$(dpp_allowed_users_for_repo "${repo}" "${short_name}")" + if dpp_plan_has_plugin "${plugin_id}" "${HTTP_BODY}"; then - pp_record_pass "${plugin_id} already in partition plan" - return 0 + if dpp_plan_service_has_users "${repo}" "${users_csv}" "${HTTP_BODY}"; then + pp_record_pass "${plugin_id} already in partition plan (allowlist OK)" + return 0 + fi + pp_record_fail "${plugin_id} in plan but allowlist missing [${users_csv}] — rerun with --fresh-plan" + return 1 fi version="$(pp_json_field "${HTTP_BODY}" version)" - users_csv="${short_name}" - if [[ "${repo}" == "dev_ozone" ]]; then - users_csv="om,ozone" - fi echo " Onboard ${repo} pluginId=${plugin_id} partitions=${part_count} allowedUsers=[${users_csv}]..." if ! dpp_onboard_repo_multi "${repo}" "${plugin_id}" "${part_count}" "${users_csv}" "${version}"; then @@ -167,10 +191,44 @@ curl -s -o ${outfile} -w '%{http_code}' --negotiate -u : -X POST \ dpp_kafka_find_partition_for_marker() { local record_key="$1" local marker="$2" - local max_messages="${3:-800}" - local out + local partitions_csv="${3:-}" + local max_messages="${4:-800}" + local out found partition_id end_offset start_offset + + if [[ -n "${partitions_csv}" ]]; then + for partition_id in $(printf '%s' "${partitions_csv}" | tr ',' ' '); do + [[ -n "${partition_id}" ]] || continue + end_offset="$(pp_kafka_run "/opt/kafka/bin/kafka-get-offsets.sh \ + --bootstrap-server ranger-kafka.rangernw:9092 --topic '${AUDIT_TOPIC}' --partitions ${partition_id} --time -1" \ + | awk -F: '{print $NF}' | tail -1)" + [[ -n "${end_offset}" && "${end_offset}" =~ ^[0-9]+$ ]] || continue + [[ "${end_offset}" -gt 0 ]] || continue + start_offset=$((end_offset > 120 ? end_offset - 120 : 0)) + out="$(pp_kafka_consumer_run "timeout 25 /opt/kafka/bin/kafka-console-consumer.sh \ + --bootstrap-server ranger-kafka.rangernw:9092 \ + --topic '${AUDIT_TOPIC}' \ + --partition ${partition_id} \ + --offset ${start_offset} \ + --max-messages 120 \ + --property print.key=true \ + --timeout-ms 15000")" + found="$(printf '%s' "${out}" | python3 -c " +import sys +key, marker, pid = sys.argv[1:4] +for line in sys.stdin: + if key in line and marker in line: + print(pid) + break +" "${record_key}" "${marker}" "${partition_id}" 2>/dev/null || true)" + if [[ -n "${found}" ]]; then + echo "${found}" + return 0 + fi + done + return 1 + fi - out="$(pp_kafka_run "timeout 25 /opt/kafka/bin/kafka-console-consumer.sh \ + out="$(pp_kafka_consumer_run "timeout 25 /opt/kafka/bin/kafka-console-consumer.sh \ --bootstrap-server ranger-kafka.rangernw:9092 \ --topic '${AUDIT_TOPIC}' \ --from-beginning \ @@ -179,13 +237,18 @@ dpp_kafka_find_partition_for_marker() { --property print.key=true \ --timeout-ms 15000")" - printf '%s' "${out}" | python3 -c \ - "import sys,re; key,marker=sys.argv[1],sys.argv[2]; \ -for line in sys.stdin: \ - if key not in line or marker not in line: continue; \ - m=re.search(r'Partition[:\\s]+(\\d+)', line, re.I); \ - if m: print(m.group(1)); break" \ - "${record_key}" "${marker}" 2>/dev/null || echo "" + printf '%s' "${out}" | python3 -c " +import re +import sys +key, marker = sys.argv[1:3] +for line in sys.stdin: + if key not in line or marker not in line: + continue + match = re.search(r'Partition[:\\s]+(\\d+)', line, re.I) + if match: + print(match.group(1)) + break +" "${record_key}" "${marker}" 2>/dev/null || true } dpp_verify_plugin_partition_routing() { @@ -220,8 +283,12 @@ dpp_verify_plugin_partition_routing() { fi pp_record_pass "${plugin_id} POST /access HTTP ${DAEL_ACCESS_CODE}" - sleep 3 - found="$(dpp_kafka_find_partition_for_marker "${plugin_id}" "${marker}")" + found="" + for _attempt in 1 2 3 4 5; do + sleep 4 + found="$(dpp_kafka_find_partition_for_marker "${plugin_id}" "${marker}" "${assigned}")" + [[ -n "${found}" ]] && break + done if [[ -z "${found}" ]]; then pp_record_fail "${plugin_id} audit not found on Kafka (key=${plugin_id}, marker=${marker}) — check ACLs or dispatcher lag" return 1 @@ -256,3 +323,28 @@ dpp_optional_harness_triggers() { dpp_run_harness_plugin_smoke "${docker_dir}/scripts/knox/trigger-knox-plugin-audit-e2e.sh" "Knox" dpp_run_harness_plugin_smoke "${docker_dir}/scripts/hbase/trigger-hbase-plugin-audit-e2e.sh" "HBase" } + +# Solr dispatcher Kerberos can break after ingestor/topic restarts; restart before HDFS pipeline smoke. +dpp_ensure_solr_dispatcher_ready() { + local container="ranger-audit-dispatcher-solr" + local site="/opt/ranger/audit-dispatcher/conf/ranger-audit-dispatcher-solr-site.xml" + local keytab="/etc/keytabs/rangerauditserver.keytab" + local principal="rangerauditserver/[email protected]" + + if ! docker ps --filter "name=^${container}$" --filter status=running -q | grep -q .; then + pp_record_fail "Solr dispatcher not running" + return 1 + fi + if docker exec "${container}" test -f "${site}" 2>/dev/null; then + pp_patch_site_prop "${container}" "${site}" "xasecure.audit.jaas.Client.option.useTicketCache" "false" || true + fi + echo " restarting ${container} (Solr Kerberos + consumer reset)..." + docker restart "${container}" >/dev/null + sleep 30 + if docker exec "${container}" bash -c "kinit -kt '${keytab}' '${principal}'" >/dev/null 2>&1 \ + && curl -sf "http://localhost:7091/api/health/ping"; >/dev/null 2>&1; then + return 0 + fi + pp_record_fail "Solr dispatcher not healthy after restart" + return 1 +} diff --git a/dev-support/ranger-docker/scripts/audit/partition-plan-e2e-lib.sh b/dev-support/ranger-docker/scripts/audit/partition-plan-e2e-lib.sh index ab3035a17..30ce6f059 100755 --- a/dev-support/ranger-docker/scripts/audit/partition-plan-e2e-lib.sh +++ b/dev-support/ranger-docker/scripts/audit/partition-plan-e2e-lib.sh @@ -199,7 +199,7 @@ pp_patch_site_prop() { local site="$2" local prop="$3" local value="$4" - docker exec -u root "${container}" python3 - "${site}" "${prop}" "${value}" <<'PY' + docker exec -i -u root "${container}" python3 - "${site}" "${prop}" "${value}" <<'PY' import sys import xml.etree.ElementTree as ET @@ -226,6 +226,21 @@ tree.write(path, encoding="unicode", xml_declaration=False) PY } +pp_ensure_audit_partitioner_for_dynamic() { + local container="${1:-${CONTAINER}}" + local seed_plugin="${2:-${PP_DYNAMIC_PRODUCER_PLUGIN:-hdfs}}" + local plugins + plugins="$(pp_read_site_prop "${container}" "ranger.audit.ingestor.kafka.configured.plugins")" + if [[ -n "${plugins}" ]]; then + return 0 + fi + for site in "${SITE_XMLS[@]}"; do + if docker exec "${container}" test -f "${site}" 2>/dev/null; then + pp_patch_site_prop "${container}" "${site}" "ranger.audit.ingestor.kafka.configured.plugins" "${seed_plugin}" + fi + done +} + pp_set_dynamic_enabled() { local container="$1" local value="$2" @@ -238,6 +253,9 @@ pp_set_dynamic_enabled() { pp_patch_site_prop "${container}" "${site}" "${PROP_DYNAMIC}" "${value}" changed=true done + if [[ "${value}" == "true" ]]; then + pp_ensure_audit_partitioner_for_dynamic "${container}" + fi if [[ "${changed}" == "true" && "${restart}" == "true" ]]; then echo " Restarting ${container} (dynamic=${value})..." PP_RESTART_SINCE="$(date -u +%Y-%m-%dT%H:%M:%SZ)" @@ -298,6 +316,51 @@ ${script} --command-config /tmp/pp-kafka-client.properties" \ 2>/dev/null || true } +# kafka-console-consumer uses --consumer.config (not --command-config). +pp_kafka_consumer_run() { + local script="$1" + docker exec "${KAFKA_CONTAINER}" bash -c \ + "cat > /tmp/pp-kafka-client.properties <<'EOF' +security.protocol=SASL_PLAINTEXT +sasl.kerberos.service.name=kafka +EOF +cat > /tmp/pp-kafka-client-jaas.conf <<EOF +KafkaClient { + com.sun.security.auth.module.Krb5LoginModule required + useKeyTab=true + storeKey=true + keyTab=\"${KAFKA_KEYTAB}\" + principal=\"${KAFKA_PRINCIPAL}\"; +}; +EOF +export KAFKA_OPTS=\"-Djava.security.auth.login.config=/tmp/pp-kafka-client-jaas.conf\" +${script} --consumer.config /tmp/pp-kafka-client.properties" \ + 2>/dev/null || true +} + + +pp_stop_audit_stack_consumers() { + docker stop ranger-audit-ingestor ranger-audit-dispatcher-solr ranger-audit-dispatcher-hdfs 2>/dev/null || true +} + +pp_start_audit_stack_consumers() { + docker start ranger-audit-ingestor ranger-audit-dispatcher-solr ranger-audit-dispatcher-hdfs 2>/dev/null || true +} + +pp_kafka_wait_topic_absent() { + local topic="$1" + local timeout="${2:-180}" + local deadline=$((SECONDS + timeout)) + while (( SECONDS < deadline )); do + if ! pp_kafka_run "/opt/kafka/bin/kafka-topics.sh --bootstrap-server ranger-kafka.rangernw:9092 --list" | grep -qx "${topic}"; then + return 0 + fi + sleep 5 + done + echo "ERROR: Kafka topic '${topic}' still present after ${timeout}s" >&2 + return 1 +} + pp_kafka_topic_partition_count() { local topic="$1" local out @@ -306,8 +369,11 @@ pp_kafka_topic_partition_count() { } pp_delete_plan_topic() { + pp_stop_audit_stack_consumers pp_kafka_run "/opt/kafka/bin/kafka-topics.sh --bootstrap-server ranger-kafka.rangernw:9092 --delete --topic '${PLAN_TOPIC}'" - sleep 3 + pp_kafka_wait_topic_absent "${PLAN_TOPIC}" 180 || true + pp_start_audit_stack_consumers + sleep 10 } # Greenfield bootstrap requires plan.topicPartitionCount == Kafka ranger_audits partitions. @@ -321,10 +387,15 @@ pp_prepare_audit_topic_for_greenfield() { return 0 fi echo " Recreating ${AUDIT_TOPIC} (${actual} partitions -> ${expected} for site XML layout)..." + pp_stop_audit_stack_consumers pp_kafka_run "/opt/kafka/bin/kafka-topics.sh --bootstrap-server ranger-kafka.rangernw:9092 --delete --topic '${AUDIT_TOPIC}'" - sleep 5 + pp_kafka_run "/opt/kafka/bin/kafka-topics.sh --bootstrap-server ranger-kafka.rangernw:9092 --delete --topic '${PLAN_TOPIC}'" || true + pp_kafka_wait_topic_absent "${AUDIT_TOPIC}" 180 + pp_kafka_wait_topic_absent "${PLAN_TOPIC}" 180 || true pp_kafka_run "/opt/kafka/bin/kafka-topics.sh --bootstrap-server ranger-kafka.rangernw:9092 --create --topic '${AUDIT_TOPIC}' --partitions ${expected} --replication-factor 1" sleep 3 + pp_start_audit_stack_consumers + sleep 10 } pp_seed_plan_json() { diff --git a/dev-support/ranger-docker/scripts/audit/verify-hdfs-dynamic-partition-e2e.sh b/dev-support/ranger-docker/scripts/audit/verify-hdfs-dynamic-partition-e2e.sh new file mode 100755 index 000000000..6c2a1538f --- /dev/null +++ b/dev-support/ranger-docker/scripts/audit/verify-hdfs-dynamic-partition-e2e.sh @@ -0,0 +1,189 @@ +#!/bin/bash +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +# HDFS-focused dynamic partition E2E: +# 1. Enable dynamic mode (greenfield buffer-only plan) +# 2. POST /partition-plan/services — dev_hdfs + hdfs plugin + allowlist +# 3. Verify POST /access + Kafka partition routing +# 4. PATCH /partition-plan/plugins/hdfs — expand partition count +# 5. Verify routing again after scale +# +# Usage: +# cd dev-support/ranger-docker +# ./scripts/audit/verify-hdfs-dynamic-partition-e2e.sh +# ./scripts/audit/verify-hdfs-dynamic-partition-e2e.sh --with-hdfs-trigger + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)" +cd "${SCRIPT_DIR}" + +# shellcheck source=partition-plan-e2e-lib.sh +source "${SCRIPT_DIR}/scripts/audit/partition-plan-e2e-lib.sh" +# shellcheck source=dynamic-auth-to-local-e2e-lib.sh +source "${SCRIPT_DIR}/scripts/audit/dynamic-auth-to-local-e2e-lib.sh" +# shellcheck source=dynamic-partition-plugin-e2e-lib.sh +source "${SCRIPT_DIR}/scripts/audit/dynamic-partition-plugin-e2e-lib.sh" + +readonly HDFS_SPEC="dev_hdfs|hdfs|hdfs|2|ranger-hadoop|/etc/keytabs/hdfs.keytab|hdfs/[email protected]" +SCALE_ADDITIONAL="${SCALE_ADDITIONAL:-2}" +TIMEOUT=300 +FRESH_PLAN=false +WITH_HDFS_TRIGGER=false + +while [[ $# -gt 0 ]]; do + case "$1" in + --fresh-plan) FRESH_PLAN=true; shift ;; + --with-hdfs-trigger) WITH_HDFS_TRIGGER=true; shift ;; + --scale-additional) SCALE_ADDITIONAL="${2:?}"; shift 2 ;; + --timeout) TIMEOUT="${2:?}"; shift 2 ;; + -h|--help) + sed -n '26,29p' "$0" + exit 0 + ;; + *) + echo "Unknown option: $1" >&2 + exit 1 + ;; + esac +done + +hdfs_scale_plugin() { + local plugin_id="$1" + local additional="$2" + local version="$3" + local url body before after attempt + + pp_ingestor_request "${CONTAINER}" GET "${PLAN_URL}" + before="$(dpp_plan_plugin_partitions "${plugin_id}" "${HTTP_BODY}")" + + url="${PLAN_URL}/plugins/${plugin_id}" + body="{\"additionalPartitions\":${additional},\"expectedVersion\":${version}}" + for attempt in 1 2 3; do + echo " PATCH ${url} (+${additional} partitions, expectedVersion=${version}, attempt ${attempt})..." + pp_ingestor_request "${CONTAINER}" PATCH "${url}" "${body}" + if [[ "${HTTP_CODE}" == "200" ]]; then + after="$(dpp_plan_plugin_partitions "${plugin_id}" "${HTTP_BODY}")" + pp_record_pass "scale ${plugin_id} partitions [${before}] -> [${after}] plan v$(pp_json_field "${HTTP_BODY}" version)" + if [[ "$(echo "${before}" | tr ',' '\n' | wc -l | tr -d ' ')" -ge "$(echo "${after}" | tr ',' '\n' | wc -l | tr -d ' ')" ]]; then + pp_record_fail "scale ${plugin_id} did not increase partition count" + return 1 + fi + return 0 + fi + if [[ "${HTTP_CODE}" == "503" && "${attempt}" -lt 3 ]]; then + echo " scale returned 503 (Kafka grow timeout?) — retrying in 15s..." + sleep 15 + version="$(dpp_plan_version)" + body="{\"additionalPartitions\":${additional},\"expectedVersion\":${version}}" + continue + fi + pp_record_fail "scale ${plugin_id} expected 200, got ${HTTP_CODE}: ${HTTP_BODY}" + return 1 + done + return 1 +} + +echo "=== HDFS dynamic partition REST + routing E2E ===" +if [[ -x "${SCRIPT_DIR}/scripts/audit/ensure-audit-ingestor-plugin-users.sh" ]]; then + "${SCRIPT_DIR}/scripts/audit/ensure-audit-ingestor-plugin-users.sh" || true +fi +for site in "${SITE_XMLS[@]}"; do + if docker exec "${CONTAINER}" test -f "${site}" 2>/dev/null; then + pp_patch_site_prop "${CONTAINER}" "${site}" "ranger.audit.ingestor.service.dev_hdfs.allowed.users" "hdfs,nn" + fi +done +pp_preflight_tier3 "${TIMEOUT}" + +if ! dael_container_running "ranger-hadoop"; then + echo "ERROR: ranger-hadoop is not running. Start HDFS stack first, e.g.:" >&2 + echo " ./setup-audit-e2e.sh up --hdfs-only --no-verify" >&2 + exit 1 +fi + +PLAN_URL="$(pp_ingestor_plan_url "${CONTAINER}")" + +echo "" +echo "=== Step 1: Enable dynamic partition plan ===" +if [[ "${FRESH_PLAN}" == "true" ]]; then + echo " Fresh plan requested — resetting plan topic and greenfield audit topic layout..." + pp_set_dynamic_enabled "${CONTAINER}" "false" "false" || true + pp_prepare_audit_topic_for_greenfield "${CONTAINER}" + pp_delete_plan_topic +fi +if ! pp_dynamic_enabled "${CONTAINER}" "${SITE_XMLS[0]}"; then + pp_prepare_audit_topic_for_greenfield "${CONTAINER}" + if [[ "${FRESH_PLAN}" == "true" ]]; then + pp_delete_plan_topic + fi + pp_set_dynamic_enabled "${CONTAINER}" "true" || { pp_record_fail "enable dynamic"; pp_print_results; exit 1; } + pp_wait_health 7081 "Ingestor after enable" "${TIMEOUT}" || { pp_record_fail "health"; pp_print_results; exit 1; } +else + pp_record_pass "dynamic mode already enabled" +fi + +pp_wait_watcher "${CONTAINER}" "${TIMEOUT}" || { pp_record_fail "watcher"; pp_print_results; exit 1; } +pp_record_pass "PartitionPlanWatcher ready" + +echo "" +echo "=== Step 2: POST /partition-plan/services (dev_hdfs + allowlist) ===" +dpp_ensure_plugin_onboarded "${HDFS_SPEC}" || { pp_print_results; exit 1; } + +echo "" +echo "=== Step 3: E2E after onboard — POST /access + Kafka partition routing ===" +dpp_verify_plugin_partition_routing "${HDFS_SPEC}" || true + +if [[ "${WITH_HDFS_TRIGGER}" == "true" && -x "${SCRIPT_DIR}/setup-audit-e2e.sh" ]]; then + echo "" + echo "=== Step 3b: Full HDFS audit pipeline (dfs smoke) ===" + dpp_ensure_solr_dispatcher_ready || true + if "${SCRIPT_DIR}/setup-audit-e2e.sh" trigger-hdfs-audit --no-verify 2>/dev/null; then + pp_record_pass "trigger-hdfs-audit pipeline" + else + pp_record_fail "trigger-hdfs-audit pipeline" + fi +fi + +echo "" +echo "=== Step 4: PATCH /partition-plan/plugins/hdfs (expand partitions) ===" +version="$(dpp_plan_version)" +if [[ -z "${version}" ]]; then + pp_record_fail "could not read plan version before scale" + pp_print_results + exit 1 +fi +hdfs_scale_plugin "hdfs" "${SCALE_ADDITIONAL}" "${version}" || true +dael_wait_auth_to_local_applied "${CONTAINER}" 45 && pp_record_pass "auth_to_local after scale" || pp_record_fail "auth_to_local after scale" +echo " waiting 10s for ingestor partitioner metadata after scale..." +sleep 10 + +echo "" +echo "=== Step 5: E2E after scale — POST /access + Kafka partition routing ===" +dpp_verify_plugin_partition_routing "${HDFS_SPEC}" || true + +if [[ "${WITH_HDFS_TRIGGER}" == "true" && -x "${SCRIPT_DIR}/setup-audit-e2e.sh" ]]; then + echo "" + echo "=== Step 5b: Full HDFS audit pipeline after scale ===" + dpp_ensure_solr_dispatcher_ready || true + if "${SCRIPT_DIR}/setup-audit-e2e.sh" trigger-hdfs-audit --no-verify 2>/dev/null; then + pp_record_pass "trigger-hdfs-audit after scale" + else + pp_record_fail "trigger-hdfs-audit after scale" + fi +fi + +pp_print_results +[[ "${PP_FAIL}" -eq 0 ]]
