Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Binary file added docs/modules/nifi/assets/images/listen-http-2.png
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Binary file added docs/modules/nifi/assets/images/listen-http-3.png
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
99 changes: 99 additions & 0 deletions docs/modules/nifi/pages/usage_guide/exposing-processors/http.adoc
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
= Exposing HTTP processors
:description: Expose NiFi ListenHTTP processor by creating Service and Ingress objects, allowing external tools to trigger workflows or send data.

You can use the `ListenHTTP` processor to start a HTTP webserver and listen for incoming connections.

In this case `POST` requests are being send to NiFi, which acts as a webhook to receive data here.
You should also be able to serve `GET` requests by using the `HandleHttpRequest` processor, however this is currently not demonstrated in this guide.

== 1. Create `ListenHTTP` processor

Let's start by creating a `ListenHTTP` processor:

1. Set `Base Path` to an empty string
2. Set `Listening Port` to `8042`.
3. Set `Record Reader` to an `JsonTreeReader` and `Record Writer` to an `JsonRecordSetWriter` instance.
This guide assumes that JSON documents are being posted to the listener. For other document formats, change the record reader and writer accordingly.

You should end up with something similar to

image:listen-http-1.png[A ListenHTTP processor]

== 2. Expose `ListenHTTP` processor

Afterwards you need to expose the processor to the outside world.

For that to work, first create a Service object as follows.
In this guide, the NifiCluster is called `simple-nifi`. The name of the Nifi cluster must match the value of the `app.kubernetes.io/instance` as shown below.

[source,yaml]
----
apiVersion: v1
kind: Service
metadata:
name: simple-nifi-listen-http # Update according to NifiCluster name
spec:
type: ClusterIP
selector:
app.kubernetes.io/component: node
app.kubernetes.io/instance: simple-nifi # Update according to NifiCluster name
app.kubernetes.io/name: nifi
ports:
- name: http
port: 8042
protocol: TCP
targetPort: 8042
----

In case you don't have an ingress controller, you can set the Service type to `LoadBalancer` or `NodePort` instead and should be ready to go.
If you are using an ingress controller, an Ingress could look something like

[source,yaml]
----
apiVersion: networking.k8s.io/v1
kind: Ingress
metadata:
name: simple-nifi-listen-http # Update according to NifiCluster name
spec:
rules:
- host: simple-nifi-listen-http.my.corp # Update to your host
http:
paths:
- pathType: Prefix
path: /
backend:
service:
name: simple-nifi-listen-http # Update to your Service name
port:
number: 8042
----

=== 3. Route

The next step is to handle different kind of messages coming in, based on the HTTP path.

First, create a `RouteOnAttribute` processor and connect it to the `success` output of the `ListenHTTP` processor as shown below.
Start the `ListenHTTP` processor.

image:listen-http-2.png[´RouteOnAttribute connected to ListenHTTP processor]

The `ListenHTTP` processor should now generate a FlowFile for every incoming HTTP request.
You can test this by calling `curl --verbose --data '{"hello":"NiFi"}' https://simple-nifi-listen-http.my.corp`, you should get a `HTTP/2 200` response.

TIP: If you get a `503 Service Temporarily Unavailable`, this probably means your Ingress controller was not able to
reach your `ListenHTTP` processor. Check that the processor is running and configured correctly.

This should result in one FlowFile being queued, as shown in the picture above and below.

image:listen-http-3.png[Resulting FlowFile]

In the `RouteOnAttribute` processor, add a field called `/webhook/foo` with the value of `${"restlistener.request.uri":equals('/webhook/foo')}`.
You can replace `/webhook/foo` with whatever URL your HTTP service should be reachable.
This guide also added `/webhook/bar` and `/webhook/baz` in a similar way.

The `RouteOnAttribute` processor now has one `unmatched` output as well as one for every field you defined.
You can connect a processor for every HTTP path you routed on.

The result should look something like below and should allow you to consume many different HTTP `POST` requests.

image:listen-http-4.png[Result flow]
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
= Exposing processors

There are two mechanisms to expose Nifi processors to the outside world.

1. xref:nifi:usage_guide/exposing-processors/http.adoc[HTTP traffic]
2. xref:nifi:usage_guide/exposing-processors/tcp.adoc[Generic TCP traffic]
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
= Exposing NiFi processor ports
:description: Expose NiFi processor ports by creating Service or Ingress objects, allowing external tools to trigger workflows or send data via Kubernetes.
= Exposing TCP ports
:description: Expose NiFi processor ports by creating Service objects, allowing external tools to trigger workflows or send data.

Some NiFi processors allow external tools to connect to them to trigger workflows or send data.
For this to work from outside the Kubernetes cluster NiFi has been deployed to, it is necessary to create Service or Ingress objects to configure Kubernetes routes for these ports.
For this to work from outside the Kubernetes cluster NiFi has been deployed to, it is necessary to create Service objects to configure Kubernetes routes for these ports.

Stackable doesn't place any restriction on the type of service that can be used here.

Expand All @@ -28,6 +28,8 @@ spec:
type: LoadBalancer
externalTrafficPolicy: Cluster
selector:
app.kubernetes.io/name: nifi
app.kubernetes.io/component: node
app.kubernetes.io/instance: simple-nifi
ports:
- name: tcp-port
Expand All @@ -51,6 +53,8 @@ metadata:
spec:
type: NodePort
selector:
app.kubernetes.io/name: nifi
app.kubernetes.io/component: node
app.kubernetes.io/instance: simple-nifi
ports:
- port: 8123
Expand Down
4 changes: 3 additions & 1 deletion docs/modules/nifi/partials/nav.adoc
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@
** xref:nifi:usage_guide/listenerclass.adoc[]
** xref:nifi:usage_guide/zookeeper-connection.adoc[]
** xref:nifi:usage_guide/extra-volumes.adoc[]
** xref:nifi:usage_guide/external_ports.adoc[]
** xref:nifi:usage_guide/security.adoc[]
** xref:nifi:usage_guide/resource-configuration.adoc[]
** xref:nifi:usage_guide/log-aggregation.adoc[]
Expand All @@ -14,6 +13,9 @@
** xref:nifi:usage_guide/overrides.adoc[]
** xref:nifi:usage_guide/writing-to-iceberg-tables.adoc[]
** xref:nifi:usage_guide/flow-versioning.adoc[]
** xref:nifi:usage_guide/exposing-processors/index.adoc[]
*** xref:nifi:usage_guide/exposing-processors/http.adoc[]
*** xref:nifi:usage_guide/exposing-processors/tcp.adoc[]
** xref:nifi:usage_guide/custom-components/index.adoc[]
*** xref:nifi:usage_guide/custom-components/custom-nars.adoc[]
*** xref:nifi:usage_guide/custom-components/custom-python-processors.adoc[]
Expand Down
Loading