Page History

...

User interaction and design

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

Image AddedImage Added

• 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.

TODO: add deployment diagram

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.

...

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}{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.

...

actor A_component
actor HttpClient
actor HttpClientTransaction
actor SiteToSiteRestApiUtilSiteToSiteRestApiClient
actor SiteToSiteResource

' comment: initialize
A_component -> HttpClient: createTransaction
HttpClient -> SiteToSiteRestApiUtilSiteToSiteRestApiClient: initiateTransaction
SiteToSiteRestApiUtilSiteToSiteRestApiClient -> SiteToSiteResource: POST /site-to-site/input-ports/{portId}/transactions
SiteToSiteRestApiUtilSiteToSiteRestApiClient <-- SiteToSiteResource: transactionUrl, transactionProtocolVersion
HttpClient <-- SiteToSiteRestApiUtilSiteToSiteRestApiClient
HttpClient -> HttpClientTransaction: new
HttpClientTransaction -> HttpClientTransaction: state = TRANSACTION_STARTED
HttpClientTransaction -> SiteToSiteRestApiUtilSiteToSiteRestApiClient: openConnectionForSend
SiteToSiteRestApiUtilSiteToSiteRestApiClient -> SiteToSiteResource: POST /site-to-site/input-ports/{portId}/transactions/{transactionId}/flow-files
HttpClient <-- HttpClientTransaction
A_component <-- HttpClient: Transaction

'comment: receive
alt while there is data packet to send
    A_component -> HttpClientTransaction: send
	HttpClientTransaction -> SiteToSiteResource: writes data to outputstream
	HttpClientTransaction -> HttpClientTransaction: state = DATA_EXCHANGED
	A_component <-- HttpClientTransaction
end
 
'comment: confirm
A_component -> HttpClientTransaction: confirm
HttpClientTransaction -> SiteToSiteRestApiUtilSiteToSiteRestApiClient: finishTransferFlowFiles
SiteToSiteRestApiUtilSiteToSiteRestApiClient <-- SiteToSiteResource: 202 Accepted: returns serverChecksum
HttpClientTransaction -> HttpClientTransaction: validate server Checksum
HttpClientTransaction -> HttpClientTransaction: state = TRANSACTION_CONFIRMED
A_component <-- HttpClientTransaction

'comment: complete
A_component -> HttpClientTransaction: complete
HttpClientTransaction -> SiteToSiteRestApiUtilSiteToSiteRestApiClient: commitTransferFlowFiles
SiteToSiteRestApiUtilSiteToSiteRestApiClient -> SiteToSiteResource: DELETE /site-to-site/input-ports/{portId}/transactions/{transactionId}
SiteToSiteRestApiUtilSiteToSiteRestApiClient <-- SiteToSiteResource: 200 OK
HttpClientTransaction <-- SiteToSiteRestApiUtilSiteToSiteRestApiClient
HttpClientTransaction -> HttpClientTransaction: state = TRANSACTION_COMPLETED
A_component <-- HttpClientTransaction

...

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

actor A_component
actor HttpClient
actor HttpClientTransaction
actor SiteToSiteRestApiUtilSiteToSiteRestApiClient
actor SiteToSiteResource

' comment: initialize
A_component -> HttpClient: createTransaction
HttpClient -> SiteToSiteRestApiUtilSiteToSiteRestApiClient: initiateTransaction
SiteToSiteRestApiUtilSiteToSiteRestApiClient -> SiteToSiteResource: POST /site-to-site/output-ports/{portId}/transactions
SiteToSiteRestApiUtilSiteToSiteRestApiClient <-- SiteToSiteResource: transactionUrl, transactionProtocolVersion
HttpClient <-- SiteToSiteRestApiUtilSiteToSiteRestApiClient
HttpClient -> HttpClientTransaction: new
HttpClientTransaction -> HttpClientTransaction: state = TRANSACTION_STARTED
HttpClientTransaction -> SiteToSiteRestApiUtilSiteToSiteRestApiClient: openConnectionForReceive
SiteToSiteRestApiUtilSiteToSiteRestApiClient -> SiteToSiteResource: GET /site-to-site/output-ports/{portId}/transactions/{transactionId}/flow-files
SiteToSiteRestApiUtilSiteToSiteRestApiClient <-- SiteToSiteResource: 202 Accepted
HttpClientTransaction <-- SiteToSiteRestApiUtilSiteToSiteRestApiClient
HttpClient <-- HttpClientTransaction
A_component <-- HttpClient: Transaction

'comment: receive
alt while there is data packet to receive
    A_component -> HttpClientTransaction: receive
	HttpClientTransaction <-- SiteToSiteResource: read from inputstream
	HttpClientTransaction -> HttpClientTransaction: state = DATA_EXCHANGED
	A_component <-- HttpClientTransaction: data packet
end
 
'comment: confirm
A_component -> HttpClientTransaction: confirm
HttpClientTransaction -> SiteToSiteRestApiUtilSiteToSiteRestApiClient: commitReceivingFlowFiles(checksum)
SiteToSiteRestApiUtilSiteToSiteRestApiClient -> SiteToSiteResource: DELETE /site-to-site/output-ports/{portId}/transactions/{transactionId}
SiteToSiteResource -> SiteToSiteResource: validate client Checksum
SiteToSiteRestApiUtilSiteToSiteRestApiClient <-- SiteToSiteResource: 200 OK
HttpClientTransaction <-- SiteToSiteRestApiUtilSiteToSiteRestApiClient
HttpClientTransaction -> HttpClientTransaction: state = TRANSACTION_CONFIRMED
A_component <-- HttpClientTransaction
 
'comment: complete
A_component -> HttpClientTransaction: complete
HttpClientTransaction -> HttpClientTransaction: state = TRANSACTION_COMPLETED
A_component <-- HttpClientTransaction

...