mirror of https://github.com/apache/nifi.git
NIFI-4538 - Add Process Group information to...
...Search results * Separated the search functionality. * Added a unit test. * Added the PG info to UI (a mere draft). * Introduce the nearest versioned group * Removed the top level group results in favour of the nearest versioned group. * This closes #2364
This commit is contained in:
yuri1969
Matt Gilman
parent
9f95a10df9
commit
91e98aa50b
|
|
@ -29,6 +29,8 @@ public class ComponentSearchResultDTO {
|
|||
|
||||
private String id;
|
||||
private String groupId;
|
||||
private SearchResultGroupDTO parentGroup;
|
||||
private SearchResultGroupDTO versionedGroup;
|
||||
private String name;
|
||||
private List<String> matches;
|
||||
|
||||
|
|
@ -60,6 +62,34 @@ public class ComponentSearchResultDTO {
|
|||
this.groupId = groupId;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return parent group of the component that matched
|
||||
*/
|
||||
@ApiModelProperty(
|
||||
value = "The parent group of the component that matched the search."
|
||||
)
|
||||
public SearchResultGroupDTO getParentGroup() {
|
||||
return parentGroup;
|
||||
}
|
||||
|
||||
public void setParentGroup(final SearchResultGroupDTO parentGroup) {
|
||||
this.parentGroup = parentGroup;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the nearest versioned ancestor group of the component that matched
|
||||
*/
|
||||
@ApiModelProperty(
|
||||
value = "The nearest versioned ancestor group of the component that matched the search."
|
||||
)
|
||||
public SearchResultGroupDTO getVersionedGroup() {
|
||||
return versionedGroup;
|
||||
}
|
||||
|
||||
public void setVersionedGroup(final SearchResultGroupDTO versionedGroup) {
|
||||
this.versionedGroup = versionedGroup;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return name of the component that matched
|
||||
*/
|
||||
|
|
|
|||
|
|
@ -0,0 +1,59 @@
|
|||
/*
|
||||
* 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.web.api.dto.search;
|
||||
|
||||
import io.swagger.annotations.ApiModelProperty;
|
||||
|
||||
import javax.xml.bind.annotation.XmlType;
|
||||
|
||||
/**
|
||||
* The result's group level of a performed search.
|
||||
*/
|
||||
@XmlType(name = "searchResultGroup")
|
||||
public class SearchResultGroupDTO {
|
||||
private String id;
|
||||
private String name;
|
||||
|
||||
/**
|
||||
* @return id of this group
|
||||
*/
|
||||
@ApiModelProperty(
|
||||
value = "The id of the group.",
|
||||
required = true
|
||||
)
|
||||
public String getId() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public void setId(final String id) {
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return name of this group
|
||||
*/
|
||||
@ApiModelProperty(
|
||||
value = "The name of the group."
|
||||
)
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(final String name) {
|
||||
this.name = name;
|
||||
}
|
||||
}
|
||||
|
|
@ -33,10 +33,8 @@ import org.apache.nifi.bundle.Bundle;
|
|||
import org.apache.nifi.bundle.BundleCoordinate;
|
||||
import org.apache.nifi.cluster.protocol.NodeIdentifier;
|
||||
import org.apache.nifi.components.ConfigurableComponent;
|
||||
import org.apache.nifi.components.PropertyDescriptor;
|
||||
import org.apache.nifi.connectable.Connectable;
|
||||
import org.apache.nifi.connectable.Connection;
|
||||
import org.apache.nifi.connectable.Funnel;
|
||||
import org.apache.nifi.connectable.Port;
|
||||
import org.apache.nifi.controller.ContentAvailability;
|
||||
import org.apache.nifi.controller.ControllerService;
|
||||
|
|
@ -44,10 +42,8 @@ import org.apache.nifi.controller.Counter;
|
|||
import org.apache.nifi.controller.FlowController;
|
||||
import org.apache.nifi.controller.ProcessorNode;
|
||||
import org.apache.nifi.controller.ReportingTaskNode;
|
||||
import org.apache.nifi.controller.ScheduledState;
|
||||
import org.apache.nifi.controller.Template;
|
||||
import org.apache.nifi.controller.label.Label;
|
||||
import org.apache.nifi.controller.queue.FlowFileQueue;
|
||||
import org.apache.nifi.controller.queue.QueueSize;
|
||||
import org.apache.nifi.controller.repository.ContentNotFoundException;
|
||||
import org.apache.nifi.controller.repository.claim.ContentDirection;
|
||||
|
|
@ -66,8 +62,6 @@ import org.apache.nifi.groups.ProcessGroup;
|
|||
import org.apache.nifi.groups.ProcessGroupCounts;
|
||||
import org.apache.nifi.groups.RemoteProcessGroup;
|
||||
import org.apache.nifi.nar.ExtensionManager;
|
||||
import org.apache.nifi.nar.NarCloseable;
|
||||
import org.apache.nifi.processor.DataUnit;
|
||||
import org.apache.nifi.processor.Processor;
|
||||
import org.apache.nifi.processor.Relationship;
|
||||
import org.apache.nifi.provenance.ProvenanceEventRecord;
|
||||
|
|
@ -80,18 +74,11 @@ import org.apache.nifi.provenance.search.QuerySubmission;
|
|||
import org.apache.nifi.provenance.search.SearchTerm;
|
||||
import org.apache.nifi.provenance.search.SearchTerms;
|
||||
import org.apache.nifi.provenance.search.SearchableField;
|
||||
import org.apache.nifi.registry.ComponentVariableRegistry;
|
||||
import org.apache.nifi.registry.VariableDescriptor;
|
||||
import org.apache.nifi.registry.VariableRegistry;
|
||||
import org.apache.nifi.registry.flow.VersionedProcessGroup;
|
||||
import org.apache.nifi.remote.RemoteGroupPort;
|
||||
import org.apache.nifi.remote.RootGroupPort;
|
||||
import org.apache.nifi.reporting.ReportingTask;
|
||||
import org.apache.nifi.scheduling.ExecutionNode;
|
||||
import org.apache.nifi.scheduling.SchedulingStrategy;
|
||||
import org.apache.nifi.search.SearchContext;
|
||||
import org.apache.nifi.search.SearchResult;
|
||||
import org.apache.nifi.search.Searchable;
|
||||
import org.apache.nifi.services.FlowService;
|
||||
import org.apache.nifi.util.BundleUtils;
|
||||
import org.apache.nifi.util.FormatUtils;
|
||||
|
|
@ -112,7 +99,6 @@ import org.apache.nifi.web.api.dto.provenance.ProvenanceSearchableFieldDTO;
|
|||
import org.apache.nifi.web.api.dto.provenance.lineage.LineageDTO;
|
||||
import org.apache.nifi.web.api.dto.provenance.lineage.LineageRequestDTO;
|
||||
import org.apache.nifi.web.api.dto.provenance.lineage.LineageRequestDTO.LineageRequestType;
|
||||
import org.apache.nifi.web.api.dto.search.ComponentSearchResultDTO;
|
||||
import org.apache.nifi.web.api.dto.search.SearchResultsDTO;
|
||||
import org.apache.nifi.web.api.dto.status.ControllerStatusDTO;
|
||||
import org.apache.nifi.web.api.dto.status.StatusHistoryDTO;
|
||||
|
|
@ -156,6 +142,7 @@ public class ControllerFacade implements Authorizable {
|
|||
private NiFiProperties properties;
|
||||
private DtoFactory dtoFactory;
|
||||
private VariableRegistry variableRegistry;
|
||||
private ControllerSearchService controllerSearchService;
|
||||
|
||||
/**
|
||||
* Returns the group id that contains the specified processor.
|
||||
|
|
@ -1524,409 +1511,17 @@ public class ControllerFacade implements Authorizable {
|
|||
*/
|
||||
public SearchResultsDTO search(final String search) {
|
||||
final ProcessGroup rootGroup = flowController.getGroup(flowController.getRootGroupId());
|
||||
|
||||
final SearchResultsDTO results = new SearchResultsDTO();
|
||||
search(results, search, rootGroup);
|
||||
|
||||
controllerSearchService.search(results, search, rootGroup);
|
||||
|
||||
return results;
|
||||
}
|
||||
|
||||
private void search(final SearchResultsDTO results, final String search, final ProcessGroup group) {
|
||||
final NiFiUser user = NiFiUserUtils.getNiFiUser();
|
||||
|
||||
if (group.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO groupMatch = search(search, group);
|
||||
if (groupMatch != null) {
|
||||
results.getProcessGroupResults().add(groupMatch);
|
||||
}
|
||||
}
|
||||
|
||||
for (final ProcessorNode procNode : group.getProcessors()) {
|
||||
if (procNode.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO match = search(search, procNode);
|
||||
if (match != null) {
|
||||
match.setGroupId(group.getIdentifier());
|
||||
results.getProcessorResults().add(match);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (final Connection connection : group.getConnections()) {
|
||||
if (connection.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO match = search(search, connection);
|
||||
if (match != null) {
|
||||
match.setGroupId(group.getIdentifier());
|
||||
results.getConnectionResults().add(match);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (final RemoteProcessGroup remoteGroup : group.getRemoteProcessGroups()) {
|
||||
if (remoteGroup.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO match = search(search, remoteGroup);
|
||||
if (match != null) {
|
||||
match.setGroupId(group.getIdentifier());
|
||||
results.getRemoteProcessGroupResults().add(match);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (final Port port : group.getInputPorts()) {
|
||||
if (port.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO match = search(search, port);
|
||||
if (match != null) {
|
||||
match.setGroupId(group.getIdentifier());
|
||||
results.getInputPortResults().add(match);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (final Port port : group.getOutputPorts()) {
|
||||
if (port.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO match = search(search, port);
|
||||
if (match != null) {
|
||||
match.setGroupId(group.getIdentifier());
|
||||
results.getOutputPortResults().add(match);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (final Funnel funnel : group.getFunnels()) {
|
||||
if (funnel.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO match = search(search, funnel);
|
||||
if (match != null) {
|
||||
match.setGroupId(group.getIdentifier());
|
||||
results.getFunnelResults().add(match);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (final ProcessGroup processGroup : group.getProcessGroups()) {
|
||||
search(results, search, processGroup);
|
||||
}
|
||||
}
|
||||
|
||||
private ComponentSearchResultDTO search(final String searchStr, final Port port) {
|
||||
final List<String> matches = new ArrayList<>();
|
||||
|
||||
addIfAppropriate(searchStr, port.getIdentifier(), "Id", matches);
|
||||
addIfAppropriate(searchStr, port.getVersionedComponentId().orElse(null), "Version Control ID", matches);
|
||||
addIfAppropriate(searchStr, port.getName(), "Name", matches);
|
||||
addIfAppropriate(searchStr, port.getComments(), "Comments", matches);
|
||||
|
||||
// consider scheduled state
|
||||
if (ScheduledState.DISABLED.equals(port.getScheduledState())) {
|
||||
if (StringUtils.containsIgnoreCase("disabled", searchStr)) {
|
||||
matches.add("Run status: Disabled");
|
||||
}
|
||||
} else {
|
||||
if (StringUtils.containsIgnoreCase("invalid", searchStr) && !port.isValid()) {
|
||||
matches.add("Run status: Invalid");
|
||||
} else if (ScheduledState.RUNNING.equals(port.getScheduledState()) && StringUtils.containsIgnoreCase("running", searchStr)) {
|
||||
matches.add("Run status: Running");
|
||||
} else if (ScheduledState.STOPPED.equals(port.getScheduledState()) && StringUtils.containsIgnoreCase("stopped", searchStr)) {
|
||||
matches.add("Run status: Stopped");
|
||||
}
|
||||
}
|
||||
|
||||
if (port instanceof RootGroupPort) {
|
||||
final RootGroupPort rootGroupPort = (RootGroupPort) port;
|
||||
|
||||
// user access controls
|
||||
for (final String userAccessControl : rootGroupPort.getUserAccessControl()) {
|
||||
addIfAppropriate(searchStr, userAccessControl, "User access control", matches);
|
||||
}
|
||||
|
||||
// group access controls
|
||||
for (final String groupAccessControl : rootGroupPort.getGroupAccessControl()) {
|
||||
addIfAppropriate(searchStr, groupAccessControl, "Group access control", matches);
|
||||
}
|
||||
}
|
||||
|
||||
if (matches.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final ComponentSearchResultDTO dto = new ComponentSearchResultDTO();
|
||||
dto.setId(port.getIdentifier());
|
||||
dto.setName(port.getName());
|
||||
dto.setMatches(matches);
|
||||
return dto;
|
||||
}
|
||||
|
||||
public void verifyComponentTypes(VersionedProcessGroup versionedFlow) {
|
||||
flowController.verifyComponentTypesInSnippet(versionedFlow);
|
||||
}
|
||||
|
||||
|
||||
private ComponentSearchResultDTO search(final String searchStr, final ProcessorNode procNode) {
|
||||
final List<String> matches = new ArrayList<>();
|
||||
final Processor processor = procNode.getProcessor();
|
||||
|
||||
addIfAppropriate(searchStr, procNode.getIdentifier(), "Id", matches);
|
||||
addIfAppropriate(searchStr, procNode.getVersionedComponentId().orElse(null), "Version Control ID", matches);
|
||||
addIfAppropriate(searchStr, procNode.getName(), "Name", matches);
|
||||
addIfAppropriate(searchStr, procNode.getComments(), "Comments", matches);
|
||||
|
||||
// consider scheduling strategy
|
||||
if (SchedulingStrategy.EVENT_DRIVEN.equals(procNode.getSchedulingStrategy()) && StringUtils.containsIgnoreCase("event", searchStr)) {
|
||||
matches.add("Scheduling strategy: Event driven");
|
||||
} else if (SchedulingStrategy.TIMER_DRIVEN.equals(procNode.getSchedulingStrategy()) && StringUtils.containsIgnoreCase("timer", searchStr)) {
|
||||
matches.add("Scheduling strategy: Timer driven");
|
||||
} else if (SchedulingStrategy.PRIMARY_NODE_ONLY.equals(procNode.getSchedulingStrategy()) && StringUtils.containsIgnoreCase("primary", searchStr)) {
|
||||
// PRIMARY_NODE_ONLY has been deprecated as a SchedulingStrategy and replaced by PRIMARY as an ExecutionNode.
|
||||
matches.add("Scheduling strategy: On primary node");
|
||||
}
|
||||
|
||||
// consider execution node
|
||||
if (ExecutionNode.PRIMARY.equals(procNode.getExecutionNode()) && StringUtils.containsIgnoreCase("primary", searchStr)) {
|
||||
matches.add("Execution node: primary");
|
||||
}
|
||||
|
||||
// consider scheduled state
|
||||
if (ScheduledState.DISABLED.equals(procNode.getScheduledState())) {
|
||||
if (StringUtils.containsIgnoreCase("disabled", searchStr)) {
|
||||
matches.add("Run status: Disabled");
|
||||
}
|
||||
} else {
|
||||
if (StringUtils.containsIgnoreCase("invalid", searchStr) && !procNode.isValid()) {
|
||||
matches.add("Run status: Invalid");
|
||||
} else if (ScheduledState.RUNNING.equals(procNode.getScheduledState()) && StringUtils.containsIgnoreCase("running", searchStr)) {
|
||||
matches.add("Run status: Running");
|
||||
} else if (ScheduledState.STOPPED.equals(procNode.getScheduledState()) && StringUtils.containsIgnoreCase("stopped", searchStr)) {
|
||||
matches.add("Run status: Stopped");
|
||||
}
|
||||
}
|
||||
|
||||
for (final Relationship relationship : procNode.getRelationships()) {
|
||||
addIfAppropriate(searchStr, relationship.getName(), "Relationship", matches);
|
||||
}
|
||||
|
||||
// Add both the actual class name and the component type. This allows us to search for 'Ghost'
|
||||
// to search for components that could not be instantiated.
|
||||
addIfAppropriate(searchStr, processor.getClass().getSimpleName(), "Type", matches);
|
||||
addIfAppropriate(searchStr, procNode.getComponentType(), "Type", matches);
|
||||
|
||||
for (final Map.Entry<PropertyDescriptor, String> entry : procNode.getProperties().entrySet()) {
|
||||
final PropertyDescriptor descriptor = entry.getKey();
|
||||
|
||||
addIfAppropriate(searchStr, descriptor.getName(), "Property name", matches);
|
||||
addIfAppropriate(searchStr, descriptor.getDescription(), "Property description", matches);
|
||||
|
||||
// never include sensitive properties values in search results
|
||||
if (descriptor.isSensitive()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
String value = entry.getValue();
|
||||
|
||||
// if unset consider default value
|
||||
if (value == null) {
|
||||
value = descriptor.getDefaultValue();
|
||||
}
|
||||
|
||||
// evaluate if the value matches the search criteria
|
||||
if (StringUtils.containsIgnoreCase(value, searchStr)) {
|
||||
matches.add("Property value: " + descriptor.getName() + " - " + value);
|
||||
}
|
||||
}
|
||||
|
||||
// consider searching the processor directly
|
||||
if (processor instanceof Searchable) {
|
||||
final Searchable searchable = (Searchable) processor;
|
||||
|
||||
final SearchContext context = new StandardSearchContext(searchStr, procNode, flowController, variableRegistry);
|
||||
|
||||
// search the processor using the appropriate thread context classloader
|
||||
try (final NarCloseable x = NarCloseable.withComponentNarLoader(processor.getClass(), processor.getIdentifier())) {
|
||||
final Collection<SearchResult> searchResults = searchable.search(context);
|
||||
if (CollectionUtils.isNotEmpty(searchResults)) {
|
||||
for (final SearchResult searchResult : searchResults) {
|
||||
matches.add(searchResult.getLabel() + ": " + searchResult.getMatch());
|
||||
}
|
||||
}
|
||||
} catch (final Throwable t) {
|
||||
// log this as error
|
||||
}
|
||||
}
|
||||
|
||||
if (matches.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final ComponentSearchResultDTO result = new ComponentSearchResultDTO();
|
||||
result.setId(procNode.getIdentifier());
|
||||
result.setMatches(matches);
|
||||
result.setName(procNode.getName());
|
||||
return result;
|
||||
}
|
||||
|
||||
private ComponentSearchResultDTO search(final String searchStr, final ProcessGroup group) {
|
||||
final List<String> matches = new ArrayList<>();
|
||||
final ProcessGroup parent = group.getParent();
|
||||
if (parent == null) {
|
||||
return null;
|
||||
}
|
||||
|
||||
addIfAppropriate(searchStr, group.getIdentifier(), "Id", matches);
|
||||
addIfAppropriate(searchStr, group.getVersionedComponentId().orElse(null), "Version Control ID", matches);
|
||||
addIfAppropriate(searchStr, group.getName(), "Name", matches);
|
||||
addIfAppropriate(searchStr, group.getComments(), "Comments", matches);
|
||||
|
||||
final ComponentVariableRegistry varRegistry = group.getVariableRegistry();
|
||||
if (varRegistry != null) {
|
||||
final Map<VariableDescriptor, String> variableMap = varRegistry.getVariableMap();
|
||||
for (final Map.Entry<VariableDescriptor, String> entry : variableMap.entrySet()) {
|
||||
addIfAppropriate(searchStr, entry.getKey().getName(), "Variable Name", matches);
|
||||
addIfAppropriate(searchStr, entry.getValue(), "Variable Value", matches);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
if (matches.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final ComponentSearchResultDTO result = new ComponentSearchResultDTO();
|
||||
result.setId(group.getIdentifier());
|
||||
result.setName(group.getName());
|
||||
result.setGroupId(parent.getIdentifier());
|
||||
result.setMatches(matches);
|
||||
return result;
|
||||
}
|
||||
|
||||
private ComponentSearchResultDTO search(final String searchStr, final Connection connection) {
|
||||
final List<String> matches = new ArrayList<>();
|
||||
|
||||
// search id and name
|
||||
addIfAppropriate(searchStr, connection.getIdentifier(), "Id", matches);
|
||||
addIfAppropriate(searchStr, connection.getVersionedComponentId().orElse(null), "Version Control ID", matches);
|
||||
addIfAppropriate(searchStr, connection.getName(), "Name", matches);
|
||||
|
||||
// search relationships
|
||||
for (final Relationship relationship : connection.getRelationships()) {
|
||||
addIfAppropriate(searchStr, relationship.getName(), "Relationship", matches);
|
||||
}
|
||||
|
||||
// search prioritizers
|
||||
final FlowFileQueue queue = connection.getFlowFileQueue();
|
||||
for (final FlowFilePrioritizer comparator : queue.getPriorities()) {
|
||||
addIfAppropriate(searchStr, comparator.getClass().getName(), "Prioritizer", matches);
|
||||
}
|
||||
|
||||
// search expiration
|
||||
if (StringUtils.containsIgnoreCase("expires", searchStr) || StringUtils.containsIgnoreCase("expiration", searchStr)) {
|
||||
final int expirationMillis = connection.getFlowFileQueue().getFlowFileExpiration(TimeUnit.MILLISECONDS);
|
||||
if (expirationMillis > 0) {
|
||||
matches.add("FlowFile expiration: " + connection.getFlowFileQueue().getFlowFileExpiration());
|
||||
}
|
||||
}
|
||||
|
||||
// search back pressure
|
||||
if (StringUtils.containsIgnoreCase("back pressure", searchStr) || StringUtils.containsIgnoreCase("pressure", searchStr)) {
|
||||
final String backPressureDataSize = connection.getFlowFileQueue().getBackPressureDataSizeThreshold();
|
||||
final Double backPressureBytes = DataUnit.parseDataSize(backPressureDataSize, DataUnit.B);
|
||||
if (backPressureBytes > 0) {
|
||||
matches.add("Back pressure data size: " + backPressureDataSize);
|
||||
}
|
||||
|
||||
final long backPressureCount = connection.getFlowFileQueue().getBackPressureObjectThreshold();
|
||||
if (backPressureCount > 0) {
|
||||
matches.add("Back pressure count: " + backPressureCount);
|
||||
}
|
||||
}
|
||||
|
||||
// search the source
|
||||
final Connectable source = connection.getSource();
|
||||
addIfAppropriate(searchStr, source.getIdentifier(), "Source id", matches);
|
||||
addIfAppropriate(searchStr, source.getName(), "Source name", matches);
|
||||
addIfAppropriate(searchStr, source.getComments(), "Source comments", matches);
|
||||
|
||||
// search the destination
|
||||
final Connectable destination = connection.getDestination();
|
||||
addIfAppropriate(searchStr, destination.getIdentifier(), "Destination id", matches);
|
||||
addIfAppropriate(searchStr, destination.getName(), "Destination name", matches);
|
||||
addIfAppropriate(searchStr, destination.getComments(), "Destination comments", matches);
|
||||
|
||||
if (matches.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final ComponentSearchResultDTO result = new ComponentSearchResultDTO();
|
||||
result.setId(connection.getIdentifier());
|
||||
|
||||
// determine the name of the search match
|
||||
if (StringUtils.isNotBlank(connection.getName())) {
|
||||
result.setName(connection.getName());
|
||||
} else if (!connection.getRelationships().isEmpty()) {
|
||||
final List<String> relationships = new ArrayList<>(connection.getRelationships().size());
|
||||
for (final Relationship relationship : connection.getRelationships()) {
|
||||
if (StringUtils.isNotBlank(relationship.getName())) {
|
||||
relationships.add(relationship.getName());
|
||||
}
|
||||
}
|
||||
if (!relationships.isEmpty()) {
|
||||
result.setName(StringUtils.join(relationships, ", "));
|
||||
}
|
||||
}
|
||||
|
||||
// ensure a name is added
|
||||
if (result.getName() == null) {
|
||||
result.setName("From source " + connection.getSource().getName());
|
||||
}
|
||||
|
||||
result.setMatches(matches);
|
||||
return result;
|
||||
}
|
||||
|
||||
private ComponentSearchResultDTO search(final String searchStr, final RemoteProcessGroup group) {
|
||||
final List<String> matches = new ArrayList<>();
|
||||
addIfAppropriate(searchStr, group.getIdentifier(), "Id", matches);
|
||||
addIfAppropriate(searchStr, group.getVersionedComponentId().orElse(null), "Version Control ID", matches);
|
||||
addIfAppropriate(searchStr, group.getName(), "Name", matches);
|
||||
addIfAppropriate(searchStr, group.getComments(), "Comments", matches);
|
||||
addIfAppropriate(searchStr, group.getTargetUris(), "URLs", matches);
|
||||
|
||||
// consider the transmission status
|
||||
if ((StringUtils.containsIgnoreCase("transmitting", searchStr) || StringUtils.containsIgnoreCase("transmission enabled", searchStr)) && group.isTransmitting()) {
|
||||
matches.add("Transmission: On");
|
||||
} else if ((StringUtils.containsIgnoreCase("not transmitting", searchStr) || StringUtils.containsIgnoreCase("transmission disabled", searchStr)) && !group.isTransmitting()) {
|
||||
matches.add("Transmission: Off");
|
||||
}
|
||||
|
||||
if (matches.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final ComponentSearchResultDTO result = new ComponentSearchResultDTO();
|
||||
result.setId(group.getIdentifier());
|
||||
result.setName(group.getName());
|
||||
result.setMatches(matches);
|
||||
return result;
|
||||
}
|
||||
|
||||
private ComponentSearchResultDTO search(final String searchStr, final Funnel funnel) {
|
||||
final List<String> matches = new ArrayList<>();
|
||||
addIfAppropriate(searchStr, funnel.getIdentifier(), "Id", matches);
|
||||
addIfAppropriate(searchStr, funnel.getVersionedComponentId().orElse(null), "Version Control ID", matches);
|
||||
|
||||
if (matches.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final ComponentSearchResultDTO dto = new ComponentSearchResultDTO();
|
||||
dto.setId(funnel.getIdentifier());
|
||||
dto.setName(funnel.getName());
|
||||
dto.setMatches(matches);
|
||||
return dto;
|
||||
}
|
||||
|
||||
private void addIfAppropriate(final String searchStr, final String value, final String label, final List<String> matches) {
|
||||
if (StringUtils.containsIgnoreCase(value, searchStr)) {
|
||||
matches.add(label + ": " + value);
|
||||
}
|
||||
}
|
||||
|
||||
/*
|
||||
* setters
|
||||
*/
|
||||
|
|
@ -1953,4 +1548,8 @@ public class ControllerFacade implements Authorizable {
|
|||
public void setVariableRegistry(VariableRegistry variableRegistry) {
|
||||
this.variableRegistry = variableRegistry;
|
||||
}
|
||||
|
||||
public void setControllerSearchService(ControllerSearchService controllerSearchService) {
|
||||
this.controllerSearchService = controllerSearchService;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,543 @@
|
|||
/*
|
||||
* 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.web.controller;
|
||||
|
||||
import org.apache.commons.collections4.CollectionUtils;
|
||||
import org.apache.commons.lang3.StringUtils;
|
||||
import org.apache.nifi.authorization.Authorizer;
|
||||
import org.apache.nifi.authorization.RequestAction;
|
||||
import org.apache.nifi.authorization.user.NiFiUser;
|
||||
import org.apache.nifi.authorization.user.NiFiUserUtils;
|
||||
import org.apache.nifi.components.PropertyDescriptor;
|
||||
import org.apache.nifi.connectable.Connectable;
|
||||
import org.apache.nifi.connectable.Connection;
|
||||
import org.apache.nifi.connectable.Funnel;
|
||||
import org.apache.nifi.connectable.Port;
|
||||
import org.apache.nifi.controller.FlowController;
|
||||
import org.apache.nifi.controller.ProcessorNode;
|
||||
import org.apache.nifi.controller.ScheduledState;
|
||||
import org.apache.nifi.controller.queue.FlowFileQueue;
|
||||
import org.apache.nifi.flowfile.FlowFilePrioritizer;
|
||||
import org.apache.nifi.groups.ProcessGroup;
|
||||
import org.apache.nifi.groups.RemoteProcessGroup;
|
||||
import org.apache.nifi.nar.NarCloseable;
|
||||
import org.apache.nifi.processor.DataUnit;
|
||||
import org.apache.nifi.processor.Processor;
|
||||
import org.apache.nifi.processor.Relationship;
|
||||
import org.apache.nifi.registry.ComponentVariableRegistry;
|
||||
import org.apache.nifi.registry.VariableDescriptor;
|
||||
import org.apache.nifi.registry.VariableRegistry;
|
||||
import org.apache.nifi.remote.RootGroupPort;
|
||||
import org.apache.nifi.scheduling.ExecutionNode;
|
||||
import org.apache.nifi.scheduling.SchedulingStrategy;
|
||||
import org.apache.nifi.search.SearchContext;
|
||||
import org.apache.nifi.search.SearchResult;
|
||||
import org.apache.nifi.search.Searchable;
|
||||
import org.apache.nifi.web.api.dto.search.ComponentSearchResultDTO;
|
||||
import org.apache.nifi.web.api.dto.search.SearchResultGroupDTO;
|
||||
import org.apache.nifi.web.api.dto.search.SearchResultsDTO;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* NiFi web controller's helper service that implements component search.
|
||||
*/
|
||||
public class ControllerSearchService {
|
||||
private FlowController flowController;
|
||||
private Authorizer authorizer;
|
||||
private VariableRegistry variableRegistry;
|
||||
|
||||
/**
|
||||
* Searches term in the controller beginning from a given process group.
|
||||
*
|
||||
* @param results Search results
|
||||
* @param search The search term
|
||||
* @param group The init process group
|
||||
*/
|
||||
public void search(final SearchResultsDTO results, final String search, final ProcessGroup group) {
|
||||
final NiFiUser user = NiFiUserUtils.getNiFiUser();
|
||||
|
||||
if (group.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO groupMatch = search(search, group);
|
||||
if (groupMatch != null) {
|
||||
// get the parent group, not the current one
|
||||
groupMatch.setParentGroup(buildResultGroup(group.getParent(), user));
|
||||
groupMatch.setVersionedGroup(buildVersionedGroup(group.getParent(), user));
|
||||
results.getProcessGroupResults().add(groupMatch);
|
||||
}
|
||||
}
|
||||
|
||||
for (final ProcessorNode procNode : group.getProcessors()) {
|
||||
if (procNode.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO match = search(search, procNode);
|
||||
if (match != null) {
|
||||
match.setGroupId(group.getIdentifier());
|
||||
match.setParentGroup(buildResultGroup(group, user));
|
||||
match.setVersionedGroup(buildVersionedGroup(group, user));
|
||||
results.getProcessorResults().add(match);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (final Connection connection : group.getConnections()) {
|
||||
if (connection.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO match = search(search, connection);
|
||||
if (match != null) {
|
||||
match.setGroupId(group.getIdentifier());
|
||||
match.setParentGroup(buildResultGroup(group, user));
|
||||
match.setVersionedGroup(buildVersionedGroup(group, user));
|
||||
results.getConnectionResults().add(match);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (final RemoteProcessGroup remoteGroup : group.getRemoteProcessGroups()) {
|
||||
if (remoteGroup.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO match = search(search, remoteGroup);
|
||||
if (match != null) {
|
||||
match.setGroupId(group.getIdentifier());
|
||||
match.setParentGroup(buildResultGroup(group, user));
|
||||
match.setVersionedGroup(buildVersionedGroup(group, user));
|
||||
results.getRemoteProcessGroupResults().add(match);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (final Port port : group.getInputPorts()) {
|
||||
if (port.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO match = search(search, port);
|
||||
if (match != null) {
|
||||
match.setGroupId(group.getIdentifier());
|
||||
match.setParentGroup(buildResultGroup(group, user));
|
||||
match.setVersionedGroup(buildVersionedGroup(group, user));
|
||||
results.getInputPortResults().add(match);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (final Port port : group.getOutputPorts()) {
|
||||
if (port.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO match = search(search, port);
|
||||
if (match != null) {
|
||||
match.setGroupId(group.getIdentifier());
|
||||
match.setParentGroup(buildResultGroup(group, user));
|
||||
match.setVersionedGroup(buildVersionedGroup(group, user));
|
||||
results.getOutputPortResults().add(match);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (final Funnel funnel : group.getFunnels()) {
|
||||
if (funnel.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
final ComponentSearchResultDTO match = search(search, funnel);
|
||||
if (match != null) {
|
||||
match.setGroupId(group.getIdentifier());
|
||||
match.setParentGroup(buildResultGroup(group, user));
|
||||
match.setVersionedGroup(buildVersionedGroup(group, user));
|
||||
results.getFunnelResults().add(match);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for (final ProcessGroup processGroup : group.getProcessGroups()) {
|
||||
search(results, search, processGroup);
|
||||
}
|
||||
}
|
||||
|
||||
private ComponentSearchResultDTO search(final String searchStr, final Port port) {
|
||||
final List<String> matches = new ArrayList<>();
|
||||
|
||||
addIfAppropriate(searchStr, port.getIdentifier(), "Id", matches);
|
||||
addIfAppropriate(searchStr, port.getVersionedComponentId().orElse(null), "Version Control ID", matches);
|
||||
addIfAppropriate(searchStr, port.getName(), "Name", matches);
|
||||
addIfAppropriate(searchStr, port.getComments(), "Comments", matches);
|
||||
|
||||
// consider scheduled state
|
||||
if (ScheduledState.DISABLED.equals(port.getScheduledState())) {
|
||||
if (StringUtils.containsIgnoreCase("disabled", searchStr)) {
|
||||
matches.add("Run status: Disabled");
|
||||
}
|
||||
} else {
|
||||
if (StringUtils.containsIgnoreCase("invalid", searchStr) && !port.isValid()) {
|
||||
matches.add("Run status: Invalid");
|
||||
} else if (ScheduledState.RUNNING.equals(port.getScheduledState()) && StringUtils.containsIgnoreCase("running", searchStr)) {
|
||||
matches.add("Run status: Running");
|
||||
} else if (ScheduledState.STOPPED.equals(port.getScheduledState()) && StringUtils.containsIgnoreCase("stopped", searchStr)) {
|
||||
matches.add("Run status: Stopped");
|
||||
}
|
||||
}
|
||||
|
||||
if (port instanceof RootGroupPort) {
|
||||
final RootGroupPort rootGroupPort = (RootGroupPort) port;
|
||||
|
||||
// user access controls
|
||||
for (final String userAccessControl : rootGroupPort.getUserAccessControl()) {
|
||||
addIfAppropriate(searchStr, userAccessControl, "User access control", matches);
|
||||
}
|
||||
|
||||
// group access controls
|
||||
for (final String groupAccessControl : rootGroupPort.getGroupAccessControl()) {
|
||||
addIfAppropriate(searchStr, groupAccessControl, "Group access control", matches);
|
||||
}
|
||||
}
|
||||
|
||||
if (matches.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final ComponentSearchResultDTO dto = new ComponentSearchResultDTO();
|
||||
dto.setId(port.getIdentifier());
|
||||
dto.setName(port.getName());
|
||||
dto.setMatches(matches);
|
||||
return dto;
|
||||
}
|
||||
|
||||
private ComponentSearchResultDTO search(final String searchStr, final ProcessorNode procNode) {
|
||||
final List<String> matches = new ArrayList<>();
|
||||
final Processor processor = procNode.getProcessor();
|
||||
|
||||
addIfAppropriate(searchStr, procNode.getIdentifier(), "Id", matches);
|
||||
addIfAppropriate(searchStr, procNode.getVersionedComponentId().orElse(null), "Version Control ID", matches);
|
||||
addIfAppropriate(searchStr, procNode.getName(), "Name", matches);
|
||||
addIfAppropriate(searchStr, procNode.getComments(), "Comments", matches);
|
||||
|
||||
// consider scheduling strategy
|
||||
if (SchedulingStrategy.EVENT_DRIVEN.equals(procNode.getSchedulingStrategy()) && StringUtils.containsIgnoreCase("event", searchStr)) {
|
||||
matches.add("Scheduling strategy: Event driven");
|
||||
} else if (SchedulingStrategy.TIMER_DRIVEN.equals(procNode.getSchedulingStrategy()) && StringUtils.containsIgnoreCase("timer", searchStr)) {
|
||||
matches.add("Scheduling strategy: Timer driven");
|
||||
} else if (SchedulingStrategy.PRIMARY_NODE_ONLY.equals(procNode.getSchedulingStrategy()) && StringUtils.containsIgnoreCase("primary", searchStr)) {
|
||||
// PRIMARY_NODE_ONLY has been deprecated as a SchedulingStrategy and replaced by PRIMARY as an ExecutionNode.
|
||||
matches.add("Scheduling strategy: On primary node");
|
||||
}
|
||||
|
||||
// consider execution node
|
||||
if (ExecutionNode.PRIMARY.equals(procNode.getExecutionNode()) && StringUtils.containsIgnoreCase("primary", searchStr)) {
|
||||
matches.add("Execution node: primary");
|
||||
}
|
||||
|
||||
// consider scheduled state
|
||||
if (ScheduledState.DISABLED.equals(procNode.getScheduledState())) {
|
||||
if (StringUtils.containsIgnoreCase("disabled", searchStr)) {
|
||||
matches.add("Run status: Disabled");
|
||||
}
|
||||
} else {
|
||||
if (StringUtils.containsIgnoreCase("invalid", searchStr) && !procNode.isValid()) {
|
||||
matches.add("Run status: Invalid");
|
||||
} else if (ScheduledState.RUNNING.equals(procNode.getScheduledState()) && StringUtils.containsIgnoreCase("running", searchStr)) {
|
||||
matches.add("Run status: Running");
|
||||
} else if (ScheduledState.STOPPED.equals(procNode.getScheduledState()) && StringUtils.containsIgnoreCase("stopped", searchStr)) {
|
||||
matches.add("Run status: Stopped");
|
||||
}
|
||||
}
|
||||
|
||||
for (final Relationship relationship : procNode.getRelationships()) {
|
||||
addIfAppropriate(searchStr, relationship.getName(), "Relationship", matches);
|
||||
}
|
||||
|
||||
// Add both the actual class name and the component type. This allows us to search for 'Ghost'
|
||||
// to search for components that could not be instantiated.
|
||||
addIfAppropriate(searchStr, processor.getClass().getSimpleName(), "Type", matches);
|
||||
addIfAppropriate(searchStr, procNode.getComponentType(), "Type", matches);
|
||||
|
||||
for (final Map.Entry<PropertyDescriptor, String> entry : procNode.getProperties().entrySet()) {
|
||||
final PropertyDescriptor descriptor = entry.getKey();
|
||||
|
||||
addIfAppropriate(searchStr, descriptor.getName(), "Property name", matches);
|
||||
addIfAppropriate(searchStr, descriptor.getDescription(), "Property description", matches);
|
||||
|
||||
// never include sensitive properties values in search results
|
||||
if (descriptor.isSensitive()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
String value = entry.getValue();
|
||||
|
||||
// if unset consider default value
|
||||
if (value == null) {
|
||||
value = descriptor.getDefaultValue();
|
||||
}
|
||||
|
||||
// evaluate if the value matches the search criteria
|
||||
if (StringUtils.containsIgnoreCase(value, searchStr)) {
|
||||
matches.add("Property value: " + descriptor.getName() + " - " + value);
|
||||
}
|
||||
}
|
||||
|
||||
// consider searching the processor directly
|
||||
if (processor instanceof Searchable) {
|
||||
final Searchable searchable = (Searchable) processor;
|
||||
|
||||
final SearchContext context = new StandardSearchContext(searchStr, procNode, flowController, variableRegistry);
|
||||
|
||||
// search the processor using the appropriate thread context classloader
|
||||
try (final NarCloseable x = NarCloseable.withComponentNarLoader(processor.getClass(), processor.getIdentifier())) {
|
||||
final Collection<SearchResult> searchResults = searchable.search(context);
|
||||
if (CollectionUtils.isNotEmpty(searchResults)) {
|
||||
for (final SearchResult searchResult : searchResults) {
|
||||
matches.add(searchResult.getLabel() + ": " + searchResult.getMatch());
|
||||
}
|
||||
}
|
||||
} catch (final Throwable t) {
|
||||
// log this as error
|
||||
}
|
||||
}
|
||||
|
||||
if (matches.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final ComponentSearchResultDTO result = new ComponentSearchResultDTO();
|
||||
result.setId(procNode.getIdentifier());
|
||||
result.setMatches(matches);
|
||||
result.setName(procNode.getName());
|
||||
return result;
|
||||
}
|
||||
|
||||
private ComponentSearchResultDTO search(final String searchStr, final ProcessGroup group) {
|
||||
final List<String> matches = new ArrayList<>();
|
||||
final ProcessGroup parent = group.getParent();
|
||||
if (parent == null) {
|
||||
return null;
|
||||
}
|
||||
|
||||
addIfAppropriate(searchStr, group.getIdentifier(), "Id", matches);
|
||||
addIfAppropriate(searchStr, group.getVersionedComponentId().orElse(null), "Version Control ID", matches);
|
||||
addIfAppropriate(searchStr, group.getName(), "Name", matches);
|
||||
addIfAppropriate(searchStr, group.getComments(), "Comments", matches);
|
||||
|
||||
final ComponentVariableRegistry varRegistry = group.getVariableRegistry();
|
||||
if (varRegistry != null) {
|
||||
final Map<VariableDescriptor, String> variableMap = varRegistry.getVariableMap();
|
||||
for (final Map.Entry<VariableDescriptor, String> entry : variableMap.entrySet()) {
|
||||
addIfAppropriate(searchStr, entry.getKey().getName(), "Variable Name", matches);
|
||||
addIfAppropriate(searchStr, entry.getValue(), "Variable Value", matches);
|
||||
}
|
||||
}
|
||||
|
||||
if (matches.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final ComponentSearchResultDTO result = new ComponentSearchResultDTO();
|
||||
result.setId(group.getIdentifier());
|
||||
result.setName(group.getName());
|
||||
result.setGroupId(parent.getIdentifier());
|
||||
result.setMatches(matches);
|
||||
return result;
|
||||
}
|
||||
|
||||
private ComponentSearchResultDTO search(final String searchStr, final Connection connection) {
|
||||
final List<String> matches = new ArrayList<>();
|
||||
|
||||
// search id and name
|
||||
addIfAppropriate(searchStr, connection.getIdentifier(), "Id", matches);
|
||||
addIfAppropriate(searchStr, connection.getVersionedComponentId().orElse(null), "Version Control ID", matches);
|
||||
addIfAppropriate(searchStr, connection.getName(), "Name", matches);
|
||||
|
||||
// search relationships
|
||||
for (final Relationship relationship : connection.getRelationships()) {
|
||||
addIfAppropriate(searchStr, relationship.getName(), "Relationship", matches);
|
||||
}
|
||||
|
||||
// search prioritizers
|
||||
final FlowFileQueue queue = connection.getFlowFileQueue();
|
||||
for (final FlowFilePrioritizer comparator : queue.getPriorities()) {
|
||||
addIfAppropriate(searchStr, comparator.getClass().getName(), "Prioritizer", matches);
|
||||
}
|
||||
|
||||
// search expiration
|
||||
if (StringUtils.containsIgnoreCase("expires", searchStr) || StringUtils.containsIgnoreCase("expiration", searchStr)) {
|
||||
final int expirationMillis = connection.getFlowFileQueue().getFlowFileExpiration(TimeUnit.MILLISECONDS);
|
||||
if (expirationMillis > 0) {
|
||||
matches.add("FlowFile expiration: " + connection.getFlowFileQueue().getFlowFileExpiration());
|
||||
}
|
||||
}
|
||||
|
||||
// search back pressure
|
||||
if (StringUtils.containsIgnoreCase("back pressure", searchStr) || StringUtils.containsIgnoreCase("pressure", searchStr)) {
|
||||
final String backPressureDataSize = connection.getFlowFileQueue().getBackPressureDataSizeThreshold();
|
||||
final Double backPressureBytes = DataUnit.parseDataSize(backPressureDataSize, DataUnit.B);
|
||||
if (backPressureBytes > 0) {
|
||||
matches.add("Back pressure data size: " + backPressureDataSize);
|
||||
}
|
||||
|
||||
final long backPressureCount = connection.getFlowFileQueue().getBackPressureObjectThreshold();
|
||||
if (backPressureCount > 0) {
|
||||
matches.add("Back pressure count: " + backPressureCount);
|
||||
}
|
||||
}
|
||||
|
||||
// search the source
|
||||
final Connectable source = connection.getSource();
|
||||
addIfAppropriate(searchStr, source.getIdentifier(), "Source id", matches);
|
||||
addIfAppropriate(searchStr, source.getName(), "Source name", matches);
|
||||
addIfAppropriate(searchStr, source.getComments(), "Source comments", matches);
|
||||
|
||||
// search the destination
|
||||
final Connectable destination = connection.getDestination();
|
||||
addIfAppropriate(searchStr, destination.getIdentifier(), "Destination id", matches);
|
||||
addIfAppropriate(searchStr, destination.getName(), "Destination name", matches);
|
||||
addIfAppropriate(searchStr, destination.getComments(), "Destination comments", matches);
|
||||
|
||||
if (matches.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final ComponentSearchResultDTO result = new ComponentSearchResultDTO();
|
||||
result.setId(connection.getIdentifier());
|
||||
|
||||
// determine the name of the search match
|
||||
if (StringUtils.isNotBlank(connection.getName())) {
|
||||
result.setName(connection.getName());
|
||||
} else if (!connection.getRelationships().isEmpty()) {
|
||||
final List<String> relationships = new ArrayList<>(connection.getRelationships().size());
|
||||
for (final Relationship relationship : connection.getRelationships()) {
|
||||
if (StringUtils.isNotBlank(relationship.getName())) {
|
||||
relationships.add(relationship.getName());
|
||||
}
|
||||
}
|
||||
if (!relationships.isEmpty()) {
|
||||
result.setName(StringUtils.join(relationships, ", "));
|
||||
}
|
||||
}
|
||||
|
||||
// ensure a name is added
|
||||
if (result.getName() == null) {
|
||||
result.setName("From source " + connection.getSource().getName());
|
||||
}
|
||||
|
||||
result.setMatches(matches);
|
||||
return result;
|
||||
}
|
||||
|
||||
private ComponentSearchResultDTO search(final String searchStr, final RemoteProcessGroup group) {
|
||||
final List<String> matches = new ArrayList<>();
|
||||
addIfAppropriate(searchStr, group.getIdentifier(), "Id", matches);
|
||||
addIfAppropriate(searchStr, group.getVersionedComponentId().orElse(null), "Version Control ID", matches);
|
||||
addIfAppropriate(searchStr, group.getName(), "Name", matches);
|
||||
addIfAppropriate(searchStr, group.getComments(), "Comments", matches);
|
||||
addIfAppropriate(searchStr, group.getTargetUris(), "URLs", matches);
|
||||
|
||||
// consider the transmission status
|
||||
if ((StringUtils.containsIgnoreCase("transmitting", searchStr) || StringUtils.containsIgnoreCase("transmission enabled", searchStr)) && group.isTransmitting()) {
|
||||
matches.add("Transmission: On");
|
||||
} else if ((StringUtils.containsIgnoreCase("not transmitting", searchStr) || StringUtils.containsIgnoreCase("transmission disabled", searchStr)) && !group.isTransmitting()) {
|
||||
matches.add("Transmission: Off");
|
||||
}
|
||||
|
||||
if (matches.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final ComponentSearchResultDTO result = new ComponentSearchResultDTO();
|
||||
result.setId(group.getIdentifier());
|
||||
result.setName(group.getName());
|
||||
result.setMatches(matches);
|
||||
return result;
|
||||
}
|
||||
|
||||
private ComponentSearchResultDTO search(final String searchStr, final Funnel funnel) {
|
||||
final List<String> matches = new ArrayList<>();
|
||||
addIfAppropriate(searchStr, funnel.getIdentifier(), "Id", matches);
|
||||
addIfAppropriate(searchStr, funnel.getVersionedComponentId().orElse(null), "Version Control ID", matches);
|
||||
|
||||
if (matches.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final ComponentSearchResultDTO dto = new ComponentSearchResultDTO();
|
||||
dto.setId(funnel.getIdentifier());
|
||||
dto.setName(funnel.getName());
|
||||
dto.setMatches(matches);
|
||||
return dto;
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds the nearest versioned parent result group for a given user.
|
||||
*
|
||||
* @param group The containing group
|
||||
* @param user The current NiFi user
|
||||
* @return Versioned parent group
|
||||
*/
|
||||
private SearchResultGroupDTO buildVersionedGroup(final ProcessGroup group, final NiFiUser user) {
|
||||
if (group == null) {
|
||||
return null;
|
||||
}
|
||||
|
||||
ProcessGroup tmpParent = group.getParent();
|
||||
ProcessGroup tmpGroup = group;
|
||||
|
||||
// search for a versioned group by traversing the group tree up to the root
|
||||
while (!tmpGroup.isRootGroup()) {
|
||||
if (tmpGroup.getVersionControlInformation() != null) {
|
||||
return buildResultGroup(tmpGroup, user);
|
||||
}
|
||||
|
||||
tmpGroup = tmpParent;
|
||||
tmpParent = tmpGroup.getParent();
|
||||
}
|
||||
|
||||
// traversed all the way to the root
|
||||
return null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Builds result group for a given user.
|
||||
*
|
||||
* @param group The containing group
|
||||
* @param user The current NiFi user
|
||||
* @return Result group
|
||||
*/
|
||||
private SearchResultGroupDTO buildResultGroup(final ProcessGroup group, final NiFiUser user) {
|
||||
if (group == null) {
|
||||
return null;
|
||||
}
|
||||
|
||||
final SearchResultGroupDTO resultGroup = new SearchResultGroupDTO();
|
||||
resultGroup.setId(group.getIdentifier());
|
||||
|
||||
// keep the group name confidential
|
||||
if (group.isAuthorized(authorizer, RequestAction.READ, user)) {
|
||||
resultGroup.setName(group.getName());
|
||||
}
|
||||
|
||||
return resultGroup;
|
||||
}
|
||||
|
||||
private void addIfAppropriate(final String searchStr, final String value, final String label, final List<String> matches) {
|
||||
if (StringUtils.containsIgnoreCase(value, searchStr)) {
|
||||
matches.add(label + ": " + value);
|
||||
}
|
||||
}
|
||||
|
||||
public void setFlowController(FlowController flowController) {
|
||||
this.flowController = flowController;
|
||||
}
|
||||
|
||||
public void setAuthorizer(Authorizer authorizer) {
|
||||
this.authorizer = authorizer;
|
||||
}
|
||||
|
||||
public void setVariableRegistry(VariableRegistry variableRegistry) {
|
||||
this.variableRegistry = variableRegistry;
|
||||
}
|
||||
}
|
||||
|
|
@ -129,6 +129,11 @@
|
|||
<bean id="policyBasedAuthorizerDAO" class="org.apache.nifi.web.dao.impl.StandardPolicyBasedAuthorizerDAO">
|
||||
<constructor-arg ref="authorizer"/>
|
||||
</bean>
|
||||
<bean id="controllerSearchService" class="org.apache.nifi.web.controller.ControllerSearchService">
|
||||
<property name="flowController" ref="flowController"/>
|
||||
<property name="authorizer" ref="authorizer"/>
|
||||
<property name="variableRegistry" ref="variableRegistry"/>
|
||||
</bean>
|
||||
<bean id="controllerFacade" class="org.apache.nifi.web.controller.ControllerFacade">
|
||||
<property name="properties" ref="nifiProperties"/>
|
||||
<property name="flowController" ref="flowController"/>
|
||||
|
|
@ -136,6 +141,7 @@
|
|||
<property name="authorizer" ref="authorizer"/>
|
||||
<property name="dtoFactory" ref="dtoFactory"/>
|
||||
<property name="variableRegistry" ref="variableRegistry"/>
|
||||
<property name="controllerSearchService" ref="controllerSearchService"/>
|
||||
</bean>
|
||||
<bean id="authorizableLookup" class="org.apache.nifi.authorization.StandardAuthorizableLookup">
|
||||
<property name="controllerFacade" ref="controllerFacade"/>
|
||||
|
|
|
|||
|
|
@ -0,0 +1,405 @@
|
|||
/*
|
||||
* 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.web.controller;
|
||||
|
||||
import org.apache.nifi.authorization.Authorizer;
|
||||
import org.apache.nifi.authorization.RequestAction;
|
||||
import org.apache.nifi.authorization.user.NiFiUser;
|
||||
import org.apache.nifi.controller.ProcessorNode;
|
||||
import org.apache.nifi.controller.StandardProcessorNode;
|
||||
import org.apache.nifi.groups.ProcessGroup;
|
||||
import org.apache.nifi.processor.Processor;
|
||||
import org.apache.nifi.registry.VariableRegistry;
|
||||
import org.apache.nifi.registry.flow.StandardVersionControlInformation;
|
||||
import org.apache.nifi.registry.flow.VersionControlInformation;
|
||||
import org.apache.nifi.registry.variable.MutableVariableRegistry;
|
||||
import org.apache.nifi.web.api.dto.search.SearchResultsDTO;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.mockito.Mockito;
|
||||
|
||||
import java.util.HashSet;
|
||||
|
||||
import static org.junit.Assert.assertTrue;
|
||||
import static org.mockito.Matchers.any;
|
||||
import static org.mockito.Matchers.eq;
|
||||
import static org.mockito.Mockito.mock;
|
||||
|
||||
public class ControllerSearchServiceTest {
|
||||
private MutableVariableRegistry variableRegistry;
|
||||
private ControllerSearchService service;
|
||||
private SearchResultsDTO searchResultsDTO;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
variableRegistry = mock(MutableVariableRegistry.class);
|
||||
service = new ControllerSearchService();
|
||||
searchResultsDTO = new SearchResultsDTO();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSearchInRootLevelAllAuthorizedNoVersionControl() {
|
||||
// root level PG
|
||||
final ProcessGroup rootProcessGroup = setupMockedProcessGroup("root", null, true, variableRegistry, null);
|
||||
|
||||
// first level PGs
|
||||
final ProcessGroup firstLevelAProcessGroup = setupMockedProcessGroup("firstLevelA", rootProcessGroup, true, variableRegistry, null);
|
||||
final ProcessGroup firstLevelBProcessGroup = setupMockedProcessGroup("firstLevelB", rootProcessGroup, true, variableRegistry, null);
|
||||
|
||||
// second level PGs
|
||||
final ProcessGroup secondLevelAProcessGroup = setupMockedProcessGroup("secondLevelA", firstLevelAProcessGroup, true, variableRegistry, null);
|
||||
final ProcessGroup secondLevelBProcessGroup = setupMockedProcessGroup("secondLevelB", firstLevelBProcessGroup, true, variableRegistry, null);
|
||||
// third level PGs
|
||||
final ProcessGroup thirdLevelAProcessGroup = setupMockedProcessGroup("thirdLevelA", secondLevelAProcessGroup, true, variableRegistry, null);
|
||||
final ProcessGroup thirdLevelBProcessGroup = setupMockedProcessGroup("thirdLevelB", secondLevelAProcessGroup, true, variableRegistry, null);
|
||||
|
||||
// link PGs together
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(firstLevelAProcessGroup);
|
||||
add(firstLevelBProcessGroup);
|
||||
}
|
||||
}).when(rootProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(secondLevelAProcessGroup);
|
||||
}
|
||||
}).when(firstLevelAProcessGroup).getProcessGroups();
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(secondLevelBProcessGroup);
|
||||
}
|
||||
}).when(firstLevelBProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(thirdLevelAProcessGroup);
|
||||
add(thirdLevelBProcessGroup);
|
||||
}
|
||||
}).when(secondLevelAProcessGroup).getProcessGroups();
|
||||
|
||||
// setup processor
|
||||
setupMockedProcessor("foobar", rootProcessGroup, true, variableRegistry);
|
||||
|
||||
// perform search
|
||||
service.search(searchResultsDTO, "foo", rootProcessGroup);
|
||||
|
||||
assertTrue(searchResultsDTO.getProcessorResults().size() == 1);
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getId().equals("foobarId"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getParentGroup().getId().equals("rootId"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getParentGroup().getName().equals("root"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getVersionedGroup() == null);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSearchInThirdLevelAllAuthorizedNoVersionControl() {
|
||||
// root level PG
|
||||
final ProcessGroup rootProcessGroup = setupMockedProcessGroup("root", null, true, variableRegistry, null);
|
||||
|
||||
// first level PGs
|
||||
final ProcessGroup firstLevelAProcessGroup = setupMockedProcessGroup("firstLevelA", rootProcessGroup, true, variableRegistry, null);
|
||||
final ProcessGroup firstLevelBProcessGroup = setupMockedProcessGroup("firstLevelB", rootProcessGroup, true, variableRegistry, null);
|
||||
|
||||
// second level PGs
|
||||
final ProcessGroup secondLevelAProcessGroup = setupMockedProcessGroup("secondLevelA", firstLevelAProcessGroup, true, variableRegistry, null);
|
||||
final ProcessGroup secondLevelBProcessGroup = setupMockedProcessGroup("secondLevelB", firstLevelBProcessGroup, true, variableRegistry, null);
|
||||
// third level PGs
|
||||
final ProcessGroup thirdLevelAProcessGroup = setupMockedProcessGroup("thirdLevelA", secondLevelAProcessGroup, true, variableRegistry, null);
|
||||
final ProcessGroup thirdLevelBProcessGroup = setupMockedProcessGroup("thirdLevelB", secondLevelAProcessGroup, true, variableRegistry, null);
|
||||
|
||||
// link PGs together
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(firstLevelAProcessGroup);
|
||||
add(firstLevelBProcessGroup);
|
||||
}
|
||||
}).when(rootProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(secondLevelAProcessGroup);
|
||||
}
|
||||
}).when(firstLevelAProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(secondLevelBProcessGroup);
|
||||
}
|
||||
}).when(firstLevelBProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(thirdLevelAProcessGroup);
|
||||
add(thirdLevelBProcessGroup);
|
||||
}
|
||||
}).when(secondLevelAProcessGroup).getProcessGroups();
|
||||
|
||||
// setup processor
|
||||
setupMockedProcessor("foobar", thirdLevelAProcessGroup, true, variableRegistry);
|
||||
|
||||
// perform search
|
||||
service.search(searchResultsDTO, "foo", rootProcessGroup);
|
||||
|
||||
assertTrue(searchResultsDTO.getProcessorResults().size() == 1);
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getId().equals("foobarId"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getParentGroup().getId().equals("thirdLevelAId"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getParentGroup().getName().equals("thirdLevelA"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getVersionedGroup() == null);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSearchInThirdLevelParentNotAuthorizedNoVersionControl() {
|
||||
// root level PG
|
||||
final ProcessGroup rootProcessGroup = setupMockedProcessGroup("root", null, true, variableRegistry, null);
|
||||
|
||||
// first level PGs
|
||||
final ProcessGroup firstLevelAProcessGroup = setupMockedProcessGroup("firstLevelA", rootProcessGroup, true, variableRegistry, null);
|
||||
final ProcessGroup firstLevelBProcessGroup = setupMockedProcessGroup("firstLevelB", rootProcessGroup, true, variableRegistry, null);
|
||||
|
||||
// second level PGs
|
||||
final ProcessGroup secondLevelAProcessGroup = setupMockedProcessGroup("secondLevelA", firstLevelAProcessGroup, true, variableRegistry, null);
|
||||
final ProcessGroup secondLevelBProcessGroup = setupMockedProcessGroup("secondLevelB", firstLevelBProcessGroup, true, variableRegistry, null);
|
||||
// third level PGs - not authorized
|
||||
final ProcessGroup thirdLevelAProcessGroup = setupMockedProcessGroup("thirdLevelA", secondLevelAProcessGroup, false, variableRegistry, null);
|
||||
final ProcessGroup thirdLevelBProcessGroup = setupMockedProcessGroup("thirdLevelB", secondLevelAProcessGroup, false, variableRegistry, null);
|
||||
|
||||
// link PGs together
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(firstLevelAProcessGroup);
|
||||
add(firstLevelBProcessGroup);
|
||||
}
|
||||
}).when(rootProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(secondLevelAProcessGroup);
|
||||
}
|
||||
}).when(firstLevelAProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(secondLevelBProcessGroup);
|
||||
}
|
||||
}).when(firstLevelBProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(thirdLevelAProcessGroup);
|
||||
add(thirdLevelBProcessGroup);
|
||||
}
|
||||
}).when(secondLevelAProcessGroup).getProcessGroups();
|
||||
|
||||
// setup processor
|
||||
setupMockedProcessor("foobar", thirdLevelAProcessGroup, true, variableRegistry);
|
||||
|
||||
// perform search
|
||||
service.search(searchResultsDTO, "foo", rootProcessGroup);
|
||||
|
||||
assertTrue(searchResultsDTO.getProcessorResults().size() == 1);
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getId().equals("foobarId"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getParentGroup().getId().equals("thirdLevelAId"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getParentGroup().getName() == null);
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getVersionedGroup() == null);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSearchInThirdLevelParentNotAuthorizedWithVersionControl() {
|
||||
// root level PG
|
||||
final ProcessGroup rootProcessGroup = setupMockedProcessGroup("root", null, true, variableRegistry, null);
|
||||
|
||||
// first level PGs
|
||||
final VersionControlInformation versionControlInformation = setupVC();
|
||||
final ProcessGroup firstLevelAProcessGroup = setupMockedProcessGroup("firstLevelA", rootProcessGroup, true, variableRegistry, versionControlInformation);
|
||||
final ProcessGroup firstLevelBProcessGroup = setupMockedProcessGroup("firstLevelB", rootProcessGroup, true, variableRegistry, null);
|
||||
|
||||
// second level PGs
|
||||
final ProcessGroup secondLevelAProcessGroup = setupMockedProcessGroup("secondLevelA", firstLevelAProcessGroup, true, variableRegistry, null);
|
||||
final ProcessGroup secondLevelBProcessGroup = setupMockedProcessGroup("secondLevelB", firstLevelBProcessGroup, true, variableRegistry, null);
|
||||
// third level PGs - not authorized
|
||||
final ProcessGroup thirdLevelAProcessGroup = setupMockedProcessGroup("thirdLevelA", secondLevelAProcessGroup, false, variableRegistry, null);
|
||||
final ProcessGroup thirdLevelBProcessGroup = setupMockedProcessGroup("thirdLevelB", secondLevelAProcessGroup, false, variableRegistry, null);
|
||||
|
||||
// link PGs together
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(firstLevelAProcessGroup);
|
||||
add(firstLevelBProcessGroup);
|
||||
}
|
||||
}).when(rootProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(secondLevelAProcessGroup);
|
||||
}
|
||||
}).when(firstLevelAProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(secondLevelBProcessGroup);
|
||||
}
|
||||
}).when(firstLevelBProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(thirdLevelAProcessGroup);
|
||||
add(thirdLevelBProcessGroup);
|
||||
}
|
||||
}).when(secondLevelAProcessGroup).getProcessGroups();
|
||||
|
||||
// setup processor
|
||||
setupMockedProcessor("foobar", thirdLevelAProcessGroup, true, variableRegistry);
|
||||
|
||||
// perform search
|
||||
service.search(searchResultsDTO, "foo", rootProcessGroup);
|
||||
|
||||
assertTrue(searchResultsDTO.getProcessorResults().size() == 1);
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getId().equals("foobarId"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getParentGroup().getId().equals("thirdLevelAId"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getParentGroup().getName() == null);
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getVersionedGroup() != null);
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getVersionedGroup().getId().equals("firstLevelAId"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getVersionedGroup().getName().equals("firstLevelA"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSearchInThirdLevelParentNotAuthorizedWithVersionControlInTheGroup() {
|
||||
// root level PG
|
||||
final ProcessGroup rootProcessGroup = setupMockedProcessGroup("root", null, true, variableRegistry, null);
|
||||
|
||||
// first level PGs
|
||||
final ProcessGroup firstLevelAProcessGroup = setupMockedProcessGroup("firstLevelA", rootProcessGroup, true, variableRegistry, null);
|
||||
final ProcessGroup firstLevelBProcessGroup = setupMockedProcessGroup("firstLevelB", rootProcessGroup, true, variableRegistry, null);
|
||||
|
||||
// second level PGs
|
||||
final ProcessGroup secondLevelAProcessGroup = setupMockedProcessGroup("secondLevelA", firstLevelAProcessGroup, true, variableRegistry, null);
|
||||
final ProcessGroup secondLevelBProcessGroup = setupMockedProcessGroup("secondLevelB", firstLevelBProcessGroup, true, variableRegistry, null);
|
||||
// third level PGs - not authorized
|
||||
final VersionControlInformation versionControlInformation = setupVC();
|
||||
final ProcessGroup thirdLevelAProcessGroup = setupMockedProcessGroup("thirdLevelA", secondLevelAProcessGroup, false, variableRegistry, versionControlInformation);
|
||||
final ProcessGroup thirdLevelBProcessGroup = setupMockedProcessGroup("thirdLevelB", secondLevelAProcessGroup, false, variableRegistry, null);
|
||||
|
||||
// link PGs together
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(firstLevelAProcessGroup);
|
||||
add(firstLevelBProcessGroup);
|
||||
}
|
||||
}).when(rootProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(secondLevelAProcessGroup);
|
||||
}
|
||||
}).when(firstLevelAProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(secondLevelBProcessGroup);
|
||||
}
|
||||
}).when(firstLevelBProcessGroup).getProcessGroups();
|
||||
|
||||
Mockito.doReturn(new HashSet<ProcessGroup>() {
|
||||
{
|
||||
add(thirdLevelAProcessGroup);
|
||||
add(thirdLevelBProcessGroup);
|
||||
}
|
||||
}).when(secondLevelAProcessGroup).getProcessGroups();
|
||||
|
||||
// setup processor
|
||||
setupMockedProcessor("foobar", thirdLevelAProcessGroup, true, variableRegistry);
|
||||
|
||||
// perform search
|
||||
service.search(searchResultsDTO, "foo", rootProcessGroup);
|
||||
|
||||
assertTrue(searchResultsDTO.getProcessorResults().size() == 1);
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getId().equals("foobarId"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getParentGroup().getId().equals("thirdLevelAId"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getParentGroup().getName() == null);
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getVersionedGroup() != null);
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getVersionedGroup().getId().equals("thirdLevelAId"));
|
||||
assertTrue(searchResultsDTO.getProcessorResults().get(0).getVersionedGroup().getName() == null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Mocks Processor including isAuthorized() and its name & id.
|
||||
*
|
||||
* @param processorName Desired processor name
|
||||
* @param containingProcessGroup The process group
|
||||
* @param authorizedToRead Can the processor data be read?
|
||||
* @param variableRegistry The variable registry
|
||||
*/
|
||||
private static void setupMockedProcessor(final String processorName, final ProcessGroup containingProcessGroup, boolean authorizedToRead, final MutableVariableRegistry variableRegistry) {
|
||||
final String processorId = processorName + "Id";
|
||||
final Processor processor1 = mock(Processor.class);
|
||||
|
||||
final ProcessorNode processorNode1 = mock(StandardProcessorNode.class);
|
||||
Mockito.doReturn(authorizedToRead).when(processorNode1).isAuthorized(any(Authorizer.class), eq(RequestAction.READ), any(NiFiUser.class));
|
||||
Mockito.doReturn(variableRegistry).when(processorNode1).getVariableRegistry();
|
||||
Mockito.doReturn(processor1).when(processorNode1).getProcessor();
|
||||
// set processor node's attributes
|
||||
Mockito.doReturn(processorId).when(processorNode1).getIdentifier();
|
||||
Mockito.doReturn(processorName).when(processorNode1).getName();
|
||||
|
||||
// assign processor node to its PG
|
||||
Mockito.doReturn(new HashSet<ProcessorNode>() {
|
||||
{
|
||||
add(processorNode1);
|
||||
}
|
||||
}).when(containingProcessGroup).getProcessors();
|
||||
}
|
||||
|
||||
/**
|
||||
* Mocks ProcessGroup due to isAuthorized(). The final class StandardProcessGroup can't be used.
|
||||
*
|
||||
* @param processGroupName Desired process group name
|
||||
* @param parent The parent process group
|
||||
* @param authorizedToRead Can the process group data be read?
|
||||
* @param variableRegistry The variable registry
|
||||
* @param versionControlInformation The version control information
|
||||
* @return Mocked process group
|
||||
*/
|
||||
private static ProcessGroup setupMockedProcessGroup(final String processGroupName, final ProcessGroup parent, boolean authorizedToRead, final VariableRegistry variableRegistry,
|
||||
final VersionControlInformation versionControlInformation) {
|
||||
final String processGroupId = processGroupName + "Id";
|
||||
final ProcessGroup processGroup = mock(ProcessGroup.class);
|
||||
|
||||
Mockito.doReturn(processGroupId).when(processGroup).getIdentifier();
|
||||
Mockito.doReturn(processGroupName).when(processGroup).getName();
|
||||
Mockito.doReturn(parent).when(processGroup).getParent();
|
||||
Mockito.doReturn(versionControlInformation).when(processGroup).getVersionControlInformation();
|
||||
Mockito.doReturn(variableRegistry).when(processGroup).getVariableRegistry();
|
||||
Mockito.doReturn(parent == null).when(processGroup).isRootGroup();
|
||||
// override process group's access rights
|
||||
Mockito.doReturn(authorizedToRead).when(processGroup).isAuthorized(any(Authorizer.class), eq(RequestAction.READ), any(NiFiUser.class));
|
||||
|
||||
return processGroup;
|
||||
}
|
||||
|
||||
/**
|
||||
* Creates a version control information using dummy attributes.
|
||||
*
|
||||
* @return Dummy version control information
|
||||
*/
|
||||
private static VersionControlInformation setupVC() {
|
||||
final StandardVersionControlInformation.Builder builder = new StandardVersionControlInformation.Builder();
|
||||
builder.registryId("regId").bucketId("bucId").flowId("flowId").version(1);
|
||||
|
||||
return builder.build();
|
||||
}
|
||||
}
|
||||
|
|
@ -200,7 +200,23 @@
|
|||
}
|
||||
},
|
||||
_renderItem: function (ul, match) {
|
||||
var itemContent = $('<a></a>').append($('<div class="search-match-header"></div>').text(match.name));
|
||||
var itemHeader = $('<div class="search-match-header"></div>').text(match.name);
|
||||
|
||||
var parentGroupHeader = $('<div class="search-match-header"></div>').append(document.createTextNode('Parent: '));
|
||||
var parentGroup = match.parentGroup.name ? match.parentGroup.name : match.parentGroup.id;
|
||||
parentGroupHeader = parentGroupHeader.append($('<span></span>').text(parentGroup));
|
||||
|
||||
var versionedGroupHeader = $('<div class="search-match-header"></div>').append(document.createTextNode('Versioned: '));
|
||||
var versionedGroup = '-';
|
||||
|
||||
if (nfCommon.isDefinedAndNotNull(match.versionedGroup)) {
|
||||
versionedGroup = match.versionedGroup.name ? match.versionedGroup.name : match.versionedGroup.id;
|
||||
}
|
||||
|
||||
versionedGroupHeader = versionedGroupHeader.append($('<span></span>').text(versionedGroup));
|
||||
// create a search item wrapper
|
||||
var itemContent = $('<a></a>').append(itemHeader).append(parentGroupHeader).append(versionedGroupHeader);
|
||||
// append all matches
|
||||
$.each(match.matches, function (i, match) {
|
||||
itemContent.append($('<div class="search-match"></div>').text(match));
|
||||
});
|
||||
|
|
@ -231,9 +247,10 @@
|
|||
},
|
||||
select: function (event, ui) {
|
||||
var item = ui.item;
|
||||
var group = item.parentGroup;
|
||||
|
||||
// show the selected component
|
||||
nfCanvasUtils.showComponent(item.groupId, item.id);
|
||||
nfCanvasUtils.showComponent(group.id, item.id);
|
||||
|
||||
searchCtrl.getInputElement().val('').blur();
|
||||
|
||||
|
|
|
|||
Loading…
Reference in New Issue