rameeshm commented on code in PR #1137: URL: https://github.com/apache/ranger/pull/1137#discussion_r3787180218
########## agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanValidator.java: ########## @@ -0,0 +1,144 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.ranger.audit.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.ranger.audit.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.partition.model.PartitionPlan; +import org.apache.ranger.audit.partition.model.PluginEntry; + +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Checks partition plan shape and append-only updates. */ +public final class PartitionPlanValidator { + private PartitionPlanValidator() { + } + + public static void validate(PartitionPlan plan) { + validate(plan, null); + } + + /** + * When kafkaPartitionCount is set, it must be at least plan.topicPartitionCount. + * Extra live Kafka partitions (e.g. after static-mode migration) are allowed. + */ + public static void validate(PartitionPlan plan, Integer kafkaPartitionCount) { + if (plan == null || StringUtils.isBlank(plan.getTopic()) || plan.getVersion() < AuditPartitionPlanConstants.INITIAL_PLAN_VERSION || plan.getTopicPartitionCount() < 1) { + throw new PartitionPlanException("Invalid partition plan"); + } + if (kafkaPartitionCount != null && kafkaPartitionCount < plan.getTopicPartitionCount()) { + throw new PartitionPlanException("Kafka topic has fewer partitions than plan requires"); + } + + Set<Integer> assigned = new HashSet<>(); + registerPartitions(plan.getBuffer().getPartitions(), assigned, true); + for (Map.Entry<String, PluginEntry> entry : plan.getPlugins().entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + throw new PartitionPlanException("Plugin id is required"); + } + registerPartitions(entry.getValue().getPartitions(), assigned, false); + } + if (assigned.size() != plan.getTopicPartitionCount()) { Review Comment: If topicPartitionCount = 3 and partition array [2, 4, 6] has be injected into partition plan then we may miss a partition in the hash map. Please check this out. ########## agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanAllocator.java: ########## @@ -0,0 +1,257 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.ranger.audit.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.ranger.audit.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.partition.model.BufferEntry; +import org.apache.ranger.audit.partition.model.PartitionPlan; +import org.apache.ranger.audit.partition.model.PluginEntry; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; + +import static java.util.Objects.requireNonNull; + +/** Append-only plan updates for Admin-managed audit partition routing. */ +public final class PartitionPlanAllocator { + private PartitionPlanAllocator() { + } + + /** + * Onboard a Ranger service repo under a plugin type. Promotes the plugin from buffer when needed, + * otherwise adds the service to an existing plugin entry. + */ + public static PartitionPlan onboardService(PartitionPlan current, String pluginId, String serviceName, int partitionCount, String updatedBy) { + requireMutationInputs(current, pluginId, partitionCount, updatedBy); + if (StringUtils.isBlank(serviceName)) { + throw new PartitionPlanException("serviceName is required"); + } + String trimmedService = serviceName.trim(); + PluginEntry existing = current.getPlugins().get(pluginId); + if (existing != null) { + return addServiceToPlugin(current, pluginId, trimmedService, updatedBy); + } + return promotePlugin(current, pluginId, partitionCount, updatedBy, trimmedService); + } + + /** Adds a service repo to an already-promoted plugin without changing partition assignment. */ + public static PartitionPlan addServiceToPlugin(PartitionPlan current, String pluginId, String serviceName, String updatedBy) { + if (current == null) { + throw new PartitionPlanException("Current plan is required"); + } + PartitionPlanValidator.validate(current); + if (StringUtils.isBlank(pluginId) || StringUtils.isBlank(serviceName) || StringUtils.isBlank(updatedBy)) { + throw new PartitionPlanException("pluginId, serviceName, and updatedBy are required"); + } + PluginEntry existing = current.getPlugins().get(pluginId); + if (existing == null) { + throw new PartitionPlanException("Plugin '" + pluginId + "' is not configured; promote it first"); + } + String trimmedService = serviceName.trim(); + if (existing.getServices().contains(trimmedService)) { + return current; + } + ensureServiceNotAssignedElsewhere(current.getPlugins(), pluginId, trimmedService); + + Map<String, PluginEntry> plugins = new LinkedHashMap<>(current.getPlugins()); + plugins.put(pluginId, existing.addService(trimmedService)); + return commitPlanUpdate(current, updatedBy, current.getTopicPartitionCount(), plugins, current.getBuffer().getPartitions()); + } + + /** Removes a service repo from whichever plugin currently owns it. */ + public static PartitionPlan removeService(PartitionPlan current, String serviceName, String updatedBy) { + if (current == null) { + throw new PartitionPlanException("Current plan is required"); + } + PartitionPlanValidator.validate(current); + if (StringUtils.isBlank(serviceName) || StringUtils.isBlank(updatedBy)) { + throw new PartitionPlanException("serviceName and updatedBy are required"); + } + String trimmedService = serviceName.trim(); + String owningPluginId = findPluginForService(current.getPlugins(), trimmedService); + if (owningPluginId == null) { + return current; + } + + PluginEntry existing = requireNonNull(current.getPlugins().get(owningPluginId)); + List<String> remainingServices = new ArrayList<>(existing.getServices()); + remainingServices.remove(trimmedService); + + Map<String, PluginEntry> plugins = new LinkedHashMap<>(current.getPlugins()); + plugins.put(owningPluginId, existing.withServices(remainingServices)); + return commitPlanUpdate(current, updatedBy, current.getTopicPartitionCount(), plugins, current.getBuffer().getPartitions()); + } + + /** + * Give a plugin its own partitions. Uses buffer IDs first; adds new tail IDs when buffer is too small. + * Optionally attaches {@code serviceName} to the new plugin entry. + */ + public static PartitionPlan promotePlugin(PartitionPlan current, String pluginId, int partitionCount, String updatedBy, String serviceName) { + requireMutationInputs(current, pluginId, partitionCount, updatedBy); + if (current.getPlugins().containsKey(pluginId)) { + assertPromoteNotConflicting(current, pluginId, partitionCount, serviceName); + throw new PartitionPlanException("Plugin '" + pluginId + "' already has dedicated partitions"); + } + if (StringUtils.isNotBlank(serviceName)) { + ensureServiceNotAssignedElsewhere(current.getPlugins(), pluginId, serviceName.trim()); + } + + List<Integer> remainingBuffer = new ArrayList<>(current.getBuffer().getPartitions()); + List<Integer> newPluginIds = takeFromBuffer(remainingBuffer, partitionCount); + int topicPartitionCount = current.getTopicPartitionCount(); + int additionalNeeded = partitionCount - newPluginIds.size(); + if (additionalNeeded > 0) { + topicPartitionCount = appendTailPartitions(newPluginIds, topicPartitionCount, additionalNeeded, collectAssignedPartitionIds(current)); + } + + List<String> services = StringUtils.isNotBlank(serviceName) ? List.of(serviceName.trim()) : List.of(); + Map<String, PluginEntry> plugins = addPluginAssignment(current, pluginId, newPluginIds, services); + return commitPlanUpdate(current, updatedBy, topicPartitionCount, plugins, remainingBuffer); + } + + public static boolean isOnboardAlreadyApplied(PartitionPlan current, String pluginId, String serviceName, int partitionCount) { + if (current == null || StringUtils.isBlank(serviceName)) { + return false; + } + PluginEntry existing = current.getPlugins().get(pluginId); + if (existing == null) { + return false; + } + return existing.getPartitions().size() == partitionCount && existing.getServices().contains(serviceName.trim()); + } + + /** Updates audit POST allow-list metadata; bumps version only when the map changes. */ + public static PartitionPlan updateServiceAllowedUsers(PartitionPlan current, Map<String, List<String>> serviceAllowedUsers, String updatedBy) { + if (current == null) { + throw new PartitionPlanException("Current plan is required"); + } + PartitionPlanValidator.validate(current); + if (StringUtils.isBlank(updatedBy)) { + throw new PartitionPlanException("updatedBy is required"); + } + + Map<String, List<String>> normalized = PolicyDownloadAuthUsersUtil.normalizeServiceAllowedUsers(serviceAllowedUsers); + if (Objects.equals(current.getServiceAllowedUsers(), normalized)) { + return current; + } + + PartitionPlan next = current.toBuilder() + .version(current.getVersion() + 1) + .serviceAllowedUsers(normalized) + .updatedAt(Instant.now().toString()) + .updatedBy(updatedBy) + .build(); + PartitionPlanValidator.validate(next); + PartitionPlanValidator.validateAppendOnly(current, next); + return next; + } + + private static List<Integer> takeFromBuffer(List<Integer> bufferIds, int count) { + List<Integer> taken = new ArrayList<>(Math.min(count, bufferIds.size())); + while (taken.size() < count && !bufferIds.isEmpty()) { + taken.add(bufferIds.remove(0)); + } + return taken; + } + + private static int appendTailPartitions(List<Integer> target, int topicPartitionCount, int count, Set<Integer> assigned) { Review Comment: Can we simplify this like the following as it is strictly append-only and partition IDs are a continuous 1..topicPartitionCount private static int appendTailPartitions(List<Integer> target, int topicPartitionCount, int count) { int nextId = topicPartitionCount + 1; for (int i = 0; i < count; i++) { target.add(nextId++); } return topicPartitionCount + count; } ########## agents-common/src/main/java/org/apache/ranger/audit/partition/PartitionPlanValidator.java: ########## @@ -0,0 +1,144 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.ranger.audit.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.ranger.audit.partition.exception.PartitionPlanException; +import org.apache.ranger.audit.partition.model.PartitionPlan; +import org.apache.ranger.audit.partition.model.PluginEntry; + +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** Checks partition plan shape and append-only updates. */ +public final class PartitionPlanValidator { + private PartitionPlanValidator() { + } + + public static void validate(PartitionPlan plan) { + validate(plan, null); + } + + /** + * When kafkaPartitionCount is set, it must be at least plan.topicPartitionCount. + * Extra live Kafka partitions (e.g. after static-mode migration) are allowed. + */ + public static void validate(PartitionPlan plan, Integer kafkaPartitionCount) { + if (plan == null || StringUtils.isBlank(plan.getTopic()) || plan.getVersion() < AuditPartitionPlanConstants.INITIAL_PLAN_VERSION || plan.getTopicPartitionCount() < 1) { + throw new PartitionPlanException("Invalid partition plan"); + } + if (kafkaPartitionCount != null && kafkaPartitionCount < plan.getTopicPartitionCount()) { + throw new PartitionPlanException("Kafka topic has fewer partitions than plan requires"); + } + + Set<Integer> assigned = new HashSet<>(); + registerPartitions(plan.getBuffer().getPartitions(), assigned, true); + for (Map.Entry<String, PluginEntry> entry : plan.getPlugins().entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + throw new PartitionPlanException("Plugin id is required"); + } + registerPartitions(entry.getValue().getPartitions(), assigned, false); + } + if (assigned.size() != plan.getTopicPartitionCount()) { + throw new PartitionPlanException("topicPartitionCount must equal the union of all assigned partition ids"); + } + validateServiceUniqueness(plan.getPlugins()); + validateServiceAllowedUsers(plan.getServiceAllowedUsers()); + } + + /** + * When a service repo is listed in {@code serviceAllowedUsers}, it must have at least one + * allowed short username (from Admin {@code policy.download.auth.users}). + */ + public static void validateServiceAllowedUsers(Map<String, List<String>> serviceAllowedUsers) { + if (serviceAllowedUsers == null || serviceAllowedUsers.isEmpty()) { + return; + } + for (Map.Entry<String, List<String>> entry : serviceAllowedUsers.entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + throw new PartitionPlanException("Service repo name is required"); + } + List<String> users = entry.getValue(); + if (users == null || users.isEmpty()) { + throw new PartitionPlanException( + "allowedUsers must not be empty for service '" + entry.getKey().trim() + "'"); + } + } + } + + /** Each Ranger service repo name may appear in at most one plugin entry. */ + public static void validateServiceUniqueness(Map<String, PluginEntry> plugins) { + if (plugins == null || plugins.isEmpty()) { + return; + } + Set<String> seenServices = new HashSet<>(); + for (Map.Entry<String, PluginEntry> entry : plugins.entrySet()) { + for (String serviceName : entry.getValue().getServices()) { + if (!seenServices.add(serviceName)) { + throw new PartitionPlanException("Service '" + serviceName + "' is assigned to more than one plugin"); + } + } + } + } + + /** New plan must only add tail partitions; existing plugin lists stay unchanged in order. */ + public static void validateAppendOnly(PartitionPlan current, PartitionPlan proposed) { + if (current == null || proposed == null) { + throw new PartitionPlanException("Current and proposed plans are required"); + } + if (proposed.getTopicPartitionCount() < current.getTopicPartitionCount() || proposed.getVersion() != current.getVersion() + 1) { + throw new PartitionPlanException("Plan must grow partition count and increment version by one"); + } + + for (Map.Entry<String, PluginEntry> entry : current.getPlugins().entrySet()) { + String pluginId = entry.getKey(); + List<Integer> before = entry.getValue().getPartitions(); + PluginEntry afterEntry = proposed.getPlugins().get(pluginId); + if (afterEntry == null) { + throw new PartitionPlanException("Append-only violation for plugin '" + pluginId + "'"); + } + List<Integer> after = afterEntry.getPartitions(); + if (after.size() < before.size()) { + throw new PartitionPlanException("Append-only violation for plugin '" + pluginId + "'"); + } + for (int i = 0; i < before.size(); i++) { Review Comment: Can this for loop and check be replaced by if (!after.subList(0, before.size()).equals(before)) { throw new PartitionPlanException("Append-only violation for plugin '" + pluginId + "': existing partitions reshuffled"); } ########## agents-common/src/main/java/org/apache/ranger/audit/partition/PolicyDownloadAuthUsersUtil.java: ########## @@ -0,0 +1,74 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.ranger.audit.partition; + +import org.apache.commons.lang3.StringUtils; +import org.apache.ranger.plugin.model.RangerService; + +import java.util.Arrays; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; + +/** Parses {@code policy.download.auth.users} service config for audit ingestor authorization. */ +public final class PolicyDownloadAuthUsersUtil { + public static final String CONFIG_NAME = "policy.download.auth.users"; + + private PolicyDownloadAuthUsersUtil() { + } + + public static List<String> parseUsers(RangerService service) { + if (service == null || service.getConfigs() == null) { + return Collections.emptyList(); + } + return parseUsers(service.getConfigs().get(CONFIG_NAME)); + } + + public static List<String> parseUsers(String configValue) { + if (StringUtils.isBlank(configValue)) { + return Collections.emptyList(); + } + return Arrays.stream(configValue.split(",")) + .map(String::trim) + .filter(StringUtils::isNotBlank) + .filter(user -> !"*".equals(user)) + .collect(Collectors.toList()); + } + + public static Map<String, List<String>> normalizeServiceAllowedUsers(Map<String, List<String>> serviceAllowedUsers) { + if (serviceAllowedUsers == null || serviceAllowedUsers.isEmpty()) { + return Collections.emptyMap(); + } + Map<String, List<String>> normalized = new LinkedHashMap<>(); + for (Map.Entry<String, List<String>> entry : serviceAllowedUsers.entrySet()) { + if (StringUtils.isBlank(entry.getKey())) { + continue; + } + List<String> users = entry.getValue() == null ? Collections.emptyList() : parseUsers(String.join(",", entry.getValue())); Review Comment: can be done like List<String> users = entry.getValue() == null ? Collections.emptyList() : entry.getValue().stream() .flatMap(s -> parseUsers(s).stream()) .collect(Collectors.toList()); -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
