Basics of WebSocket protocol WebSocket is a fully duplex and bi-directional protocol. The most interesting characteristics (at least for me) is that there is no difference or roll such as server or client for each peer after those established a WebSocket connection. Each peer can send messages at will. However, a connection is always initiated by a WebSocket client. For example: A client sends a HTTP request to a server (URL e.g. ws://example.com/web-socket). The server accepts the request and upgrade HTTP protocol to WebSocket protocol. (at this point, each peer can send or receive messages asynchronously) The client can send a message to the server The server receives it The server can send a message to the client The client receives it I wondered, should NiFi be a client? or a server? Then decided to support both. NiFi can be not only a WebSocket client, but also a WebSocket server! How it works? Modules Since the protocol itself is more complex than (or simply different from) other protocols, it was tough to put it into NiFi data flow processing model. Some might feel it is too complex to use, so let me try to explain how it works. As shown in the diagram above, it is divided into three modules described below, for extensibility and testability. Each of these are individual NAR. Both nifi-websocket-processors-nar and nifi-websocket-services-jetty-nar have NAR dependency to nifi-websocket-services-api-nar . nifi-websocket-services-api WebSocketClientService: An Interface acts as a WebSocket client. WebSocketServerService: An Interface acts as a WebSocket server. Features: WebSocket events: Both Client and Server services adopt event-driven processing so that WebSocket processors can react to. connected : Fired when a remote client connects with WebSocketServerService, or a WebSocketClientService connects with a remote WebSocket server. text message : Fired when it receives text WebSocket message. binary message : Fired when it receives binary WebSocket message. Send message: Provides methods to send text and binary WebSocket message to a connected remote peer. Multiple endpoints: It registers processors to endpoints. WebSocketServerService uses an URI path as an endpoint. For example, the same WebSocket server instance can manage two WebSocket endpoints, such as ws://hostname:listen-port/endpoint-1 and ws://hostname:listen-port/endpoint-2 . Connected sessions are manages separately within each endpoints. Likewise, WebSocketClientService uses a clientId to distinguish endpoints. Multiple WebSocket client instances can share the same WebSocketClientService instance. nifi-websocket-services-jetty This module contains actual implementation of nifi-websocket-services-api using Jetty. Features: Plain WebSocket (ws://), and Secure WebSocket (wss://) protocols are supported. Uses SSLContextService to refer Java keystore and truststore for secure communication. nifi-websocket-processors In order to use these functionalities in a NiFi data flow, we need to put it on a canvas as Processors. ConnectWebSocket and ListenWebSocket : Work as WebSocket gateways. These processors are registered to WebSocketServer and receives WebSocket events described earlier. When those events are fired, it will be converted to NiFi FlowFile, then sent to relationships accordingly. There are three relationships, connected , text message and binary message . ConnectWebSocket uses WebSocketClientService to actively connect to a remote WebSocket endpoint, while ListenWebSocket uses WebSocketServerService to wait passively for remote WebSocket clients to connect. PutWebSocket : This processor can use with both ConnectWebSocket and ListenWebSocket, since there is no distinction after connection is made. It sends a WebSocket message using an incoming FlowFile content as message payload. How can I use? Use Cases Ok, enough descriptions, let’s see how can we use these component in NiFi data flow! NiFi as a client to talk with a remote WebSocket server To use NiFi as a WebSocket client, we need a WebSocketClientService. To add the service: Click the gear icon on Operate palette Click the plus sign Enter ‘WebSocket’ tag to search the ControllerService Click the edit icon of the JettyWebSocketClient controller service Then, the service needs to be configured as follows: Set ws://echo.websocket.org to WebSocket URI. This URI is publicly available to test WebSocket client. It simply echoes back the message it receives. Click the enable icon, and the service is ready! Next, let’s setup the data flow using processors: ConnectWebSocket: Uses the JettyWebSocketClientService added earlier. connected and text message are routed to ReplaceText. binary message is terminated here because we don’t use it in this example. ReplaceText: Add some prefix to update text content PutWebSocket: This processor sends messages to the remote WebSocket server. Don’t forget to set Run Schedule longer than default like 3 sec, otherwise this ping-pong loop goes too fast like DoS attack… UpdateAttribute: This is the end of data flow, and keep it stopped so that we can accumulate the FlowFiles in the relationship to check the contents By right click the success relationship, the queued FlowFiles can be seen. Its file size is growing as ReplaceText prepend text each time. NiFi as a server to talk with a remote WebSocket client Once you get the idea, setting up NiFi as a WebSocket server is easy, almost the same! We just need to use JettyWebSocketServer controller service instead, and set Listen Port : Then, replace the ConnectWebSocket processor with ListenWebSocket processor, and specify the Server URL Path that you want to receive WebSocket requests: Then, open websocket.org echo from a web browser, set location as ws://localhost:9001/server-demo , and click Connect, then Send. NiFi will echo back the message! Secure WebSocket connection To use secure WebSocket connection, we need another controller service, StandardSSLContextService . Then use it from JettyWebSocketClient or JettyWebSocketServer. The URL should use wss:// protocol for secure connection. Scalability When NiFi is deployed as a cluster for scalability, we can run these WebSocket component on every node. To distribute loads when you use NiFi as WebSocket server, you will need a Load Balancer such as HAProxy in front of NiFi cluster. Please also refer the previous post, NiFi Cluster and Load Balancer. Summary In this post, I covered the basic usage of these WebSocket controller services and processors. Since the WebSocket gateway processors (ConnectWebSocket/ListenWebSocket) and PutWebSocket can be used separately, we can design more complex and interesting data flow by putting more flow in between. I hope this post will help you to get started with NiFi WebSocket support, and create some interesting data flows! Thanks for reading! ... View more

Why do we need a Load Balancer for NiFi cluster? The easiest way to start using NiFi is deploying it as a standalone NiFi instance. However, when you need more throughput, NiFi can form a cluster to distribute load among multiple NiFi nodes. From 1.0, thanks to the Zero Master Clustering architecture, we can access NiFi Web UI via any node in a cluster. Although it’s more reliable, it can be unclear which endpoint to specify for things like Remote Process Group url, clients for processors like ListenTCP or ListenSyslog acting as servers. NiFi Site-to-Site protocol is cluster-topology aware and automatically distributes load among nodes, but users may wonder which hostname to specify. You can point to any node in the cluster, but what if the node goes down? If ListenXXX processors runs in a cluster, we could configure it with On Primary Node scheduler to make it simply runs only on a single node (we used to specify a primary node). But it doesn’t scale, and will be a SPOF. If every node can receive incoming request, NiFi can be more powerful. To address above concerns, adding a LB in front of your NiFi cluster would be a great solution. Use Docker Compose to create an environment There’re so many docker containers nowadays, so I started with searching docker hub. I was so glad to find following containers to form a NiFi environment looks like following diagram, these containers made my task way easier: mkobit/nifi, I added startup script so that it automatically picks up container’s hostname and updates nifi.properties. zookeeper, NiFi uses Zookeeper for cluster coordination. dockercloud/haproxy, this docker image detect exposed ports on linked service containers, it’s really useful with docker-compose. Exposed ports on NiFi Node On a NiFi node container, following ports have to be accessible from other hosts (port numbers can be configurable, so it might be different from your environment). I exposed these at nifi-node/Dockerfile. Among these ports, only 8080 and 9001 are the ones that facing external network, thus candidates to be accessed through a LB. Port Protocol LB? Memo 8080 HTTP Yes NiFi Web API (UI, REST, HTTP S2S) 8081 TCP No RAW S2S 8082 TCP No Cluster Node Protocol 9001 TCP Yes Misc. used by server type processors Technically, RAW S2S is accessed from external network, but since Site-to-Site clients handle load-balancing, it’s unnecessary to add it under LB. Specify which ports are accessible via LB As mentioned in dockercloud/haproxy document, it uses all exported ports on application containers as routing destination. So unless specifying the purpose of those ports, haproxy load balances incoming HTTP requests toward them. Default setting caused an odd behavior, that only 1/4 NiFi HTTP requests succeed (there’re four ports, 8080, 8081, 8082 and 9001, but only 8080 can accept the request). I excluded port 8081 and 8082 to be used by LB, by setting EXCLUDE_PORTS environment value for nifi-nodes docker-compose service. Then, used TCP_PORTS environment value to specify that 9001 is a TCP port, not for http protocol. These settings allow haproxy to route http request to NiFi nodes 8080 port, and TCP connection to 9001 port. Scaling number of NiFi nodes Once Docker compose file is setup correctly, scaling out number of NiFi node is easy as executing following single command line: $ docker-compose scale nifi-nodes=2 Now I can see a two nodes cluster by accessing Docker host address from a web browser: ListenTCP Let’s set up a NiFi data flow to confirm whether incoming requests get distributed as expected. Added a Listen TCP, configured it to listen on port 9001, and execute following netcat command several times: $ netcat 192.168.99.100 9001 a (input some characters and enter, these are sent to ListenTCP) e g t s Then, look at the stats from NiFi UI… Summary There’s many stuff to learn, such as NiFi clustering, Zookeeper, Docker container, Docker compose and HAProxy. I struggled with setting it up correctly. But once it’s done, you can get a nice distributed testing environment up and running really quickly, and also, I expect these containers can be deployed on larger container pools for production use. Or of course, these stack can be deployed on cloud VM instances or physical on-premise servers without docker. The docker compose sample project is available on Github. I’m planning to explorer deeper with SSL termination and other protocol such as WebSocket using this environment. See you next time! ... View more