shounakmk219 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1265250678
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotInstanceRestletResource.java:
##########
@@ -416,4 +422,115 @@ public List<OperationValidationResponse>
instanceDropSafetyCheck(
Response.Status.INTERNAL_SERVER_ERROR, e);
}
}
+
+ @POST
+ @Path("/instances/updateTags/validate")
+ @Produces(MediaType.APPLICATION_JSON)
+ @ApiOperation(value = "Check if it's safe to update the tags of the given
instances. If not list all the reasons.")
+ @ApiResponses(value = {
+ @ApiResponse(code = 200, message = "Success"),
+ @ApiResponse(code = 500, message = "Internal error")
+ })
+ public List<OperationValidationResponse>
instanceTagUpdateSafetyCheck(List<InstanceTagUpdateRequest> instances) {
+ LOGGER.info("Performing safety check on tag update request received for
instances: {}",
+
instances.stream().map(InstanceTagUpdateRequest::getInstanceName).collect(Collectors.toList()));
+ Map<String, Integer> tagMinServerMap =
_pinotHelixResourceManager.minimumInstancesRequiredForTags();
+ Map<String, Integer> tagDeficiency = computeTagDeficiency(instances,
tagMinServerMap);
+
+ Map<String, List<OperationValidationResponse.ErrorWrapper>> responseMap =
new HashMap<>(instances.size());
+ List<OperationValidationResponse.ErrorWrapper> tenantIssues = new
ArrayList<>();
+ instances.forEach(instance -> responseMap.put(instance.getInstanceName(),
new ArrayList<>()));
+ for (InstanceTagUpdateRequest instance : instances) {
+ String name = instance.getInstanceName();
+ Set<String> oldTags;
+ try {
+ oldTags = new
HashSet<>(_pinotHelixResourceManager.getTagsForInstance(name));
+ } catch (NullPointerException exception) {
+ throw new ControllerApplicationException(LOGGER,
+ String.format("Instance %s is not a valid instance name.", name),
Response.Status.PRECONDITION_FAILED);
+ }
+ Set<String> newTags = new HashSet<>(instance.getNewTags());
+ // tags removed from instance
+ for (String tag : Sets.difference(oldTags, newTags)) {
+ Integer deficiency = tagDeficiency.get(tag);
+ if (deficiency != null && deficiency > 0) {
+ String tenant = TagNameUtils.getTenantFromTag(tag);
+ String tagType = getInstanceTypeFromTag(tag);
+ responseMap.get(name).add(new
OperationValidationResponse.ErrorWrapper(
+
OperationValidationResponse.ErrorCode.MINIMUM_INSTANCE_UNSATISFIED, tenant,
tagType, tag, tagType, name));
+ tagDeficiency.put(tag, deficiency - 1);
+ }
+ }
+ // newly added tags to instance
+ for (String tag : newTags) {
+ String tagType = getInstanceTypeFromTag(tag);
+ if (tagType == null) {
+ responseMap.get(name).add(new
OperationValidationResponse.ErrorWrapper(
+ OperationValidationResponse.ErrorCode.UNRECOGNISED_TAG_TYPE,
tag));
+ continue;
+ }
+ Integer deficiency = tagDeficiency.get(tag);
+ if (deficiency != null && deficiency > 0) {
+ tenantIssues.add(new OperationValidationResponse.ErrorWrapper(
+ OperationValidationResponse.ErrorCode.ALREADY_DEFICIENT_TENANT,
TagNameUtils.getTenantFromTag(tag),
+ tagType, deficiency.toString(), name));
+ }
+ }
+ }
+
+ // consolidate all the issues based on instances
+ List<OperationValidationResponse> response = new
ArrayList<>(instances.size());
+ responseMap.forEach((instance, issueList) ->
response.add(issueList.isEmpty()
+ ? new
OperationValidationResponse().setInstanceName(instance).setSafe(true)
+ : new
OperationValidationResponse().putAllIssues(issueList).setInstanceName(instance).setSafe(false)));
+ // separate entry to group all the deficient tenant issues as it's not
related to any instance
+ if (!tenantIssues.isEmpty()) {
+ response.add(new
OperationValidationResponse().putAllIssues(tenantIssues).setSafe(false));
+ }
+ return response;
+ }
+
+ private String getInstanceTypeFromTag(String tag) {
+ if (TagNameUtils.isServerTag(tag)) {
+ return "server";
+ } else if (TagNameUtils.isBrokerTag(tag)) {
+ return "broker";
+ } else {
+ return null;
+ }
+ }
+
+ private Map<String, Integer>
computeTagDeficiency(List<InstanceTagUpdateRequest> instances,
+ Map<String, Integer> tagMinServerMap) {
+ Map<String, Integer> updatedTagInstanceMap =
getUpdatedTagInstanceMap(instances);
+ Map<String, Integer> tagDeficiency = new HashMap<>();
+ tagMinServerMap.forEach((tag, minInstances) -> {
+ Integer updatedInstances = updatedTagInstanceMap.remove(tag);
+ tagDeficiency.put(tag, minInstances - (updatedInstances != null ?
updatedInstances : 0));
+ });
+ updatedTagInstanceMap.forEach((tag, updatedInstances) ->
tagDeficiency.put(tag, 0));
+ return tagDeficiency;
+ }
+
+ private Map<String, Integer>
getUpdatedTagInstanceMap(List<InstanceTagUpdateRequest> instances) {
+ Map<String, Integer> updatedTagInstanceMap = new HashMap<>();
+ Set<String> visitedInstances = new HashSet<>();
+ instances.forEach(instance -> {
+ instance.getNewTags().forEach(tag -> {
+ Integer count = updatedTagInstanceMap.get(tag);
+ updatedTagInstanceMap.put(tag, count != null ? count + 1 : 1);
Review Comment:
done
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
