tpalfy commented on a change in pull request #4337: URL: https://github.com/apache/nifi/pull/4337#discussion_r440965363
########## File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorSummaryStatusEndpointMerger.java ########## @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.cluster.coordination.http.endpoints; + +import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; +import org.apache.nifi.cluster.manager.NodeResponse; +import org.apache.nifi.cluster.manager.PermissionsDtoMerger; +import org.apache.nifi.controller.status.RunStatus; +import org.apache.nifi.web.api.dto.ProcessorScheduleSummaryDTO; +import org.apache.nifi.web.api.entity.ProcessorScheduleSummariesEntity; +import org.apache.nifi.web.api.entity.ProcessorScheduleSummaryEntity; + +import java.net.URI; +import java.util.ArrayList; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +public class ProcessorSummaryStatusEndpointMerger implements EndpointResponseMerger { + public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/schedule-summaries/query"; + + @Override + public boolean canHandle(final URI uri, final String method) { + return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath()); + } + + @Override + public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) { + if (!canHandle(uri, method)) { + throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method); + } + + final ProcessorScheduleSummariesEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class); + + // Create mapping of Processor ID to its schedule Summary. + final Map<String, ProcessorScheduleSummaryEntity> scheduleSummaries = responseEntity.getScheduleSummaries().stream() + .collect(Collectors.toMap(entity -> entity.getScheduleSummary().getId(), entity -> entity)); + + for (final NodeResponse nodeResponse : successfulResponses) { + final ProcessorScheduleSummariesEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : + nodeResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class); + + for (final ProcessorScheduleSummaryEntity processorEntity : nodeResponseEntity.getScheduleSummaries()) { + final String processorId = processorEntity.getScheduleSummary().getId(); + + final ProcessorScheduleSummaryEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorScheduleSummaryEntity()); + merge(mergedEntity, processorEntity); + } + } + + final ProcessorScheduleSummariesEntity mergedEntity = new ProcessorScheduleSummariesEntity(); + mergedEntity.setScheduleSummaries(new ArrayList<>(scheduleSummaries.values())); + return new NodeResponse(clientResponse, mergedEntity); + } + + private void merge(final ProcessorScheduleSummaryEntity target, final ProcessorScheduleSummaryEntity additional) { + PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions()); + + final ProcessorScheduleSummaryDTO targetSummaryDto = target.getScheduleSummary(); + final ProcessorScheduleSummaryDTO additionalSummaryDto = additional.getScheduleSummary(); + + // If name is null, it's because of permissions, so we want to nullify it in the target. + if (additionalSummaryDto.getName() == null) { + targetSummaryDto.setName(null); + } + + targetSummaryDto.setActiveThreadCount(targetSummaryDto.getActiveThreadCount() + additionalSummaryDto.getActiveThreadCount()); + + final String additionalRunStatus = additionalSummaryDto.getRunStatus(); Review comment: To me it seems the `runStatus` is simply overwritten with the one coming from the last node we are processing (in case of `running`, `validating` and `invalid`). Is that intentional? What if the 1st node returns an `invalid` and the 2nd a `validating`? ########## File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java ########## @@ -3332,6 +3336,38 @@ private ProcessorEntity createProcessorEntity(final ProcessorNode processor, fin .collect(Collectors.toSet()); } + @Override + public ProcessorsRunStatusDetailsEntity getProcessorsRunStatusDetails(final Set<String> processorIds, final NiFiUser user) { + final List<ProcessorRunStatusDetailsEntity> summaryEntities = processorIds.stream() Review comment: Have some old `summary...` names. ########## File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorSummaryStatusEndpointMerger.java ########## @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.cluster.coordination.http.endpoints; + +import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; +import org.apache.nifi.cluster.manager.NodeResponse; +import org.apache.nifi.cluster.manager.PermissionsDtoMerger; +import org.apache.nifi.controller.status.RunStatus; +import org.apache.nifi.web.api.dto.ProcessorScheduleSummaryDTO; +import org.apache.nifi.web.api.entity.ProcessorScheduleSummariesEntity; +import org.apache.nifi.web.api.entity.ProcessorScheduleSummaryEntity; + +import java.net.URI; +import java.util.ArrayList; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +public class ProcessorSummaryStatusEndpointMerger implements EndpointResponseMerger { + public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/schedule-summaries/query"; + + @Override + public boolean canHandle(final URI uri, final String method) { + return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath()); + } + + @Override + public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) { + if (!canHandle(uri, method)) { + throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method); + } + + final ProcessorScheduleSummariesEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class); + + // Create mapping of Processor ID to its schedule Summary. + final Map<String, ProcessorScheduleSummaryEntity> scheduleSummaries = responseEntity.getScheduleSummaries().stream() Review comment: Minor: This map initialization suggests that we want to make sure the values in the map are the same objects that are coming with the clientResponse (maybe because of the `revision` which we do not merge). Is that true? Otherwise I could imagine something like this: ```java if (!canHandle(uri, method)) { throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method); } Collection<ProcessorScheduleSummaryEntity> mergedEntities = successfulResponses.stream() .map(nodeResponse -> nodeResponse.getClientResponse().readEntity(ProcessorScheduleSummariesEntity.class)) .flatMap(nodeResponseEntity -> nodeResponseEntity.getScheduleSummaries().stream()) .collect(Collectors.toMap(processorEntity -> processorEntity.getScheduleSummary().getId(), processorEntity -> processorEntity, this::merge)) .values(); final ProcessorScheduleSummariesEntity mergedEntity = new ProcessorScheduleSummariesEntity(); mergedEntity.setScheduleSummaries(new ArrayList<>(mergedEntities)); return new NodeResponse(clientResponse, mergedEntity); ``` ########## File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/util/ClusterReplicationComponentLifecycle.java ########## @@ -171,34 +172,37 @@ private boolean waitForProcessorValidation(final NiFiUser user, final URI origin URI groupUri; try { groupUri = new URI(originalUri.getScheme(), originalUri.getUserInfo(), originalUri.getHost(), - originalUri.getPort(), "/nifi-api/process-groups/" + groupId + "/processors", "includeDescendantGroups=true", originalUri.getFragment()); + originalUri.getPort(), "/nifi-api/processors/run-status-details/queries", null, originalUri.getFragment()); } catch (URISyntaxException e) { throw new RuntimeException(e); } final Map<String, String> headers = new HashMap<>(); - final MultivaluedMap<String, String> requestEntity = new MultivaluedHashMap<>(); + final RunStatusDetailsRequestEntity requestEntity = new RunStatusDetailsRequestEntity(); + final Set<String> processorIds = processors.values().stream() + .map(AffectedComponentEntity::getId) + .collect(Collectors.toSet()); + requestEntity.setProcessorIds(processorIds); boolean continuePolling = true; while (continuePolling) { // Determine whether we should replicate only to the cluster coordinator, or if we should replicate directly to the cluster nodes themselves. final NodeResponse clusterResponse; if (getReplicationTarget() == ReplicationTarget.CLUSTER_NODES) { - clusterResponse = getRequestReplicator().replicate(user, HttpMethod.GET, groupUri, requestEntity, headers).awaitMergedResponse(); + clusterResponse = getRequestReplicator().replicate(user, HttpMethod.POST, groupUri, requestEntity, headers).awaitMergedResponse(); } else { clusterResponse = getRequestReplicator().forwardToCoordinator( - getClusterCoordinatorNode(), user, HttpMethod.GET, groupUri, requestEntity, headers).awaitMergedResponse(); + getClusterCoordinatorNode(), user, HttpMethod.POST, groupUri, requestEntity, headers).awaitMergedResponse(); } if (clusterResponse.getStatus() != Status.OK.getStatusCode()) { return false; } - final ProcessorsEntity processorsEntity = getResponseEntity(clusterResponse, ProcessorsEntity.class); - final Set<ProcessorEntity> processorEntities = processorsEntity.getProcessors(); + final ProcessorsRunStatusDetailsEntity summariesEntity = getResponseEntity(clusterResponse, ProcessorsRunStatusDetailsEntity.class); Review comment: There are some `summariesEntity` old names. ########## File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java ########## @@ -3332,6 +3336,38 @@ private ProcessorEntity createProcessorEntity(final ProcessorNode processor, fin .collect(Collectors.toSet()); } + @Override + public ProcessorsRunStatusDetailsEntity getProcessorsRunStatusDetails(final Set<String> processorIds, final NiFiUser user) { + final List<ProcessorRunStatusDetailsEntity> summaryEntities = processorIds.stream() + .map(processorDAO::getProcessor) + .map(processor -> createRunStatusDetailsEntity(processor, user)) + .collect(Collectors.toList()); + + final ProcessorsRunStatusDetailsEntity summariesEntity = new ProcessorsRunStatusDetailsEntity(); + summariesEntity.setRunStatusDetails(summaryEntities); + return summariesEntity; + } + + private ProcessorRunStatusDetailsEntity createRunStatusDetailsEntity(final ProcessorNode processor, final NiFiUser user) { + final RevisionDTO revision = dtoFactory.createRevisionDTO(revisionManager.getRevision(processor.getIdentifier())); + final PermissionsDTO permissions = dtoFactory.createPermissionsDto(processor, user); + final ProcessorStatus processorStatus = controllerFacade.getProcessorStatus(processor.getIdentifier()); + final ProcessorRunStatusDetailsDTO runStatusDetailsDto = dtoFactory.createProcessorRunStatusDetailsDto(processor, processorStatus); + + if (!Boolean.TRUE.equals(permissions.getCanRead())) { Review comment: Minor: Not sure why `getCanRead()` has the return type `Boolean` when it's actually a getter for a `boolean` property. Simply doing ```suggestion if (permissions.getCanRead()) { ``` should be safe. ########## File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorRunStatusDetailsEndpointMerger.java ########## @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.cluster.coordination.http.endpoints; + +import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; +import org.apache.nifi.cluster.manager.NodeResponse; +import org.apache.nifi.cluster.manager.PermissionsDtoMerger; +import org.apache.nifi.controller.status.RunStatus; +import org.apache.nifi.web.api.dto.ProcessorRunStatusDetailsDTO; +import org.apache.nifi.web.api.entity.ProcessorsRunStatusDetailsEntity; +import org.apache.nifi.web.api.entity.ProcessorRunStatusDetailsEntity; + +import java.net.URI; +import java.util.ArrayList; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +public class ProcessorRunStatusDetailsEndpointMerger implements EndpointResponseMerger { + public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/run-status-details/queries"; + + @Override + public boolean canHandle(final URI uri, final String method) { + return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath()); + } + + @Override + public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) { + if (!canHandle(uri, method)) { + throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method); + } + + final ProcessorsRunStatusDetailsEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class); + + // Create mapping of Processor ID to its schedule Summary. + final Map<String, ProcessorRunStatusDetailsEntity> scheduleSummaries = responseEntity.getRunStatusDetails().stream() + .collect(Collectors.toMap(entity -> entity.getRunStatusDetails().getId(), entity -> entity)); + + for (final NodeResponse nodeResponse : successfulResponses) { + final ProcessorsRunStatusDetailsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : + nodeResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class); + + for (final ProcessorRunStatusDetailsEntity processorEntity : nodeResponseEntity.getRunStatusDetails()) { + final String processorId = processorEntity.getRunStatusDetails().getId(); + + final ProcessorRunStatusDetailsEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorRunStatusDetailsEntity()); + merge(mergedEntity, processorEntity); + } + } + + final ProcessorsRunStatusDetailsEntity mergedEntity = new ProcessorsRunStatusDetailsEntity(); + mergedEntity.setRunStatusDetails(new ArrayList<>(scheduleSummaries.values())); + return new NodeResponse(clientResponse, mergedEntity); + } + + private void merge(final ProcessorRunStatusDetailsEntity target, final ProcessorRunStatusDetailsEntity additional) { + PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions()); + + final ProcessorRunStatusDetailsDTO targetSummaryDto = target.getRunStatusDetails(); + final ProcessorRunStatusDetailsDTO additionalSummaryDto = additional.getRunStatusDetails(); + + // If name is null, it's because of permissions, so we want to nullify it in the target. + if (additionalSummaryDto.getName() == null) { Review comment: Wouldn't `!additional.getPermissions().getCanRead()` be a bit more descriptive? ########## File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-cluster/src/main/java/org/apache/nifi/cluster/coordination/http/endpoints/ProcessorRunStatusDetailsEndpointMerger.java ########## @@ -0,0 +1,100 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.nifi.cluster.coordination.http.endpoints; + +import org.apache.nifi.cluster.coordination.http.EndpointResponseMerger; +import org.apache.nifi.cluster.manager.NodeResponse; +import org.apache.nifi.cluster.manager.PermissionsDtoMerger; +import org.apache.nifi.controller.status.RunStatus; +import org.apache.nifi.web.api.dto.ProcessorRunStatusDetailsDTO; +import org.apache.nifi.web.api.entity.ProcessorsRunStatusDetailsEntity; +import org.apache.nifi.web.api.entity.ProcessorRunStatusDetailsEntity; + +import java.net.URI; +import java.util.ArrayList; +import java.util.Map; +import java.util.Set; +import java.util.stream.Collectors; + +public class ProcessorRunStatusDetailsEndpointMerger implements EndpointResponseMerger { + public static final String SCHEDULE_SUMMARY_URI = "/nifi-api/processors/run-status-details/queries"; + + @Override + public boolean canHandle(final URI uri, final String method) { + return "POST".equalsIgnoreCase(method) && SCHEDULE_SUMMARY_URI.equals(uri.getPath()); + } + + @Override + public NodeResponse merge(final URI uri, final String method, final Set<NodeResponse> successfulResponses, final Set<NodeResponse> problematicResponses, final NodeResponse clientResponse) { + if (!canHandle(uri, method)) { + throw new IllegalArgumentException("Cannot use Endpoint Mapper of type " + getClass().getSimpleName() + " to map responses for URI " + uri + ", HTTP Method " + method); + } + + final ProcessorsRunStatusDetailsEntity responseEntity = clientResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class); + + // Create mapping of Processor ID to its schedule Summary. + final Map<String, ProcessorRunStatusDetailsEntity> scheduleSummaries = responseEntity.getRunStatusDetails().stream() + .collect(Collectors.toMap(entity -> entity.getRunStatusDetails().getId(), entity -> entity)); + + for (final NodeResponse nodeResponse : successfulResponses) { + final ProcessorsRunStatusDetailsEntity nodeResponseEntity = nodeResponse == clientResponse ? responseEntity : + nodeResponse.getClientResponse().readEntity(ProcessorsRunStatusDetailsEntity.class); + + for (final ProcessorRunStatusDetailsEntity processorEntity : nodeResponseEntity.getRunStatusDetails()) { + final String processorId = processorEntity.getRunStatusDetails().getId(); + + final ProcessorRunStatusDetailsEntity mergedEntity = scheduleSummaries.computeIfAbsent(processorId, id -> new ProcessorRunStatusDetailsEntity()); + merge(mergedEntity, processorEntity); + } + } + + final ProcessorsRunStatusDetailsEntity mergedEntity = new ProcessorsRunStatusDetailsEntity(); + mergedEntity.setRunStatusDetails(new ArrayList<>(scheduleSummaries.values())); + return new NodeResponse(clientResponse, mergedEntity); + } + + private void merge(final ProcessorRunStatusDetailsEntity target, final ProcessorRunStatusDetailsEntity additional) { + PermissionsDtoMerger.mergePermissions(target.getPermissions(), additional.getPermissions()); + + final ProcessorRunStatusDetailsDTO targetSummaryDto = target.getRunStatusDetails(); + final ProcessorRunStatusDetailsDTO additionalSummaryDto = additional.getRunStatusDetails(); + + // If name is null, it's because of permissions, so we want to nullify it in the target. + if (additionalSummaryDto.getName() == null) { + targetSummaryDto.setName(null); + } + + targetSummaryDto.setActiveThreadCount(targetSummaryDto.getActiveThreadCount() + additionalSummaryDto.getActiveThreadCount()); + + final String additionalRunStatus = additionalSummaryDto.getRunStatus(); + if (RunStatus.Running.name().equals(additionalRunStatus)) { + targetSummaryDto.setRunStatus(RunStatus.Running.name()); + } else if (RunStatus.Validating.name().equals(additionalRunStatus)) { + targetSummaryDto.setRunStatus(RunStatus.Validating.name()); + } else if (RunStatus.Invalid.name().equals(additionalRunStatus)) { + targetSummaryDto.setRunStatus(RunStatus.Invalid.name()); + } + + // If validation errors is null, it's due to eprmissions, so we want to nullify it in the target. + if (additionalSummaryDto.getValidationErrors() == null) { Review comment: Typo: eprmissions Again, could use `!additional.getPermissions().getCanRead()` instead. We may be relying on a "secondary" truth when we have access to the "primary". Also to me it looks like the check could be incorrect in this way. Wouldn't the `validationErrors` be set to `null` if the processor is in `VALIDATING` according to this logic: ```java dto.setValidationErrors(convertValidationErrors(processor.getValidationErrors())); ... protected Collection<ValidationResult> getValidationErrors(final Set<ControllerServiceNode> servicesToIgnore) { final ValidationState validationState = this.validationState.get(); if (validationState.getStatus() == ValidationStatus.VALIDATING) { return null; } ... private Set<String> convertValidationErrors(final Collection<ValidationResult> validationErrors) { if (validationErrors == null) { return null; } ``` ---------------------------------------------------------------- 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. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
