Support HTTP(S) as a transport mechanism for Site-to-Site

Goals

• Exchange flow files between two NiFi environments using Site-to-Site via HTTP(S) 

Background and strategic fit

Some environments only allow network communication through HTTP(S) port, typically with multi-datacenter deployments. In order to exchange data between NiFi environments using Site-to-Site in such restricted deployments, we should add HTTP(S) as a transport protocol for Site-to-Site.

Assumptions

Requirements

User interaction and design

This proposal add new UI input in Remote Process Group configuration dialog as the following image:

• Transport Protocol: defaults to RAW
• HTTP Proxy server hostname: Specify the proxy server's hostname to use. If not specified, HTTP traffics are sent directly to the target NiFi instance.
• HTTP Proxy server port: Specify the proxy server's port number, optional. If not specified, default port 80 will be used.
• HTTP Proxy user: Specify an user name to connect to the proxy server, optional.
• HTTP Proxy password: Specify an user password to connect to the proxy server, optional.

nifi.properties

This proposal uses following configurations in nifi.properties :

Deployment examples

The following diagrams illustrate some deployment options to describe key features, it isn't meant to limit the deployment patterns. Although, following diagrams only show a single Site-to-Site client server,  the client can be one of a NiFi node within a NiFi cluster. Site-to-Site supports cluster to cluster data transport.

To Standalone NiFi : Socket

This is an existing deployment option using RAW Socket, here to describe the difference between HTTP Site-to-Site. It also supports secure communication, and NiFi cluster. With RAW Socket, it first retrieves the remote NiFi site info by sending a HTTP request to /nifi-api/site-to-site. After that, it uses Socket networking to exchange data.

To Standalone NiFi : HTTP

If HTTP is used for Transport Protocol, then all communications between Site-to-Site client and the remote NiFi instance are done with HTTP protocol.

To Standalone NiFi : HTTPS

Network traffic to a remote NiFi can be secured by setting nifi.remote.input.secure to true. When it's true, a remote NiFi instance is only accessible with HTTPS protocol.

To Standalone NiFi : HTTP using Proxy

 If a remote NiFi instance is behind a firewall which only expose http port to a Proxy Server, its Site-to-Site client can be configured as shown in this diagram to use that proxy server.

To NiFi Cluster : HTTP

If the target NiFi is a cluster, its client chooses which NiFi node to transport data based on Peer Selection each time when it transfers data, for example if the Site-to-Site client component is a Remote Process Group, it does peer selection when it's scheduled.

In order to allow a NiFi cluster to use HTTPS for Site-to-Site, but HTTP for communications within a cluster, siteToSiteHttpApiPort is added to NodeIdentifier. Because the existing apiPort is determined by if cluster protocol manager to node is secure.

Peer Selection

If the remote NiFi forms a cluster, a Site-to-Site client has to determine which NiFi node to transfer data to/from, let's call the decision making process as 'Peer Selection'. There're two aspects for that, Flow file count, and Port Status. A Site-to-Site client does Peer Selection when startTransaction method is called.

Based on Flow file count

The first step is done before connecting any particular peer. Site-to-Site client uses Peer statuses that is retrieved from NCM. Peer statuses contains the number of flow files queued in each peer, and it is used to calculate how often a peer will be selected.

The frequency calculation logic is implemented in PeerSelector.java and both Raw socket and HTTP transport protocol uses it.

PeerSelector provides following load balancing characteristic when its getNextPeerStatus method called: "The target port which has more data in its queue will receive less than others for sending flow-files, and it will be pulled more often than others for receiving."

Based on Port Status

The second step is done at handshaking phase with a peer that is selected by PeerSelector.getNextPeerStatus().

In a clustered NiFi, the state of a Data Flow is synchronized. It means that if there is a Port named 'input-1', the port exists on every NiFi node with the same name and port identifier. Also If the port is stopped, the port is stopped on every NiFi node. So, in case of receiving UNKNOWN_PORT and PORT_NOT_IN_VALID_STATE, Site-to-Site client stops looking for another peer, because the same response will be returned from every NiFi node. Site-to-Site penalizes the peer and raises exception in these cases.

If a Site-to-Site client receives PORTS_DESTINATION_FULL, it only means that the port running on a particular NiFi node is full. So the client penalizes the peer, but continues looking for another peer. If all peers destination are full, then Site-to-Site client returns null as a return value for startTransaction method.

REST endpoints

Following REST endpoints will be added by this proposal:

• /site-to-site/
  • GET: Returns required information of Site-to-Site for the source NiFi environment. Representing Controller of target NiFi environment.
• /site-to-site/peers/
  • GET: Returns available peers of this NiFi environment.
• /site-to-site/input-ports/{portId}/transactions/
  • POST: Initiate new transaction to send data from source to target NiFi. A new transaction id is published and returned.
• /site-to-site/input-ports/{portId}/transactions/{transactionId}
  • PUT: Extends the transaction's TTL, used to let server know the client still working
  • DELETE: Commit the transaction which is held on server side.
• /site-to-site/input-ports/{portId}/transactions/{transactionId}/flow-files
  • POST: Transfer data from source to target NiFi. The transaction will be held on server side instead of commit it immediately, in order to provide 2-phase style commit. Returns Checksum calculated on server side.
• /site-to-site/output-ports/{portId}/transactions/
  • POST: Initiate new transaction to receive data from target to source NiFi. A new transaction id is published and returned.
• /site-to-site/output-ports/{portId}/transactions/{transactionId}
  • PUT: Extends the transaction's TTL, used to let server know the client still working
  • DELETE: Commit the transaction which is held on server side. Client sends a Checksum calculated on client side.
• /site-to-site/output-ports/{portId}/transactions/{transactionId}/flow-files
  • GET: Transfer data from target to source NiFi.  The transaction will be held on server side instead of commit it immediately, in order to provide 2-phase style commit.

Refer sequence diagrams 'REST interactions and scenarios' below for the details of how these REST endpoints are used.

Site-to-Site specific HTTP Headers

Following HTTP Headers are used in Site-to-Site REST requests:

REST interaction sequence diagrams

Here are sequence diagrams describing complete HTTP request/response sequence of successful scenario for both input-ports and output-ports. Other semi-normal and error cases are described in Interaction Scenarios.

The A_component in the diagrams is a component which uses SiteToSiteClient class, such as:

• NiFiReceiver for Apache Spark
• NiFiBolt and NiFiSpoutReceiver for Apache Storm
• StandardRemoteGroupPort, this is the Remote Process Group processor in a NiFi data flow

input-ports/

The input-ports endpoint is used for sending data from source NiFi to a remote target NiFi. When a transaction is created, a HTTP POST request is initiated with a flow-file url within a transactionId. While there is more data packet to be sent, the same HTTP POST request is used, to send data in streaming manner. When the client finishes sending all data packets to output stream, it flushes the stream. Finally, the client send a DELETE request to transactionId url to complete the transaction.

  

output-ports/

The output-ports endpoint is used for receiving data from a remote target NiFi to source NiFi. When a transaction is created, a HTTP GET request is initiated with a flow-file url within a transactionId. While there is more data packet to be received, the same HTTP GET request is used, to receive data in streaming manner. When the client finishes consuming all data packets from input stream, and confirm() method is called, the client send a DELETE request with a Checksum calculated at client side to complete the transaction.

The complete() method doesn't do anything other than update state to TRANSACTION_COMPLETED. 

REST interactions and scenarios

input-ports/

output-ports/

Questions

Below is a list of questions to be addressed as a result of this requirements document:

Not Doing

• Unable to render Jira issues macro, execution error.
• Unable to render Jira issues macro, execution error.