Hello everyone!

The task is as follows — there is a flow, presented in the image above, that needs to be deployed across N servers with Apache NiFi. The flow is a test — it generates a file and sends it to another NiFi instance. Data transmission occurs using the NiFi Site to Site protocol.

NiFi Site to Site (S2S) is a secure, easily configurable method of transferring data between NiFi instances. For details on how S2S works, see the documentation and it is important not to forget to configure the NiFi instance to allow S2S as described in here.

In cases where data is transmitted using S2S, one instance is referred to as the client, and the other as the server. The client sends data, while the server receives it. There are two ways to configure data transfer between them:

1. Push. Data from the client instance is sent using a Remote Process Group (RPG). On the server instance, data is received via an Input Port.
2. Pull. The server receives data using RPG, while the client sends it using an Output port.

The flow for deployment is stored in Apache Registry.

Apache NiFi Registry is a sub-project of Apache NiFi that serves as a tool for storing flows and managing versions. It's like GIT. Information about installing, configuring, and working with the registry can be found in the official documentation.. The flow for storage is combined into a process group and is stored in the registry in that form. We will return to this later in the article.

Initially, when N is a small number, the flow is delivered and updated manually within a reasonable time.

However, as N grows, the problems increase:

1. more time is required to update the flow. One needs to access all servers
2. there are errors in updating templates. This was updated here, but forgotten there
3. human errors occur when performing a large number of repetitive operations

All of this leads us to the necessity of automating the process. I have tried the following solutions to this problem:

1. Using MiNiFi instead of NiFi
2. NiFi CLI
3. NiPyAPI

Using MiNiFi

Apache MiNiFi — an Apache NiFi subproject. MiNiFy is a lightweight agent that uses the same processors as NiFi, enabling the creation of the same flows as in NiFi. The agent's lightweight design is partly due to its lack of a graphical interface for configuring flows. The absence of a graphical interface in MiNiFy means addressing the challenge of delivering flows to Minifi. Since MiNiFy is widely used in IoT, there are many components, and the flow delivery to the final instances of Minifi needs to be automated. A familiar task, right?

Solving such a problem can be aided by another subproject — the MiNiFi C2 Server. This product is designed to be the central point in the configuration rollout architecture. How to configure the environment is described in from one of the authors. on Habr, and there is enough information to tackle the task at hand. MiNiFi, in conjunction with the C2 Server, updates its configuration automatically. The only downside to this approach is that you have to create templates on the C2 Server; a simple commit to the registry is not enough.

The option described in the article above is workable and not complicated to implement, but it is important to remember the following:

1. Not all processors from NiFi are available in Minifi.
2. The versions of processors in MiNiFi lag behind those in NiFi.

As of the writing of this publication, the latest version of NiFi is 1.9.2. The processor version of the latest MiNiFi release is 1.7.0. Processors can be added to MiNiFi, but due to version discrepancies between NiFi and MiNiFi processors, this may not work.

NiFi CLI

According to the description of the tool on the official website, this tool is designed to automate the interaction between NiFi and NiFi Registry in the areas of flow delivery or process management. To get started with this tool, you need to download it from here.

We run the utility

.\/bin\/cli.sh
           _     ___  _
 Apache   (_)  .' ..](_)   ,
 _ .--.   __  _| |_  __    )
[ `.-. | [  |'-| |-'[  |  \/  
|  | | |  | |  | |   | | '    '
[___||__][___][___] [___]',  ,'
                           `'
          CLI v1.9.2

Type 'help' to see a list of available commands, use tab to auto-complete.

In order to load the required flow from the registry, we need to know the bucket identifiers and the flow identifiers. This information can be obtained either via CLI or from the web interface of NiFi Registry. In the web interface, it looks like this:

Using CLI, it's done like this:

#> registry list-buckets -u http://nifi-registry:18080

#   Name             Id                                     Description
-   --------------   ------------------------------------   -----------
1   test_bucket   709d387a-9ce9-4535-8546-3621efe38e96   (empty)

#> registry list-flows -b 709d387a-9ce9-4535-8546-3621efe38e96 -u http://nifi-registry:18080

#   Name           Id                                     Description
-   ------------   ------------------------------------   -----------
1   test_flow   d27af00a-5b47-4910-89cd-9c664cd91e85

We run the import process group from the registry:

#> nifi pg-import -b 709d387a-9ce9-4535-8546-3621efe38e96 -f d27af00a-5b47-4910-89cd-9c664cd91e85 -fv 1 -u http://nifi:8080

7f522a13-016e-1000-e504-d5b15587f2f3

An important point is that any NiFi instance can be specified as the host for the process group we are deploying.

The process group has been added with processors stopped; they need to be started.

#> nifi pg-start -pgid 7f522a13-016e-1000-e504-d5b15587f2f3 -u http://nifi:8080

Great, the processors have started. However, according to our task requirements, NiFi instances need to send data to other instances. Let's assume that the Push method was chosen for data transmission. To establish data transfer, we must enable transmission on the added Remote Process Group (RPG) already included in our flow.

I couldn't find a way to enable data transmission in the CLI documentation and other sources. If you know how to do this, please let me know in the comments.

Since we have bash and are ready to go all the way, let's find a solution! We can use the NiFi API to resolve this issue. We'll use the following method, with the ID taken from the examples above (in our case, it is 7f522a13-016e-1000-e504-d5b15587f2f3). Description of NiFi API methods. here.

The body needs to send JSON in the following format:

{
    "revision": {
	    "clientId": "value",
	    "version": 0,
	    "lastModifier": "value"
	},
    "state": "value",
    "disconnectedNodeAcknowledged": true
}

The parameters that need to be filled in for it to "work":
state — state of data transmission. TRANSMITTING is available to enable data transmission, STOPPED to disable it.
version — processor version.

The version will be 0 by default when created, but these parameters can be obtained using the method.

For fans of bash scripts, this method may seem suitable, but I find it somewhat difficult—bash scripts are not my favorite. The next method is more interesting and convenient in my opinion.

NiPyAPI

NiPyAPI is a Python library for interacting with NiFi instances. The documentation page contains the necessary information to work with the library. Quick start is described in the project on GitHub.

Our script for deploying configuration is a program written in Python. Let's move on to coding.
We'll set up configurations for further work. We will need the following parameters:

nipyapi.config.nifi_config.host = 'http://nifi:8080/nifi-api' # path to the NiFi API instance where the process group is being deployed
nipyapi.config.registry_config.host = 'http://nifi-registry:18080/nifi-registry-api' # path to the NiFi registry API registry
nipyapi.config.registry_name = 'MyBeautifulRegistry' # name of the registry as it will be called in the NiFi instance
nipyapi.config.bucket_name = 'BucketName' # name of the bucket from which we are pulling the flow
nipyapi.config.flow_name = 'FlowName' # name of the flow that we are pulling

Next, I will insert the method names of this library that are described here.

We connect the registry to the nifi instance using

nipyapi.versioning.create_registry_client

At this step, you can also add a check to see if the registry has already been added to the instance, for this you can use the method

nipyapi.versioning.list_registry_clients

We find the bucket for further flow searching in the basket

nipyapi.versioning.get_registry_bucket

For the found bucket, we search for the flow

nipyapi.versioning.get_flow_in_bucket

Next, it's important to understand whether this process group has already been added. The process group is placed according to coordinates, and it can happen that one component overlaps another. I checked, this can happen 🙂 To get all added process groups, we use the method

nipyapi.canvas.list_all_process_groups

and then we can search, for example by name.

I won't describe the process of updating the template, I'll just say that if in the new version of the template processors are added, there are no problems with the presence of messages in the queues. But if processors are removed, problems may arise (nifi does not allow you to remove a processor if there is a queue of messages accumulated before it). If you're interested in how I solved this problem — please write to me, we will discuss this matter. Contacts are at the end of the article. Let’s move to the step of adding a process group.

When debugging the script, I encountered the peculiarity that the latest version of the flow is not always pulled in, so I recommend first clarifying this version:

nipyapi.versioning.get_latest_flow_ver

We deploy the process group:

nipyapi.versioning.deploy_flow_version

We start the processors:

nipyapi.canvas.schedule_process_group

In the section about the CLI, it was mentioned that data transfer is not automatically included in the remote process group? When implementing the script, I faced this issue too. At that time, I was unable to start data transfer using the API, so I decided to write to the developer of the NiPyAPI library and ask for advice/help. The developer replied to me, we discussed the problem and he wrote that he needed time to 'check something.' And then, a couple of days later, I received an email with a function in Python that solved my issue with launching!!! At that time, NiPyAPI version was 0.13.3 and of course, there was nothing like that in it. However, in version 0.14.0, which was released very recently, this function was already included in the library. Here it is,

nipyapi.canvas.set_remote_process_group_transmission

So, using the NiPyAPI library, we connected to the registry, deployed the flow, and even started the processors and data transfer. Next, we can refine the code, add various checks, logging, and all that. But that's a whole different story.

Among the automation options I considered, the last one seemed the most functional. First, it's still code in Python, which allows integrating auxiliary code and leveraging all the advantages of the programming language. Second, the NiPyAPI project is actively developed, and if issues arise, you can contact the developer. Third, NiPyAPI is indeed a more flexible tool for interacting with NiFi to solve complex tasks, such as determining whether the message queues in the flow are empty and if the process group can be updated.

That's all for now. I described three approaches to automating the delivery of flows to NiFi, the pitfalls developers might encounter, and provided working code for automation delivery. If you are as interested in this topic as I am — feel free to reach out!

Source: habr.com