| /* |
| * 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.remote; |
| |
| import com.sun.jersey.api.client.ClientHandlerException; |
| import com.sun.jersey.api.client.ClientResponse.Status; |
| import com.sun.jersey.api.client.UniformInterfaceException; |
| import org.apache.nifi.authorization.Resource; |
| import org.apache.nifi.authorization.resource.Authorizable; |
| import org.apache.nifi.authorization.resource.ResourceFactory; |
| import org.apache.nifi.authorization.resource.ResourceType; |
| import org.apache.nifi.connectable.ConnectableType; |
| import org.apache.nifi.connectable.Connection; |
| import org.apache.nifi.connectable.Port; |
| import org.apache.nifi.connectable.Position; |
| import org.apache.nifi.controller.FlowController; |
| import org.apache.nifi.controller.ProcessScheduler; |
| import org.apache.nifi.controller.ScheduledState; |
| import org.apache.nifi.controller.exception.CommunicationsException; |
| import org.apache.nifi.engine.FlowEngine; |
| import org.apache.nifi.events.BulletinFactory; |
| import org.apache.nifi.events.EventReporter; |
| import org.apache.nifi.groups.ProcessGroup; |
| import org.apache.nifi.groups.ProcessGroupCounts; |
| import org.apache.nifi.groups.RemoteProcessGroup; |
| import org.apache.nifi.groups.RemoteProcessGroupPortDescriptor; |
| import org.apache.nifi.remote.protocol.SiteToSiteTransportProtocol; |
| import org.apache.nifi.remote.protocol.http.HttpProxy; |
| import org.apache.nifi.remote.util.SiteToSiteRestApiClient; |
| import org.apache.nifi.reporting.BulletinRepository; |
| import org.apache.nifi.reporting.ComponentType; |
| import org.apache.nifi.reporting.Severity; |
| import org.apache.nifi.util.FormatUtils; |
| import org.apache.nifi.util.NiFiProperties; |
| import org.apache.nifi.web.api.dto.ControllerDTO; |
| import org.apache.nifi.web.api.dto.PortDTO; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| |
| import javax.net.ssl.SSLContext; |
| import java.io.File; |
| import java.io.IOException; |
| import java.net.URI; |
| import java.net.URISyntaxException; |
| import java.util.ArrayList; |
| import java.util.Date; |
| import java.util.HashMap; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.LinkedHashSet; |
| import java.util.List; |
| import java.util.Map; |
| import java.util.Set; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.concurrent.atomic.AtomicReference; |
| import java.util.concurrent.locks.Lock; |
| import java.util.concurrent.locks.ReadWriteLock; |
| import java.util.concurrent.locks.ReentrantReadWriteLock; |
| |
| import static java.util.Objects.requireNonNull; |
| |
| /** |
| * Represents the Root Process Group of a remote NiFi Instance. Holds information about that remote instance, as well as {@link IncomingPort}s and {@link OutgoingPort}s for communicating with the |
| * remote instance. |
| */ |
| public class StandardRemoteProcessGroup implements RemoteProcessGroup { |
| |
| private static final Logger logger = LoggerFactory.getLogger(StandardRemoteProcessGroup.class); |
| |
| // status codes |
| private static final int UNAUTHORIZED_STATUS_CODE = Status.UNAUTHORIZED.getStatusCode(); |
| private static final int FORBIDDEN_STATUS_CODE = Status.FORBIDDEN.getStatusCode(); |
| |
| private final String id; |
| |
| private final URI targetUri; |
| private final URI apiUri; |
| private final String host; |
| private final String protocol; |
| private final ProcessScheduler scheduler; |
| private final EventReporter eventReporter; |
| |
| private final AtomicReference<String> name = new AtomicReference<>(); |
| private final AtomicReference<Position> position = new AtomicReference<>(); |
| private final AtomicReference<String> comments = new AtomicReference<>(); |
| private final AtomicReference<ProcessGroup> processGroup; |
| private final AtomicBoolean transmitting = new AtomicBoolean(false); |
| private final FlowController flowController; |
| private final SSLContext sslContext; |
| private final AtomicReference<Boolean> pointsToCluster = new AtomicReference<>(null); |
| private final AtomicBoolean targetIsUnreachable = new AtomicBoolean(false); |
| |
| private volatile String communicationsTimeout = "30 sec"; |
| private volatile String targetId; |
| private volatile String yieldDuration = "10 sec"; |
| private volatile SiteToSiteTransportProtocol transportProtocol = SiteToSiteTransportProtocol.RAW; |
| private volatile String proxyHost; |
| private volatile Integer proxyPort; |
| private volatile String proxyUser; |
| private volatile String proxyPassword; |
| |
| |
| private final ReadWriteLock rwLock = new ReentrantReadWriteLock(); |
| private final Lock readLock = rwLock.readLock(); |
| private final Lock writeLock = rwLock.writeLock(); |
| |
| // the following variables are all protected by the read/write lock above. |
| // Maps a Port Name to an OutgoingPort that can be used to push files to that port |
| private final Map<String, StandardRemoteGroupPort> inputPorts = new HashMap<>(); |
| // Maps a Port Name to a PullingPort that can be used to receive files from that port |
| private final Map<String, StandardRemoteGroupPort> outputPorts = new HashMap<>(); |
| |
| private ProcessGroupCounts counts = new ProcessGroupCounts(0, 0, 0, 0, 0, 0, 0, 0); |
| private Long refreshContentsTimestamp = null; |
| private Boolean destinationSecure; |
| private Integer listeningPort; |
| private Integer listeningHttpPort; |
| |
| private volatile String authorizationIssue; |
| |
| private final ScheduledExecutorService backgroundThreadExecutor; |
| |
| public StandardRemoteProcessGroup(final String id, final String targetUri, final ProcessGroup processGroup, |
| final FlowController flowController, final SSLContext sslContext) { |
| this.id = requireNonNull(id); |
| this.flowController = requireNonNull(flowController); |
| final URI uri; |
| try { |
| uri = new URI(requireNonNull(targetUri)); |
| |
| // Trim the trailing / |
| String uriPath = uri.getPath(); |
| if (uriPath.endsWith("/")) { |
| uriPath = uriPath.substring(0, uriPath.length() - 1); |
| } |
| final String apiPath = uri.getScheme() + "://" + uri.getHost() + ":" + uri.getPort() + uriPath + "-api"; |
| apiUri = new URI(apiPath); |
| } catch (final URISyntaxException e) { |
| throw new IllegalArgumentException(e); |
| } |
| |
| this.host = uri.getHost(); |
| this.protocol = uri.getAuthority(); |
| this.targetUri = uri; |
| this.targetId = null; |
| this.processGroup = new AtomicReference<>(processGroup); |
| this.sslContext = sslContext; |
| this.scheduler = flowController.getProcessScheduler(); |
| this.authorizationIssue = "Establishing connection to " + targetUri; |
| |
| final BulletinRepository bulletinRepository = flowController.getBulletinRepository(); |
| eventReporter = new EventReporter() { |
| private static final long serialVersionUID = 1L; |
| |
| @Override |
| public void reportEvent(final Severity severity, final String category, final String message) { |
| final String groupId = StandardRemoteProcessGroup.this.getProcessGroup().getIdentifier(); |
| final String sourceId = StandardRemoteProcessGroup.this.getIdentifier(); |
| final String sourceName = StandardRemoteProcessGroup.this.getName(); |
| bulletinRepository.addBulletin(BulletinFactory.createBulletin(groupId, sourceId, ComponentType.REMOTE_PROCESS_GROUP, |
| sourceName, category, severity.name(), message)); |
| } |
| }; |
| |
| final Runnable checkAuthorizations = new InitializationTask(); |
| backgroundThreadExecutor = new FlowEngine(1, "Remote Process Group " + id + ": " + targetUri); |
| backgroundThreadExecutor.scheduleWithFixedDelay(checkAuthorizations, 5L, 30L, TimeUnit.SECONDS); |
| } |
| |
| @Override |
| public void reinitialize(boolean isClustered) { |
| this.pointsToCluster.set(null); |
| backgroundThreadExecutor.submit(new InitializationTask()); |
| } |
| |
| @Override |
| public void onRemove() { |
| backgroundThreadExecutor.shutdown(); |
| |
| final File file = getPeerPersistenceFile(); |
| if (file.exists() && !file.delete()) { |
| logger.warn("Failed to remove {}. This file should be removed manually.", file); |
| } |
| } |
| |
| @Override |
| public void shutdown() { |
| backgroundThreadExecutor.shutdown(); |
| } |
| |
| @Override |
| public String getIdentifier() { |
| return id; |
| } |
| |
| @Override |
| public Authorizable getParentAuthorizable() { |
| return getProcessGroup(); |
| } |
| |
| @Override |
| public Resource getResource() { |
| return ResourceFactory.getComponentResource(ResourceType.RemoteProcessGroup, getIdentifier(), getName()); |
| } |
| |
| @Override |
| public ProcessGroup getProcessGroup() { |
| return processGroup.get(); |
| } |
| |
| @Override |
| public void setProcessGroup(final ProcessGroup group) { |
| this.processGroup.set(group); |
| for (final RemoteGroupPort port : getInputPorts()) { |
| port.setProcessGroup(group); |
| } |
| for (final RemoteGroupPort port : getOutputPorts()) { |
| port.setProcessGroup(group); |
| } |
| } |
| |
| public void setTargetId(final String targetId) { |
| this.targetId = targetId; |
| } |
| |
| @Override |
| public void setTransportProtocol(final SiteToSiteTransportProtocol transportProtocol) { |
| this.transportProtocol = transportProtocol; |
| } |
| |
| @Override |
| public SiteToSiteTransportProtocol getTransportProtocol() { |
| return transportProtocol; |
| } |
| |
| @Override |
| public String getProxyHost() { |
| return proxyHost; |
| } |
| |
| @Override |
| public void setProxyHost(String proxyHost) { |
| this.proxyHost = proxyHost; |
| } |
| |
| @Override |
| public Integer getProxyPort() { |
| return proxyPort; |
| } |
| |
| @Override |
| public void setProxyPort(Integer proxyPort) { |
| this.proxyPort = proxyPort; |
| } |
| |
| @Override |
| public String getProxyUser() { |
| return proxyUser; |
| } |
| |
| @Override |
| public void setProxyUser(String proxyUser) { |
| this.proxyUser = proxyUser; |
| } |
| |
| @Override |
| public String getProxyPassword() { |
| return proxyPassword; |
| } |
| |
| @Override |
| public void setProxyPassword(String proxyPassword) { |
| this.proxyPassword = proxyPassword; |
| } |
| |
| /** |
| * @return the ID of the Root Group on the remote instance |
| */ |
| public String getTargetId() { |
| return targetId; |
| } |
| |
| public String getProtocol() { |
| return protocol; |
| } |
| |
| @Override |
| public String getName() { |
| final String name = this.name.get(); |
| return name == null ? targetUri.toString() : name; |
| } |
| |
| @Override |
| public void setName(final String name) { |
| this.name.set(name); |
| } |
| |
| @Override |
| public String getCommunicationsTimeout() { |
| return communicationsTimeout; |
| } |
| |
| @Override |
| public void setCommunicationsTimeout(final String timePeriod) throws IllegalArgumentException { |
| // verify the timePeriod is legit |
| try { |
| final long millis = FormatUtils.getTimeDuration(timePeriod, TimeUnit.MILLISECONDS); |
| if (millis <= 0) { |
| throw new IllegalArgumentException("Time Period must be more than 0 milliseconds; Invalid Time Period: " + timePeriod); |
| } |
| if (millis > Integer.MAX_VALUE) { |
| throw new IllegalArgumentException("Timeout is too long; cannot be greater than " + Integer.MAX_VALUE + " milliseconds"); |
| } |
| this.communicationsTimeout = timePeriod; |
| } catch (final Exception e) { |
| throw new IllegalArgumentException("Invalid Time Period: " + timePeriod); |
| } |
| } |
| |
| @Override |
| public int getCommunicationsTimeout(final TimeUnit timeUnit) { |
| return (int) FormatUtils.getTimeDuration(communicationsTimeout, timeUnit); |
| } |
| |
| @Override |
| public String getComments() { |
| return comments.get(); |
| } |
| |
| @Override |
| public void setComments(final String comments) { |
| this.comments.set(comments); |
| } |
| |
| @Override |
| public Position getPosition() { |
| return position.get(); |
| } |
| |
| @Override |
| public void setPosition(final Position position) { |
| this.position.set(position); |
| } |
| |
| @Override |
| public URI getTargetUri() { |
| return targetUri; |
| } |
| |
| @Override |
| public String getAuthorizationIssue() { |
| return authorizationIssue; |
| } |
| |
| public String getHost() { |
| return host; |
| } |
| |
| public int getInputPortCount() { |
| readLock.lock(); |
| try { |
| return inputPorts.size(); |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| public int getOutputPortCount() { |
| readLock.lock(); |
| try { |
| return outputPorts.size(); |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| public boolean containsInputPort(final String id) { |
| readLock.lock(); |
| try { |
| return inputPorts.containsKey(id); |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| /** |
| * Changes the currently configured input ports to the ports described in the given set. If any port is currently configured that is not in the set given, that port will be shutdown and removed. |
| * If any port is currently not configured and is in the set given, that port will be instantiated and started. |
| * |
| * @param ports the new ports |
| * |
| * @throws NullPointerException if the given argument is null |
| */ |
| @Override |
| public void setInputPorts(final Set<RemoteProcessGroupPortDescriptor> ports) { |
| writeLock.lock(); |
| try { |
| final List<String> newPortIds = new ArrayList<>(); |
| for (final RemoteProcessGroupPortDescriptor descriptor : ports) { |
| newPortIds.add(descriptor.getId()); |
| |
| if (!inputPorts.containsKey(descriptor.getId())) { |
| addInputPort(descriptor); |
| } |
| |
| // set the comments to ensure current description |
| final StandardRemoteGroupPort sendPort = inputPorts.get(descriptor.getId()); |
| sendPort.setTargetExists(true); |
| sendPort.setName(descriptor.getName()); |
| if (descriptor.isTargetRunning() != null) { |
| sendPort.setTargetRunning(descriptor.isTargetRunning()); |
| } |
| sendPort.setComments(descriptor.getComments()); |
| } |
| |
| // See if we have any ports that no longer exist; cannot be removed within the loop because it would cause |
| // a ConcurrentModificationException. |
| final Iterator<Map.Entry<String, StandardRemoteGroupPort>> itr = inputPorts.entrySet().iterator(); |
| while (itr.hasNext()) { |
| final Map.Entry<String, StandardRemoteGroupPort> entry = itr.next(); |
| if (!newPortIds.contains(entry.getKey())) { |
| final StandardRemoteGroupPort port = entry.getValue(); |
| port.setTargetExists(false); |
| |
| // If port has incoming connection, it will be cleaned up when the connection is removed |
| if (!port.hasIncomingConnection()) { |
| itr.remove(); |
| } |
| } |
| } |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| /** |
| * Returns a boolean indicating whether or not an Output Port exists with the given ID |
| * |
| * @param id identifier of port |
| * @return <code>true</code> if an Output Port exists with the given ID, <code>false</code> otherwise. |
| */ |
| public boolean containsOutputPort(final String id) { |
| readLock.lock(); |
| try { |
| return outputPorts.containsKey(id); |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| /** |
| * Changes the currently configured output ports to the ports described in the given set. If any port is currently configured that is not in the set given, that port will be shutdown and removed. |
| * If any port is currently not configured and is in the set given, that port will be instantiated and started. |
| * |
| * @param ports the new ports |
| * |
| * @throws NullPointerException if the given argument is null |
| */ |
| @Override |
| public void setOutputPorts(final Set<RemoteProcessGroupPortDescriptor> ports) { |
| writeLock.lock(); |
| try { |
| final List<String> newPortIds = new ArrayList<>(); |
| for (final RemoteProcessGroupPortDescriptor descriptor : requireNonNull(ports)) { |
| newPortIds.add(descriptor.getId()); |
| |
| if (!outputPorts.containsKey(descriptor.getId())) { |
| addOutputPort(descriptor); |
| } |
| |
| // set the comments to ensure current description |
| final StandardRemoteGroupPort receivePort = outputPorts.get(descriptor.getId()); |
| receivePort.setTargetExists(true); |
| receivePort.setName(descriptor.getName()); |
| if (descriptor.isTargetRunning() != null) { |
| receivePort.setTargetRunning(descriptor.isTargetRunning()); |
| } |
| receivePort.setComments(descriptor.getComments()); |
| } |
| |
| // See if we have any ports that no longer exist; cannot be removed within the loop because it would cause |
| // a ConcurrentModificationException. |
| final Iterator<Map.Entry<String, StandardRemoteGroupPort>> itr = outputPorts.entrySet().iterator(); |
| while (itr.hasNext()) { |
| final Map.Entry<String, StandardRemoteGroupPort> entry = itr.next(); |
| if (!newPortIds.contains(entry.getKey())) { |
| final StandardRemoteGroupPort port = entry.getValue(); |
| port.setTargetExists(false); |
| |
| // If port has connections, it will be cleaned up when connections are removed |
| if (port.getConnections().isEmpty()) { |
| itr.remove(); |
| } |
| } |
| } |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| /** |
| * Shuts down and removes the given port |
| * |
| * |
| * @throws NullPointerException if the given output Port is null |
| * @throws IllegalStateException if the port does not belong to this remote process group |
| */ |
| @Override |
| public void removeNonExistentPort(final RemoteGroupPort port) { |
| writeLock.lock(); |
| try { |
| if (requireNonNull(port).getTargetExists()) { |
| throw new IllegalStateException("Cannot remove Remote Port " + port.getIdentifier() + " because it still exists on the Remote Instance"); |
| } |
| if (!port.getConnections().isEmpty() || port.hasIncomingConnection()) { |
| throw new IllegalStateException("Cannot remove Remote Port because it is connected to other components"); |
| } |
| |
| scheduler.stopPort(port); |
| |
| if (outputPorts.containsKey(port.getIdentifier())) { |
| outputPorts.remove(port.getIdentifier()); |
| } else { |
| if (!inputPorts.containsKey(port.getIdentifier())) { |
| throw new IllegalStateException("Cannot remove Remote Port because it does not belong to this Remote Process Group"); |
| } |
| |
| inputPorts.remove(port.getIdentifier()); |
| } |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| @Override |
| public void removeAllNonExistentPorts() { |
| writeLock.lock(); |
| try { |
| final Set<String> inputPortIds = new HashSet<>(); |
| final Set<String> outputPortIds = new HashSet<>(); |
| |
| for (final Map.Entry<String, StandardRemoteGroupPort> entry : inputPorts.entrySet()) { |
| final RemoteGroupPort port = entry.getValue(); |
| |
| if (port.getTargetExists()) { |
| continue; |
| } |
| |
| // If there's a connection, we don't remove it. |
| if (port.hasIncomingConnection()) { |
| continue; |
| } |
| |
| inputPortIds.add(entry.getKey()); |
| } |
| |
| for (final Map.Entry<String, StandardRemoteGroupPort> entry : outputPorts.entrySet()) { |
| final RemoteGroupPort port = entry.getValue(); |
| |
| if (port.getTargetExists()) { |
| continue; |
| } |
| |
| // If there's a connection, we don't remove it. |
| if (!port.getConnections().isEmpty()) { |
| continue; |
| } |
| |
| outputPortIds.add(entry.getKey()); |
| } |
| |
| for (final String id : inputPortIds) { |
| inputPorts.remove(id); |
| } |
| for (final String id : outputPortIds) { |
| outputPorts.remove(id); |
| } |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| /** |
| * Adds an Output Port to this Remote Process Group that is described by this DTO. |
| * |
| * @param descriptor |
| * |
| * @throws IllegalStateException if an Output Port already exists with the ID given by dto.getId() |
| */ |
| private void addOutputPort(final RemoteProcessGroupPortDescriptor descriptor) { |
| writeLock.lock(); |
| try { |
| if (outputPorts.containsKey(requireNonNull(descriptor).getId())) { |
| throw new IllegalStateException("Output Port with ID " + descriptor.getId() + " already exists"); |
| } |
| |
| final StandardRemoteGroupPort port = new StandardRemoteGroupPort(descriptor.getId(), descriptor.getName(), getProcessGroup(), |
| this, TransferDirection.RECEIVE, ConnectableType.REMOTE_OUTPUT_PORT, sslContext, scheduler); |
| outputPorts.put(descriptor.getId(), port); |
| |
| if (descriptor.getConcurrentlySchedulableTaskCount() != null) { |
| port.setMaxConcurrentTasks(descriptor.getConcurrentlySchedulableTaskCount()); |
| } |
| if (descriptor.getUseCompression() != null) { |
| port.setUseCompression(descriptor.getUseCompression()); |
| } |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| /** |
| * @param portIdentifier the ID of the Port to send FlowFiles to |
| * @return {@link RemoteGroupPort} that can be used to send FlowFiles to the port whose ID is given on the remote instance |
| */ |
| @Override |
| public RemoteGroupPort getInputPort(final String portIdentifier) { |
| readLock.lock(); |
| try { |
| if (requireNonNull(portIdentifier).startsWith(id + "-")) { |
| return inputPorts.get(portIdentifier.substring(id.length() + 1)); |
| } else { |
| return inputPorts.get(portIdentifier); |
| } |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| /** |
| * @return a set of {@link OutgoingPort}s used for transmitting FlowFiles to the remote instance |
| */ |
| @Override |
| public Set<RemoteGroupPort> getInputPorts() { |
| readLock.lock(); |
| try { |
| final Set<RemoteGroupPort> set = new HashSet<>(); |
| set.addAll(inputPorts.values()); |
| return set; |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| /** |
| * Adds an InputPort to this ProcessGroup that is described by the given DTO. |
| * |
| * @param descriptor port descriptor |
| * |
| * @throws IllegalStateException if an Input Port already exists with the ID given by the ID of the DTO. |
| */ |
| private void addInputPort(final RemoteProcessGroupPortDescriptor descriptor) { |
| writeLock.lock(); |
| try { |
| if (inputPorts.containsKey(descriptor.getId())) { |
| throw new IllegalStateException("Input Port with ID " + descriptor.getId() + " already exists"); |
| } |
| |
| final StandardRemoteGroupPort port = new StandardRemoteGroupPort(descriptor.getId(), descriptor.getName(), getProcessGroup(), this, |
| TransferDirection.SEND, ConnectableType.REMOTE_INPUT_PORT, sslContext, scheduler); |
| |
| if (descriptor.getConcurrentlySchedulableTaskCount() != null) { |
| port.setMaxConcurrentTasks(descriptor.getConcurrentlySchedulableTaskCount()); |
| } |
| if (descriptor.getUseCompression() != null) { |
| port.setUseCompression(descriptor.getUseCompression()); |
| } |
| |
| inputPorts.put(descriptor.getId(), port); |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| @Override |
| public RemoteGroupPort getOutputPort(final String portIdentifier) { |
| readLock.lock(); |
| try { |
| if (requireNonNull(portIdentifier).startsWith(id + "-")) { |
| return outputPorts.get(portIdentifier.substring(id.length() + 1)); |
| } else { |
| return outputPorts.get(portIdentifier); |
| } |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| /** |
| * @return a set of {@link RemoteGroupPort}s used for receiving FlowFiles from the remote instance |
| */ |
| @Override |
| public Set<RemoteGroupPort> getOutputPorts() { |
| readLock.lock(); |
| try { |
| final Set<RemoteGroupPort> set = new HashSet<>(); |
| set.addAll(outputPorts.values()); |
| return set; |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public String toString() { |
| return "RemoteProcessGroup[" + targetUri + "]"; |
| } |
| |
| @Override |
| public ProcessGroupCounts getCounts() { |
| readLock.lock(); |
| try { |
| return counts; |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| private void setCounts(final ProcessGroupCounts counts) { |
| writeLock.lock(); |
| try { |
| this.counts = counts; |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| private ProcessGroup getRootGroup() { |
| return getRootGroup(getProcessGroup()); |
| } |
| |
| private ProcessGroup getRootGroup(final ProcessGroup context) { |
| final ProcessGroup parent = context.getParent(); |
| return parent == null ? context : getRootGroup(parent); |
| } |
| |
| private void refreshFlowContentsFromLocal() { |
| final ProcessGroup rootGroup = getRootGroup(); |
| setName(rootGroup.getName()); |
| setTargetId(rootGroup.getIdentifier()); |
| setComments(rootGroup.getComments()); |
| setCounts(rootGroup.getCounts()); |
| |
| final Set<RemoteProcessGroupPortDescriptor> convertedInputPorts = new HashSet<>(); |
| for (final Port port : rootGroup.getInputPorts()) { |
| convertedInputPorts.add(convertPortToRemotePortDescriptor(port)); |
| } |
| |
| final Set<RemoteProcessGroupPortDescriptor> convertedOutputPorts = new HashSet<>(); |
| for (final Port port : rootGroup.getOutputPorts()) { |
| convertedOutputPorts.add(convertPortToRemotePortDescriptor(port)); |
| } |
| |
| setInputPorts(convertedInputPorts); |
| setOutputPorts(convertedOutputPorts); |
| |
| writeLock.lock(); |
| try { |
| final NiFiProperties props = NiFiProperties.getInstance(); |
| this.destinationSecure = props.isSiteToSiteSecure(); |
| this.listeningPort = props.getRemoteInputPort(); |
| this.listeningHttpPort = props.getRemoteInputHttpPort(); |
| |
| refreshContentsTimestamp = System.currentTimeMillis(); |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| @Override |
| public Date getLastRefreshTime() { |
| readLock.lock(); |
| try { |
| return refreshContentsTimestamp == null ? null : new Date(refreshContentsTimestamp); |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public void refreshFlowContents() throws CommunicationsException { |
| // check to see if we know whether we pointing to a local cluster or not - |
| // we must know this before we attempt to update since querying the cluster |
| // will cause a deadlock if we are in the context of another replicated request |
| if (pointsToCluster.get() == null) { |
| return; |
| } |
| |
| if (pointsToCluster.get()) { |
| refreshFlowContentsFromLocal(); |
| return; |
| } |
| |
| if (targetIsUnreachable.get()) { |
| return; |
| } |
| |
| try { |
| // perform the request |
| final ControllerDTO dto; |
| try ( |
| final SiteToSiteRestApiClient apiClient = getSiteToSiteRestApiClient(); |
| ){ |
| dto = apiClient.getController(); |
| } catch (IOException e) { |
| writeLock.lock(); |
| try { |
| for (final Iterator<StandardRemoteGroupPort> iter = inputPorts.values().iterator(); iter.hasNext();) { |
| final StandardRemoteGroupPort inputPort = iter.next(); |
| if (!inputPort.hasIncomingConnection()) { |
| iter.remove(); |
| } |
| } |
| |
| for (final Iterator<StandardRemoteGroupPort> iter = outputPorts.values().iterator(); iter.hasNext();) { |
| final StandardRemoteGroupPort outputPort = iter.next(); |
| if (outputPort.getConnections().isEmpty()) { |
| iter.remove(); |
| } |
| } |
| } finally { |
| writeLock.unlock(); |
| } |
| |
| throw new CommunicationsException("Unable to communicate with Remote NiFi at URI " + apiUri + " due to: " + e.getMessage()); |
| } |
| |
| writeLock.lock(); |
| try { |
| if (dto.getInputPorts() != null) { |
| setInputPorts(convertRemotePort(dto.getInputPorts())); |
| } |
| if (dto.getOutputPorts() != null) { |
| setOutputPorts(convertRemotePort(dto.getOutputPorts())); |
| } |
| |
| // set the controller details |
| setTargetId(dto.getId()); |
| setName(dto.getName()); |
| setComments(dto.getComments()); |
| |
| // get the component counts |
| int inputPortCount = 0; |
| if (dto.getInputPortCount() != null) { |
| inputPortCount = dto.getInputPortCount(); |
| } |
| int outputPortCount = 0; |
| if (dto.getOutputPortCount() != null) { |
| outputPortCount = dto.getOutputPortCount(); |
| } |
| int runningCount = 0; |
| if (dto.getRunningCount() != null) { |
| runningCount = dto.getRunningCount(); |
| } |
| int stoppedCount = 0; |
| if (dto.getStoppedCount() != null) { |
| stoppedCount = dto.getStoppedCount(); |
| } |
| int invalidCount = 0; |
| if (dto.getInvalidCount() != null) { |
| invalidCount = dto.getInvalidCount(); |
| } |
| int disabledCount = 0; |
| if (dto.getDisabledCount() != null) { |
| disabledCount = dto.getDisabledCount(); |
| } |
| |
| int activeRemotePortCount = 0; |
| if (dto.getActiveRemotePortCount() != null) { |
| activeRemotePortCount = dto.getActiveRemotePortCount(); |
| } |
| |
| int inactiveRemotePortCount = 0; |
| if (dto.getInactiveRemotePortCount() != null) { |
| inactiveRemotePortCount = dto.getInactiveRemotePortCount(); |
| } |
| |
| this.listeningPort = dto.getRemoteSiteListeningPort(); |
| this.listeningHttpPort = dto.getRemoteSiteHttpListeningPort(); |
| this.destinationSecure = dto.isSiteToSiteSecure(); |
| |
| final ProcessGroupCounts newCounts = new ProcessGroupCounts(inputPortCount, outputPortCount, |
| runningCount, stoppedCount, invalidCount, disabledCount, activeRemotePortCount, inactiveRemotePortCount); |
| setCounts(newCounts); |
| this.refreshContentsTimestamp = System.currentTimeMillis(); |
| } finally { |
| writeLock.unlock(); |
| } |
| } catch (final ClientHandlerException | UniformInterfaceException e) { |
| throw new CommunicationsException(e); |
| } |
| } |
| |
| private SiteToSiteRestApiClient getSiteToSiteRestApiClient() { |
| SiteToSiteRestApiClient apiClient = new SiteToSiteRestApiClient(sslContext, new HttpProxy(proxyHost, proxyPort, proxyUser, proxyPassword)); |
| apiClient.setBaseUrl(apiUri.toString()); |
| apiClient.setConnectTimeoutMillis(getCommunicationsTimeout(TimeUnit.MILLISECONDS)); |
| apiClient.setReadTimeoutMillis(getCommunicationsTimeout(TimeUnit.MILLISECONDS)); |
| return apiClient; |
| } |
| |
| /** |
| * Converts a set of ports into a set of remote process group ports. |
| * |
| * @param ports to convert |
| * @return descriptors of ports |
| */ |
| private Set<RemoteProcessGroupPortDescriptor> convertRemotePort(final Set<PortDTO> ports) { |
| Set<RemoteProcessGroupPortDescriptor> remotePorts = null; |
| if (ports != null) { |
| remotePorts = new LinkedHashSet<>(ports.size()); |
| for (final PortDTO port : ports) { |
| final StandardRemoteProcessGroupPortDescriptor descriptor = new StandardRemoteProcessGroupPortDescriptor(); |
| final ScheduledState scheduledState = ScheduledState.valueOf(port.getState()); |
| descriptor.setId(port.getId()); |
| descriptor.setName(port.getName()); |
| descriptor.setComments(port.getComments()); |
| descriptor.setTargetRunning(ScheduledState.RUNNING.equals(scheduledState)); |
| remotePorts.add(descriptor); |
| } |
| } |
| return remotePorts; |
| } |
| |
| private RemoteProcessGroupPortDescriptor convertPortToRemotePortDescriptor(final Port port) { |
| final StandardRemoteProcessGroupPortDescriptor descriptor = new StandardRemoteProcessGroupPortDescriptor(); |
| descriptor.setComments(port.getComments()); |
| descriptor.setExists(true); |
| descriptor.setGroupId(port.getProcessGroup().getIdentifier()); |
| descriptor.setId(port.getIdentifier()); |
| descriptor.setName(port.getName()); |
| descriptor.setTargetRunning(port.isRunning()); |
| return descriptor; |
| } |
| |
| @Override |
| public boolean isTransmitting() { |
| return transmitting.get(); |
| } |
| |
| @Override |
| public void startTransmitting() { |
| writeLock.lock(); |
| try { |
| verifyCanStartTransmitting(); |
| |
| for (final Port port : getInputPorts()) { |
| // if port is not valid, don't start it because it will never become valid. |
| // Validation is based on connections and whether or not the remote target exists. |
| if (port.isValid() && port.hasIncomingConnection()) { |
| scheduler.startPort(port); |
| } |
| } |
| |
| for (final Port port : getOutputPorts()) { |
| if (port.isValid() && !port.getConnections().isEmpty()) { |
| scheduler.startPort(port); |
| } |
| } |
| |
| transmitting.set(true); |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| @Override |
| public void startTransmitting(final RemoteGroupPort port) { |
| writeLock.lock(); |
| try { |
| if (!inputPorts.containsValue(port) && !outputPorts.containsValue(port)) { |
| throw new IllegalArgumentException("Port does not belong to this Remote Process Group"); |
| } |
| |
| port.verifyCanStart(); |
| |
| scheduler.startPort(port); |
| |
| transmitting.set(true); |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| @Override |
| public void stopTransmitting() { |
| writeLock.lock(); |
| try { |
| verifyCanStopTransmitting(); |
| |
| for (final RemoteGroupPort port : getInputPorts()) { |
| scheduler.stopPort(port); |
| } |
| |
| for (final RemoteGroupPort port : getOutputPorts()) { |
| scheduler.stopPort(port); |
| } |
| |
| // Wait for the ports to stop |
| for (final RemoteGroupPort port : getInputPorts()) { |
| while (port.isRunning()) { |
| try { |
| Thread.sleep(50L); |
| } catch (final InterruptedException e) { |
| } |
| } |
| } |
| |
| for (final RemoteGroupPort port : getOutputPorts()) { |
| while (port.isRunning()) { |
| try { |
| Thread.sleep(50L); |
| } catch (final InterruptedException e) { |
| } |
| } |
| } |
| |
| transmitting.set(false); |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| @Override |
| public void stopTransmitting(final RemoteGroupPort port) { |
| writeLock.lock(); |
| try { |
| if (!inputPorts.containsValue(port) && !outputPorts.containsValue(port)) { |
| throw new IllegalArgumentException("Port does not belong to this Remote Process Group"); |
| } |
| |
| port.verifyCanStop(); |
| |
| scheduler.stopPort(port); |
| |
| // Wait for the port to stop |
| while (port.isRunning()) { |
| try { |
| Thread.sleep(50L); |
| } catch (final InterruptedException e) { |
| } |
| } |
| |
| // Determine if any other ports are still running |
| boolean stillTransmitting = false; |
| for (final Port inputPort : getInputPorts()) { |
| if (inputPort.isRunning()) { |
| stillTransmitting = true; |
| break; |
| } |
| } |
| |
| if (!stillTransmitting) { |
| for (final Port outputPort : getOutputPorts()) { |
| if (outputPort.isRunning()) { |
| stillTransmitting = true; |
| break; |
| } |
| } |
| } |
| |
| transmitting.set(stillTransmitting); |
| } finally { |
| writeLock.unlock(); |
| } |
| } |
| |
| @Override |
| public boolean isSecure() throws CommunicationsException { |
| Boolean secure; |
| readLock.lock(); |
| try { |
| secure = this.destinationSecure; |
| if (secure != null) { |
| return secure; |
| } |
| } finally { |
| readLock.unlock(); |
| } |
| |
| refreshFlowContents(); |
| |
| readLock.lock(); |
| try { |
| secure = this.destinationSecure; |
| if (secure == null) { |
| throw new CommunicationsException("Unable to determine whether or not site-to-site communications with peer should be secure"); |
| } |
| |
| return secure; |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public Boolean getSecureFlag() { |
| readLock.lock(); |
| try { |
| return this.destinationSecure; |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public boolean isSiteToSiteEnabled() { |
| readLock.lock(); |
| try { |
| return (this.listeningPort != null || this.listeningHttpPort != null); |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public EventReporter getEventReporter() { |
| return eventReporter; |
| } |
| |
| private class InitializationTask implements Runnable { |
| |
| @Override |
| public void run() { |
| try (final SiteToSiteRestApiClient apiClient = getSiteToSiteRestApiClient()){ |
| try { |
| final ControllerDTO dto = apiClient.getController(); |
| |
| if (dto.getRemoteSiteListeningPort() == null && SiteToSiteTransportProtocol.RAW.equals(transportProtocol)) { |
| authorizationIssue = "Remote instance is not configured to allow RAW Site-to-Site communications at this time."; |
| } else if (dto.getRemoteSiteHttpListeningPort() == null && SiteToSiteTransportProtocol.HTTP.equals(transportProtocol)) { |
| authorizationIssue = "Remote instance is not configured to allow HTTP Site-to-Site communications at this time."; |
| } else { |
| authorizationIssue = null; |
| } |
| |
| writeLock.lock(); |
| try { |
| listeningPort = dto.getRemoteSiteListeningPort(); |
| listeningHttpPort = dto.getRemoteSiteHttpListeningPort(); |
| destinationSecure = dto.isSiteToSiteSecure(); |
| } finally { |
| writeLock.unlock(); |
| } |
| |
| final String remoteInstanceId = dto.getInstanceId(); |
| final boolean isPointingToCluster = flowController.getInstanceId().equals(remoteInstanceId); |
| pointsToCluster.set(isPointingToCluster); |
| |
| } catch (SiteToSiteRestApiClient.HttpGetFailedException e) { |
| |
| if (e.getResponseCode() == UNAUTHORIZED_STATUS_CODE) { |
| // TODO: implement registration request |
| /* |
| try { |
| final ClientResponse requestAccountResponse = utils.issueRegistrationRequest(apiUri.toString()); |
| if (Response.Status.Family.SUCCESSFUL.equals(requestAccountResponse.getStatusInfo().getFamily())) { |
| logger.info("{} Issued a Request to communicate with remote instance", this); |
| } else { |
| logger.error("{} Failed to request account: got unexpected response code of {}:{}", new Object[]{ |
| this, requestAccountResponse.getStatus(), requestAccountResponse.getStatusInfo().getReasonPhrase()}); |
| } |
| } catch (final Exception e) { |
| logger.error("{} Failed to request account due to {}", this, e.toString()); |
| if (logger.isDebugEnabled()) { |
| logger.error("", e); |
| } |
| } |
| */ |
| authorizationIssue = e.getDescription(); |
| |
| } else if (e.getResponseCode() == FORBIDDEN_STATUS_CODE) { |
| authorizationIssue = e.getDescription(); |
| } else { |
| final String message = e.getDescription(); |
| logger.warn("{} When communicating with remote instance, got unexpected result. {}", |
| new Object[]{this, e.getMessage()}); |
| authorizationIssue = "Unable to determine Site-to-Site availability."; |
| } |
| } |
| |
| } catch (final Exception e) { |
| logger.warn(String.format("Unable to connect to %s due to %s", StandardRemoteProcessGroup.this, e)); |
| getEventReporter().reportEvent(Severity.WARNING, "Site to Site", String.format("Unable to connect to %s due to %s", |
| StandardRemoteProcessGroup.this.getTargetUri().toString(), e)); |
| } |
| } |
| } |
| |
| @Override |
| public void setYieldDuration(final String yieldDuration) { |
| // verify the syntax |
| if (!FormatUtils.TIME_DURATION_PATTERN.matcher(yieldDuration).matches()) { |
| throw new IllegalArgumentException("Improperly formatted Time Period; should be of syntax <number> <unit> where " |
| + "<number> is a positive integer and unit is one of the valid Time Units, such as nanos, millis, sec, min, hour, day"); |
| } |
| |
| this.yieldDuration = yieldDuration; |
| } |
| |
| @Override |
| public String getYieldDuration() { |
| return yieldDuration; |
| } |
| |
| @Override |
| public void verifyCanDelete() { |
| verifyCanDelete(false); |
| } |
| |
| @Override |
| public void verifyCanDelete(final boolean ignoreConnections) { |
| readLock.lock(); |
| try { |
| if (isTransmitting()) { |
| throw new IllegalStateException(this.getIdentifier() + " is transmitting"); |
| } |
| |
| for (final Port port : inputPorts.values()) { |
| if (!ignoreConnections && port.hasIncomingConnection()) { |
| throw new IllegalStateException(this.getIdentifier() + " is the destination of another component"); |
| } |
| if (port.isRunning()) { |
| throw new IllegalStateException(this.getIdentifier() + " has running Port: " + port.getIdentifier()); |
| } |
| } |
| |
| for (final Port port : outputPorts.values()) { |
| if (!ignoreConnections) { |
| for (final Connection connection : port.getConnections()) { |
| connection.verifyCanDelete(); |
| } |
| } |
| |
| if (port.isRunning()) { |
| throw new IllegalStateException(this.getIdentifier() + " has running Port: " + port.getIdentifier()); |
| } |
| } |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public void verifyCanStartTransmitting() { |
| readLock.lock(); |
| try { |
| if (isTransmitting()) { |
| throw new IllegalStateException(this.getIdentifier() + " is already transmitting"); |
| } |
| |
| for (final StandardRemoteGroupPort port : inputPorts.values()) { |
| if (port.isRunning()) { |
| throw new IllegalStateException(this.getIdentifier() + " has running Port: " + port.getIdentifier()); |
| } |
| |
| if (port.hasIncomingConnection() && !port.getTargetExists()) { |
| throw new IllegalStateException(this.getIdentifier() + " has a Connection to Port " + port.getIdentifier() + ", but that Port no longer exists on the remote system"); |
| } |
| } |
| |
| for (final StandardRemoteGroupPort port : outputPorts.values()) { |
| if (port.isRunning()) { |
| throw new IllegalStateException(this.getIdentifier() + " has running Port: " + port.getIdentifier()); |
| } |
| |
| if (!port.getConnections().isEmpty() && !port.getTargetExists()) { |
| throw new IllegalStateException(this.getIdentifier() + " has a Connection to Port " + port.getIdentifier() + ", but that Port no longer exists on the remote system"); |
| } |
| } |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| @Override |
| public void verifyCanStopTransmitting() { |
| if (!isTransmitting()) { |
| throw new IllegalStateException(this.getIdentifier() + " is not transmitting"); |
| } |
| } |
| |
| @Override |
| public void verifyCanUpdate() { |
| readLock.lock(); |
| try { |
| if (isTransmitting()) { |
| throw new IllegalStateException(this.getIdentifier() + " is currently transmitting"); |
| } |
| |
| for (final Port port : inputPorts.values()) { |
| if (port.isRunning()) { |
| throw new IllegalStateException(this.getIdentifier() + " has running Port: " + port.getIdentifier()); |
| } |
| } |
| |
| for (final Port port : outputPorts.values()) { |
| if (port.isRunning()) { |
| throw new IllegalStateException(this.getIdentifier() + " has running Port: " + port.getIdentifier()); |
| } |
| } |
| } finally { |
| readLock.unlock(); |
| } |
| } |
| |
| private File getPeerPersistenceFile() { |
| final File stateDir = NiFiProperties.getInstance().getPersistentStateDirectory(); |
| return new File(stateDir, getIdentifier() + ".peers"); |
| } |
| |
| } |
