snleee commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1270389415
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotInstanceRestletResource.java:
##########
@@ -416,4 +422,112 @@ 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)) {
Review Comment:
Does this mean that we only support update the entire tag list `List<String>
tags` and not support to add an extra tag to the existing tag list?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotInstanceRestletResource.java:
##########
@@ -416,4 +422,112 @@ 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) {
Review Comment:
It would be great if we can provide the documentation on what are we
validating here.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotInstanceRestletResource.java:
##########
@@ -416,4 +422,112 @@ 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 &&
(name.startsWith(CommonConstants.Helix.PREFIX_OF_BROKER_INSTANCE)
+ ||
name.startsWith(CommonConstants.Helix.PREFIX_OF_SERVER_INSTANCE))) {
+ 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,
Review Comment:
`instances` -> `requests`?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotInstanceRestletResource.java:
##########
@@ -416,4 +422,112 @@ 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) {
Review Comment:
`instances` -> `requests`?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotInstanceRestletResource.java:
##########
@@ -416,4 +422,112 @@ 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();
Review Comment:
Why don't we compute minInstances only for tags that show up in the request?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotInstanceRestletResource.java:
##########
@@ -416,4 +422,112 @@ 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 &&
(name.startsWith(CommonConstants.Helix.PREFIX_OF_BROKER_INSTANCE)
+ ||
name.startsWith(CommonConstants.Helix.PREFIX_OF_SERVER_INSTANCE))) {
+ 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,
Review Comment:
`computeTagDeficiency()` and `getUpdatedTagInstanceMap()` are hard to follow
the logic due to a lot of stream is being used. Can you add the documentation
on what we are trying to compute here? Also, can we try to review the logic and
see if we can simplify some parts (if possible)?
##########
pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotInstanceRestletResource.java:
##########
@@ -416,4 +422,112 @@ 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 &&
(name.startsWith(CommonConstants.Helix.PREFIX_OF_BROKER_INSTANCE)
+ ||
name.startsWith(CommonConstants.Helix.PREFIX_OF_SERVER_INSTANCE))) {
+ 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),
Review Comment:
What's the difference between `ALREADY_DEFICIENT_TENANT` vs
`MINIMUM_INSTANCE_UNSATISFIED`?
Aren't they essentially the same error?
--
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]
