You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by "shounakmk219 (via GitHub)" <gi...@apache.org> on 2023/07/11 15:51:44 UTC
[GitHub] [pinot] shounakmk219 opened a new pull request, #11077: Instance retag validation check api
shounakmk219 opened a new pull request, #11077:
URL: https://github.com/apache/pinot/pull/11077
Expose an endpoint to perform safety check validation of the instance retag/ tag update operation.
Right now in order to update the tags on an instance we use
- `/instances/{instanceName}/updateTags`
**Problem**
This operation blindly updates the instance tags without taking into consideration the possible consequences such as table replication not being satsfied.
**Proposal**
Provide a validation API that takes the desired user state as input and provides the user with the information if the retag is safe else give all the possible reasons for it being unsafe.
Endpoint : `/instances/updateTags/validate`
Method type : `POST`
Request body :
```
[
{
"instanceName": "Server_a.c.com_20000",
"newTags": [
"tenantA_OFFLINE",
"tenantA_REALTIME"
]
},
{
"instanceName": "Server_b.c.com_20000",
"newTags": [
"tenantB_OFFLINE",
"tenantB_REALTIME"
]
}
]
```
Sample Response :
```
[
{
"instanceName": "Broker_127.0.0.1_8000",
"issues": [
{
"code": "MINIMUM_INSTANCE_UNSATISFIED",
"message": "Tenant 'DefaultTenant' will not satisfy minimum 'broker' requirement if tag 'DefaultTenant_BROKER' is removed from broker instance 'Broker_127.0.0.1_8000'."
},
{
"code": "UNRECOGNISED_TAG_TYPE",
"message": "The tag 'tenantA' does not follow the suffix convention of either broker or server"
}
],
"isSafe": false
},
{
"instanceName": "Server_127.0.0.1_7050",
"issues": [
{
"code": "MINIMUM_INSTANCE_UNSATISFIED",
"message": "Tenant 'DefaultTenant' will not satisfy minimum 'server' requirement if tag 'DefaultTenant_REALTIME' is removed from server instance 'Server_127.0.0.1_7050'."
}
],
"isSafe": false
},
{
"instanceName": null,
"issues": [
{
"code": "ALREADY_DEFICIENT_TENANT",
"message": "Tenant 'DefaultTenant_OFFLINE' is low on 'server' instances by 1 even with given allocation"
}
],
"isSafe": false
}
]
```
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] shounakmk219 commented on a diff in pull request #11077: Instance retag validation check api
Posted by "shounakmk219 (via GitHub)" <gi...@apache.org>.
shounakmk219 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1270578701
##########
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:
`MINIMUM_INSTANCE_UNSATISFIED ` -> The tenant currently satisfies minimum instances requirement, but the validation request removes few instances from it due to which the tenant ends up with instance deficiency.
`ALREADY_DEFICIENT_TENANT` -> The tenant is already deficient on instances, and even though the validation request adds few instances to it, its still deficient.
I have made this segregation so that user is aware about the existing deficiencies and not wonder why the validation is failing even though we are adding instances to a tenant.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] shounakmk219 commented on a diff in pull request #11077: Instance retag validation check api
Posted by "shounakmk219 (via GitHub)" <gi...@apache.org>.
shounakmk219 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1270557583
##########
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:
Sure, will add some doc and comments as well
##########
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:
Will do
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] saurabhd336 commented on a diff in pull request #11077: (WIP) Instance retag validation check api
Posted by "saurabhd336 (via GitHub)" <gi...@apache.org>.
saurabhd336 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1263552337
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4128,6 +4138,33 @@ public PeriodicTaskInvocationResponse invokeControllerPeriodicTask(String tableN
return new PeriodicTaskInvocationResponse(periodicTaskRequestId, messageCount > 0);
}
+ /**
+ * Construct a map of all the tags and their respective minimum instance requirements.
+ * The minimum instance requirement is computed by
+ * - for BROKER tenant tag set it to 1.
+ * - for SERVER tenant tag iterate over all the tables of that tenant and find the maximum table replication.
+ * - for rest of the tags just set it to 0
+ * @return map of tags and their minimum instance requirements
+ */
+ public Map<String, Integer> minimumInstancesRequiredForTags() {
+ Map<String, Integer> tagMinServerMap = new HashMap<>();
+ for (InstanceConfig instanceConfig : getAllHelixInstanceConfigs()) {
+ for (String tag : instanceConfig.getTags()) {
+ tagMinServerMap.put(tag, TagNameUtils.isBrokerTag(tag) ? 1 : 0);
+ }
+ }
+ for (TableConfig tableConfig : getAllTableConfigs()) {
+ String serverTag = TagNameUtils.getServerTagForTenant(tableConfig.getTenantConfig().getServer(),
+ tableConfig.getTableType());
+ int maxReplication = Math.max(Objects.requireNonNullElse(tagMinServerMap.get(serverTag), 0),
+ tableConfig.getReplication());
+ tagMinServerMap.put(serverTag, maxReplication);
Review Comment:
Can be replaced with
``tagMinServerMap.put(serverTag, Math.max(tagMinServerMap.getOrDefault(serverTag, 0), tableConfig.getReplication()) )```
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] shounakmk219 commented on a diff in pull request #11077: Instance retag validation check api
Posted by "shounakmk219 (via GitHub)" <gi...@apache.org>.
shounakmk219 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1270555663
##########
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:
Just the tags in request won't do, we also need to fetch the existing/old tags first too for minInstances. We can pass a list of tags to get required minInstances but anyway we are forced to parse all the table configs hence didn’t complicate things for now. Let me know if accepting the tags list makes more sense and I will be happy to do it that way.
##########
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:
Just the tags in request won't do, we also need to fetch the existing/old tags first too for minInstances. We can pass a list of tags to get required minInstances but anyway we are forced to parse all the table configs hence didn’t complicate things for now. Let me know if accepting the tags list makes more sense and I will be happy to do it that way.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] shounakmk219 commented on a diff in pull request #11077: (WIP) Instance retag validation check api
Posted by "shounakmk219 (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] shounakmk219 commented on a diff in pull request #11077: (WIP) Instance retag validation check api
Posted by "shounakmk219 (via GitHub)" <gi...@apache.org>.
shounakmk219 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1262024816
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4128,6 +4129,29 @@ public PeriodicTaskInvocationResponse invokeControllerPeriodicTask(String tableN
return new PeriodicTaskInvocationResponse(periodicTaskRequestId, messageCount > 0);
}
+ /**
+ * Construct a map of all the tenants and their respective minimum server requirements.
+ * The minimum server requirement is computed by iterating over all the tables of the tenant and
+ * find the table with maximum replication.
+ * @return map of tenants and their minimum server requirements
+ */
+ public Map<String, Integer> minimumServersRequiredForTenants() {
+ Map<String, Integer> tenantMinServerMap = new HashMap<>();
+ getAllServerTenantNames().forEach(tenant -> tenantMinServerMap.put(tenant, 0));
+ for (String table : getAllTables()) {
+ TableConfig tableConfig = getTableConfig(table);
Review Comment:
Makes sense to pull all the table configs at once and iterate on them directly. Not sure if passing the list would be a good option atleast for now as that puts unnecessary burden on the caller function.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] codecov-commenter commented on pull request #11077: (WIP) Instance retag validation check api
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #11077:
URL: https://github.com/apache/pinot/pull/11077#issuecomment-1631133395
## [Codecov](https://app.codecov.io/gh/apache/pinot/pull/11077?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#11077](https://app.codecov.io/gh/apache/pinot/pull/11077?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (6831db1) into [master](https://app.codecov.io/gh/apache/pinot/commit/6d99eec9977677cca1493e6ac0ecfd7d20502cdf?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (6d99eec) will **increase** coverage by `0.00%`.
> The diff coverage is `0.00%`.
```diff
@@ Coverage Diff @@
## master #11077 +/- ##
=========================================
Coverage 0.11% 0.11%
=========================================
Files 2200 2148 -52
Lines 118802 115664 -3138
Branches 17991 17577 -414
=========================================
Hits 137 137
+ Misses 118645 115507 -3138
Partials 20 20
```
| Flag | Coverage Δ | |
|---|---|---|
| integration1temurin11 | `?` | |
| integration1temurin17 | `?` | |
| integration1temurin20 | `?` | |
| integration2temurin17 | `?` | |
| integration2temurin20 | `?` | |
| unittests1temurin11 | `?` | |
| unittests1temurin17 | `?` | |
| unittests1temurin20 | `?` | |
| unittests2temurin11 | `?` | |
| unittests2temurin17 | `?` | |
| unittests2temurin20 | `0.11% <0.00%> (+<0.01%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Impacted Files](https://app.codecov.io/gh/apache/pinot/pull/11077?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...roller/api/resources/InstanceTagUpdateRequest.java](https://app.codecov.io/gh/apache/pinot/pull/11077?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9hcGkvcmVzb3VyY2VzL0luc3RhbmNlVGFnVXBkYXRlUmVxdWVzdC5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...ler/api/resources/OperationValidationResponse.java](https://app.codecov.io/gh/apache/pinot/pull/11077?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9hcGkvcmVzb3VyY2VzL09wZXJhdGlvblZhbGlkYXRpb25SZXNwb25zZS5qYXZh) | `0.00% <0.00%> (ø)` | |
| [...er/api/resources/PinotInstanceRestletResource.java](https://app.codecov.io/gh/apache/pinot/pull/11077?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9hcGkvcmVzb3VyY2VzL1Bpbm90SW5zdGFuY2VSZXN0bGV0UmVzb3VyY2UuamF2YQ==) | `0.00% <0.00%> (ø)` | |
| [...ntroller/helix/core/PinotHelixResourceManager.java](https://app.codecov.io/gh/apache/pinot/pull/11077?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cGlub3QtY29udHJvbGxlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcGlub3QvY29udHJvbGxlci9oZWxpeC9jb3JlL1Bpbm90SGVsaXhSZXNvdXJjZU1hbmFnZXIuamF2YQ==) | `0.00% <0.00%> (ø)` | |
... and [88 files with indirect coverage changes](https://app.codecov.io/gh/apache/pinot/pull/11077/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] shounakmk219 commented on a diff in pull request #11077: Instance retag validation check api
Posted by "shounakmk219 (via GitHub)" <gi...@apache.org>.
shounakmk219 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1270529277
##########
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:
will change it
##########
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:
will change it
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] saurabhd336 commented on a diff in pull request #11077: (WIP) Instance retag validation check api
Posted by "saurabhd336 (via GitHub)" <gi...@apache.org>.
saurabhd336 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1261988572
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4128,6 +4129,29 @@ public PeriodicTaskInvocationResponse invokeControllerPeriodicTask(String tableN
return new PeriodicTaskInvocationResponse(periodicTaskRequestId, messageCount > 0);
}
+ /**
+ * Construct a map of all the tenants and their respective minimum server requirements.
+ * The minimum server requirement is computed by iterating over all the tables of the tenant and
+ * find the table with maximum replication.
+ * @return map of tenants and their minimum server requirements
+ */
+ public Map<String, Integer> minimumServersRequiredForTenants() {
+ Map<String, Integer> tenantMinServerMap = new HashMap<>();
+ getAllServerTenantNames().forEach(tenant -> tenantMinServerMap.put(tenant, 0));
+ for (String table : getAllTables()) {
+ TableConfig tableConfig = getTableConfig(table);
Review Comment:
Could we fetch the tableConfigs all at once instead? We may pass list<TableConfigs> to this function
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] shounakmk219 commented on a diff in pull request #11077: Instance retag validation check api
Posted by "shounakmk219 (via GitHub)" <gi...@apache.org>.
shounakmk219 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1270527540
##########
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:
That's correct. Its mostly aimed to act as a prerequisite to the `/instances/{instanceName}/updateTags` endpoint so that user can validate the operation safety before performing the actual updateTags operation.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] Jackie-Jiang commented on pull request #11077: Instance retag validation check api
Posted by "Jackie-Jiang (via GitHub)" <gi...@apache.org>.
Jackie-Jiang commented on PR #11077:
URL: https://github.com/apache/pinot/pull/11077#issuecomment-1691105046
@shounakmk219 The new added test is flaky, can you please take a look?
```
Error: Failures:
Error: PinotInstanceRestletResourceTest.instanceRetagHappyPathTest:211->getCurrentInstanceTagsMap:282->getInstanceTags:291->ControllerTest.sendGetRequest:770->ControllerTest.sendGetRequest:780 » IO
Error: PinotInstanceRestletResourceTest.instanceRetagServerDeficiencyTest:237->getCurrentInstanceTagsMap:282->getInstanceTags:291->ControllerTest.sendGetRequest:770->ControllerTest.sendGetRequest:780 » IO
Error: PinotInstanceRestletResourceTest.testInstanceListingAndCreation:74->checkNumInstances:163 expected [8] but found [9]
```
https://github.com/apache/pinot/actions/runs/5959747075/job/16165968277
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] saurabhd336 commented on a diff in pull request #11077: (WIP) Instance retag validation check api
Posted by "saurabhd336 (via GitHub)" <gi...@apache.org>.
saurabhd336 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1263552337
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4128,6 +4138,33 @@ public PeriodicTaskInvocationResponse invokeControllerPeriodicTask(String tableN
return new PeriodicTaskInvocationResponse(periodicTaskRequestId, messageCount > 0);
}
+ /**
+ * Construct a map of all the tags and their respective minimum instance requirements.
+ * The minimum instance requirement is computed by
+ * - for BROKER tenant tag set it to 1.
+ * - for SERVER tenant tag iterate over all the tables of that tenant and find the maximum table replication.
+ * - for rest of the tags just set it to 0
+ * @return map of tags and their minimum instance requirements
+ */
+ public Map<String, Integer> minimumInstancesRequiredForTags() {
+ Map<String, Integer> tagMinServerMap = new HashMap<>();
+ for (InstanceConfig instanceConfig : getAllHelixInstanceConfigs()) {
+ for (String tag : instanceConfig.getTags()) {
+ tagMinServerMap.put(tag, TagNameUtils.isBrokerTag(tag) ? 1 : 0);
+ }
+ }
+ for (TableConfig tableConfig : getAllTableConfigs()) {
+ String serverTag = TagNameUtils.getServerTagForTenant(tableConfig.getTenantConfig().getServer(),
+ tableConfig.getTableType());
+ int maxReplication = Math.max(Objects.requireNonNullElse(tagMinServerMap.get(serverTag), 0),
+ tableConfig.getReplication());
+ tagMinServerMap.put(serverTag, maxReplication);
Review Comment:
Can be replaced with
`tagMinServerMap.put(serverTag, Math.max(tagMinServerMap.getOrDefault(serverTag, 0), tableConfig.getReplication()))`
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] shounakmk219 commented on a diff in pull request #11077: Instance retag validation check api
Posted by "shounakmk219 (via GitHub)" <gi...@apache.org>.
shounakmk219 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1270555663
##########
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:
Just the tags in request won't do, we also need to fetch the existing/old tags too for minInstances. We can pass a list of tags to get required minInstances but anyway we are forced to parse all the table configs hence didn’t complicate things for now. Let me know if accepting the tags list makes more sense and I will be happy to do it that way.
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] shounakmk219 commented on a diff in pull request #11077: (WIP) Instance retag validation check api
Posted by "shounakmk219 (via GitHub)" <gi...@apache.org>.
shounakmk219 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1265250074
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4128,6 +4138,33 @@ public PeriodicTaskInvocationResponse invokeControllerPeriodicTask(String tableN
return new PeriodicTaskInvocationResponse(periodicTaskRequestId, messageCount > 0);
}
+ /**
+ * Construct a map of all the tags and their respective minimum instance requirements.
+ * The minimum instance requirement is computed by
+ * - for BROKER tenant tag set it to 1.
+ * - for SERVER tenant tag iterate over all the tables of that tenant and find the maximum table replication.
+ * - for rest of the tags just set it to 0
+ * @return map of tags and their minimum instance requirements
+ */
+ public Map<String, Integer> minimumInstancesRequiredForTags() {
+ Map<String, Integer> tagMinServerMap = new HashMap<>();
+ for (InstanceConfig instanceConfig : getAllHelixInstanceConfigs()) {
+ for (String tag : instanceConfig.getTags()) {
+ tagMinServerMap.put(tag, TagNameUtils.isBrokerTag(tag) ? 1 : 0);
+ }
Review Comment:
Now that you point out I guess there's no harm to have 0 brokers as long as it hosts no tables. Fixed it.
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4128,6 +4138,33 @@ public PeriodicTaskInvocationResponse invokeControllerPeriodicTask(String tableN
return new PeriodicTaskInvocationResponse(periodicTaskRequestId, messageCount > 0);
}
+ /**
+ * Construct a map of all the tags and their respective minimum instance requirements.
+ * The minimum instance requirement is computed by
+ * - for BROKER tenant tag set it to 1.
+ * - for SERVER tenant tag iterate over all the tables of that tenant and find the maximum table replication.
+ * - for rest of the tags just set it to 0
+ * @return map of tags and their minimum instance requirements
+ */
+ public Map<String, Integer> minimumInstancesRequiredForTags() {
+ Map<String, Integer> tagMinServerMap = new HashMap<>();
+ for (InstanceConfig instanceConfig : getAllHelixInstanceConfigs()) {
+ for (String tag : instanceConfig.getTags()) {
+ tagMinServerMap.put(tag, TagNameUtils.isBrokerTag(tag) ? 1 : 0);
+ }
+ }
+ for (TableConfig tableConfig : getAllTableConfigs()) {
+ String serverTag = TagNameUtils.getServerTagForTenant(tableConfig.getTenantConfig().getServer(),
+ tableConfig.getTableType());
+ int maxReplication = Math.max(Objects.requireNonNullElse(tagMinServerMap.get(serverTag), 0),
+ tableConfig.getReplication());
+ tagMinServerMap.put(serverTag, maxReplication);
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] saurabhd336 commented on a diff in pull request #11077: (WIP) Instance retag validation check api
Posted by "saurabhd336 (via GitHub)" <gi...@apache.org>.
saurabhd336 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1263555663
##########
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:
`updatedTagInstanceMap.put(tag, updatedTagInstanceMap.getOrDefault(tag, 0) + 1)`
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] snleee merged pull request #11077: Instance retag validation check api
Posted by "snleee (via GitHub)" <gi...@apache.org>.
snleee merged PR #11077:
URL: https://github.com/apache/pinot/pull/11077
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] saurabhd336 commented on a diff in pull request #11077: (WIP) Instance retag validation check api
Posted by "saurabhd336 (via GitHub)" <gi...@apache.org>.
saurabhd336 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1263548505
##########
pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java:
##########
@@ -4128,6 +4138,33 @@ public PeriodicTaskInvocationResponse invokeControllerPeriodicTask(String tableN
return new PeriodicTaskInvocationResponse(periodicTaskRequestId, messageCount > 0);
}
+ /**
+ * Construct a map of all the tags and their respective minimum instance requirements.
+ * The minimum instance requirement is computed by
+ * - for BROKER tenant tag set it to 1.
+ * - for SERVER tenant tag iterate over all the tables of that tenant and find the maximum table replication.
+ * - for rest of the tags just set it to 0
+ * @return map of tags and their minimum instance requirements
+ */
+ public Map<String, Integer> minimumInstancesRequiredForTags() {
+ Map<String, Integer> tagMinServerMap = new HashMap<>();
+ for (InstanceConfig instanceConfig : getAllHelixInstanceConfigs()) {
+ for (String tag : instanceConfig.getTags()) {
+ tagMinServerMap.put(tag, TagNameUtils.isBrokerTag(tag) ? 1 : 0);
+ }
Review Comment:
Does this mean for broker tags, atleast 1 broker is always needed even if no table is tagged to it?
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] saurabhd336 commented on a diff in pull request #11077: (WIP) Instance retag validation check api
Posted by "saurabhd336 (via GitHub)" <gi...@apache.org>.
saurabhd336 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1263557236
##########
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:
same below
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] snleee commented on a diff in pull request #11077: Instance retag validation check api
Posted by "snleee (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [pinot] shounakmk219 commented on a diff in pull request #11077: Instance retag validation check api
Posted by "shounakmk219 (via GitHub)" <gi...@apache.org>.
shounakmk219 commented on code in PR #11077:
URL: https://github.com/apache/pinot/pull/11077#discussion_r1270636330
##########
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:
Added the docs and some comments, also made few changes to these utility methods to make the code flow more readable
--
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: commits-unsubscribe@pinot.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org