MINIFICPP-2448 Add minifi.sh flowStatus command

Closes #1941

Signed-off-by: Marton Szasz <szaszm@apache.org>

Author: lordgamez

OPS.md

@@ -19,10 +19,25 @@
 This readme defines operational commands for managing instances.
 
 ## Table of Contents
-
 - [Description](#description)
-- [Managing](#managing-minifi)
-  - [Commands](#commands)
+- [Managing MiNiFi](#managing-minifi)
+- [SSL](#ssl)
+- [Commands](#commands)
+    - [Specifying connecting information](#specifying-connecting-information)
+    - [Start Command](#start-command)
+    - [Stop command](#stop-command)
+    - [List connections command](#list-connections-command)
+    - [List components command](#list-components-command)
+    - [Clear connection command](#clear-connection-command)
+    - [GetSize command](#getsize-command)
+    - [Update flow](#update-flow)
+    - [Get full connection command](#get-full-connection-command)
+    - [Get manifest command](#get-manifest-command)
+    - [Flowstatus command](#flowstatus-command)
+    - [Processor](#processor)
+    - [Connection](#connection)
+    - [Instance](#instance)
+    - [System Diagnostics](#system-diagnostics)
 
 ## Description
 
@@ -109,3 +124,57 @@ Provides a list of full connections, if any.
     ./minifi-controller --manifest
 
 Writes the agent manifest json to standard output
+
+#### Flowstatus command
+    ./minificontroller --flowstatus "processor:TailFile:health,stats,bulletins"
+
+The command returns the flow status for the specified query in JSON format.
+The query consists of the query type, the element identifier, and the query options. Each part is separated by the ':' colon character. Multiple query options are specified as a comma-separated list. In some query types the identifier is omitted, in this case only the query type and the query options are specified. Multiple queries can also be specified in a flowStatus command, in this case the queries are separated by the ';' semicolon character. For example: `./minificontroller --flowstatus "processor:TailFile:health,stats,bulletins;processor:LogAttribute:stats"`
+
+Supported query types:
+
+##### Processor
+
+To query the processors, use the `processor` flag and specify the processor (by ID, name or "all") followed by one of the processor options. The processor options are below:
+
+- health: The processor's run status, whether or not it has bulletins.
+- bulletins: A list of all the current bulletins (if there are any).
+- stats: The current stats of the processor.
+
+An example query to get the health and stats of the "GenerateFlowFile" processor is below.
+
+`./minificontroller --flowstatus "processor:GenerateFlowFile:health,stats"`
+
+##### Connection
+
+To query the connections, use the `connection` flag and specify the connection (by ID, name or "all") followed by one of the connection options. The connection options are below:
+
+- health: The processor's run status, whether or not it has bulletins.
+
+An example query to get the health and stats of the "Connection1" connection is below.
+
+`./minificontroller --flowstatus "connection:Connection1:health"`
+
+##### Instance
+
+To query the status of the MiNiFi instance, use the `instance` flag followed by one of the instance options. The instance options are below.
+
+- health: The instance reporting the aggregated state of the connections, and whether or not it has bulletins.
+- bulletins: A list of all the current bulletins (if there are any).
+- stats: The aggregated stats of all processors (bytes read, written, transferred, and flowfiles transferred).
+
+An example query to get the all the statuses of the instance is below.
+
+`./minificontroller --flowstatus "instance:health,stats,bulletins"`
+
+##### System Diagnostics
+
+To query the system diagnostics, use the `systemdiagnostics` flag followed by one of the system diagnostics options. The system diagnostics options are below.
+
+- processorstats: The system processor stats. This includes the available processors and load average.
+- contentrepositoryusage: Disk usage stats on the partition or volume where the content repository is located.
+- flowfilerepositoryusage: Disk usage stats on the partition or volume where the flowfile repository is located.
+
+An example query to get the processor stats, content repository usage and FlowFile repository usage from the system diagnostics is below.
+
+`./minificontroller --flowStatus "systemdiagnostics:processorstats,contentrepositoryusage,flowfilerepositoryusage"`

README.md

@@ -521,6 +521,19 @@ MiNiFi can then be stopped by issuing:
 
     $ ./bin/minifi.sh stop
 
+### Query flow status
+
+To query the status of the flow, you can use the following command on Unix systems:
+
+    $ ./bin/minifi.sh flowStatus [host:optional] [port:optional] "<query>"
+
+On Windows systems, you can use the following command:
+
+    $ ./bin/flowstatus-minifi.bat [host:optional] [port:optional] "<query>"
+
+The query can look like the following: "processor:TailFile:health,stats,bulletins". For more information on the query syntax and options, please see the [flow status documentation](OPS.md#Flowstatus-command).
+Note: The command requires minifi controller to be enabled in the minifi.properties file.
+
 ### Running as a docker container
 You can use the officially released image pulled from the [apache/nifi-minifi-cpp](https://hub.docker.com/r/apache/nifi-minifi-cpp) repository on dockerhub or you can use your locally built image.
 The container can be run with a specific configuration by mounting the locally edited configuration files to your docker container.

bin/flowstatus-minifi.bat

@@ -0,0 +1,46 @@
+@echo off &setlocal enabledelayedexpansion
+rem
+rem    Licensed to the Apache Software Foundation (ASF) under one or more
+rem    contributor license agreements.  See the NOTICE file distributed with
+rem    this work for additional information regarding copyright ownership.
+rem    The ASF licenses this file to You under the Apache License, Version 2.0
+rem    (the "License"); you may not use this file except in compliance with
+rem    the License.  You may obtain a copy of the License at
+rem
+rem       http://www.apache.org/licenses/LICENSE-2.0
+rem
+rem    Unless required by applicable law or agreed to in writing, software
+rem    distributed under the License is distributed on an "AS IS" BASIS,
+rem    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+rem    See the License for the specific language governing permissions and
+rem    limitations under the License.
+rem
+
+set "bin_dir=%~dp0"
+
+if not exist "%bin_dir%\minificontroller.exe" (
+    echo MiNiFi Controller is not installed
+    exit /b
+)
+
+if "%~1"=="" (
+    echo MiNiFi flowStatus operation requires a flow status query parameter like "processor:TailFile:health,stats,bulletins"
+    goto usage
+)
+
+if "%~2"=="" (
+    "%bin_dir%\minificontroller.exe" --flowstatus "%~1"
+    exit /b
+)
+
+if "%~3"=="" (
+    "%bin_dir%\minificontroller.exe" --port "%~1" --flowstatus "%~2"
+    exit /b
+)
+
+"%bin_dir%\minificontroller.exe" --host "%~1" --port "%~2" --flowstatus "%~3"
+exit /b
+
+:usage
+echo Usage: flowStatus.bat [^<host^>] [^<port^>] ^<flowstatus^>
+exit /b

bin/minifi.sh

@@ -183,6 +183,25 @@ status_service() {
     fi
 }
 
+flowStatus() {
+    if ! [ -f "${bin_dir}/minificontroller" ]; then
+        echo "MiNiFi Controller is not installed"
+        return
+    fi
+    if [ "$#" -lt 2 ]; then
+        echo "MiNiFi flowStatus operation requires a flow status query parameter like \"processor:TailFile:health,stats,bulletins\""
+        return
+    fi
+
+    if [ "$#" -lt 3 ]; then
+        exec "${bin_dir}/minificontroller" --flowstatus "$2"
+    elif [ "$#" -lt 4 ]; then
+        exec "${bin_dir}/minificontroller" --port "$2" --flowstatus "$3"
+    else
+        exec "${bin_dir}/minificontroller" --host "$2" --port "$3" --flowstatus "$4"
+    fi
+}
+
 case "$1" in
     start)
       start_service
@@ -207,7 +226,10 @@ case "$1" in
       uninstall "$@"
       echo "Service minifi uninstalled. Please remove the ${MINIFI_HOME} directory manually."
       ;;
+    flowStatus)
+      flowStatus "$@"
+      ;;
     *)
-      echo "Usage: minifi.sh {start|stop|run|restart|status|install|uninstall}"
+      echo "Usage: minifi.sh {start|stop|run|restart|status|install|uninstall|flowStatus}"
       ;;
 esac

controller/CMakeLists.txt

@@ -63,3 +63,6 @@ if (NOT WIN32)
 endif()
 
 install(TARGETS minifi-controller RUNTIME DESTINATION ${CMAKE_INSTALL_BINDIR} COMPONENT bin)
+if(WIN32)
+    install(FILES "${CMAKE_SOURCE_DIR}/bin/flowstatus-minifi.bat" DESTINATION bin COMPONENT bin)
+endif()

controller/Controller.cpp

@@ -33,7 +33,7 @@
 namespace org::apache::nifi::minifi::controller {
 
 bool sendSingleCommand(const utils::net::SocketData& socket_data, uint8_t op, const std::string& value) {
-  std::unique_ptr<io::BaseStream> connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
+  const auto connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
   if (connection_stream->initialize() < 0) {
     return false;
   }
@@ -56,7 +56,7 @@ bool clearConnection(const utils::net::SocketData& socket_data, const std::strin
 }
 
 bool updateFlow(const utils::net::SocketData& socket_data, std::ostream &out, const std::string& file) {
-  std::unique_ptr<io::BaseStream> connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
+  const auto connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
   if (connection_stream->initialize() < 0) {
     return false;
   }
@@ -75,7 +75,7 @@ bool updateFlow(const utils::net::SocketData& socket_data, std::ostream &out, co
     uint16_t connections = 0;
     connection_stream->read(connections);
     out << connections << " are full" << std::endl;
-    for (int i = 0; i < connections; i++) {
+    for (uint16_t i = 0; i < connections; i++) {
       std::string fullcomponent;
       connection_stream->read(fullcomponent);
       out << fullcomponent << " is full" << std::endl;
@@ -85,7 +85,7 @@ bool updateFlow(const utils::net::SocketData& socket_data, std::ostream &out, co
 }
 
 bool getFullConnections(const utils::net::SocketData& socket_data, std::ostream &out) {
-  std::unique_ptr<io::BaseStream> connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
+  const auto connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
   if (connection_stream->initialize() < 0) {
     return false;
   }
@@ -103,7 +103,7 @@ bool getFullConnections(const utils::net::SocketData& socket_data, std::ostream
     uint16_t connections = 0;
     connection_stream->read(connections);
     out << connections << " are full" << std::endl;
-    for (int i = 0; i < connections; i++) {
+    for (uint16_t i = 0; i < connections; i++) {
       std::string fullcomponent;
       connection_stream->read(fullcomponent);
       out << fullcomponent << " is full" << std::endl;
@@ -113,7 +113,7 @@ bool getFullConnections(const utils::net::SocketData& socket_data, std::ostream
 }
 
 bool getConnectionSize(const utils::net::SocketData& socket_data, std::ostream &out, const std::string& connection) {
-  std::unique_ptr<io::BaseStream> connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
+  const auto connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
   if (connection_stream->initialize() < 0) {
     return false;
   }
@@ -137,7 +137,7 @@ bool getConnectionSize(const utils::net::SocketData& socket_data, std::ostream &
 }
 
 bool listComponents(const utils::net::SocketData& socket_data, std::ostream &out, bool show_header) {
-  std::unique_ptr<io::BaseStream> connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
+  const auto connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
   if (connection_stream->initialize() < 0) {
     return false;
   }
@@ -154,7 +154,7 @@ bool listComponents(const utils::net::SocketData& socket_data, std::ostream &out
   if (show_header)
     out << "Components:" << std::endl;
 
-  for (int i = 0; i < responses; i++) {
+  for (uint16_t i = 0; i < responses; i++) {
     std::string name;
     connection_stream->read(name, false);
     std::string status;
@@ -165,7 +165,7 @@ bool listComponents(const utils::net::SocketData& socket_data, std::ostream &out
 }
 
 bool listConnections(const utils::net::SocketData& socket_data, std::ostream &out, bool show_header) {
-  std::unique_ptr<io::BaseStream> connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
+  const auto connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
   if (connection_stream->initialize() < 0) {
     return false;
   }
@@ -182,7 +182,7 @@ bool listConnections(const utils::net::SocketData& socket_data, std::ostream &ou
   if (show_header)
     out << "Connection Names:" << std::endl;
 
-  for (int i = 0; i < responses; i++) {
+  for (uint16_t i = 0; i < responses; i++) {
     std::string name;
     connection_stream->read(name, false);
     out << name << std::endl;
@@ -191,7 +191,7 @@ bool listConnections(const utils::net::SocketData& socket_data, std::ostream &ou
 }
 
 bool printManifest(const utils::net::SocketData& socket_data, std::ostream &out) {
-  std::unique_ptr<io::BaseStream> connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
+  const auto connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
   if (connection_stream->initialize() < 0) {
     return false;
   }
@@ -210,7 +210,7 @@ bool printManifest(const utils::net::SocketData& socket_data, std::ostream &out)
 }
 
 bool getJstacks(const utils::net::SocketData& socket_data, std::ostream &out) {
-  std::unique_ptr<io::BaseStream> connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
+  const auto connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
   if (connection_stream->initialize() < 0) {
     return false;
   }
@@ -229,7 +229,7 @@ bool getJstacks(const utils::net::SocketData& socket_data, std::ostream &out) {
 }
 
 nonstd::expected<void, std::string> getDebugBundle(const utils::net::SocketData& socket_data, const std::filesystem::path& target_dir) {
-  std::unique_ptr<io::BaseStream> connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
+  const auto connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
   if (connection_stream->initialize() < 0) {
     return nonstd::make_unexpected("Could not connect to remote host " + socket_data.host + ":" + std::to_string(socket_data.port));
   }
@@ -267,4 +267,24 @@ nonstd::expected<void, std::string> getDebugBundle(const utils::net::SocketData&
   return {};
 }
 
+bool getFlowStatus(const utils::net::SocketData& socket_data, const std::string& status_query, std::ostream &out) {
+  const auto connection_stream = std::make_unique<utils::net::AsioSocketConnection>(socket_data);
+  if (connection_stream->initialize() < 0) {
+    return false;
+  }
+  io::BufferStream buffer;
+  auto op = static_cast<uint8_t>(c2::Operation::describe);
+  buffer.write(&op, 1);
+  buffer.write("flowstatus");
+  buffer.write(status_query);
+  if (io::isError(connection_stream->write(buffer.getBuffer()))) {
+    return false;
+  }
+  connection_stream->read(op);
+  std::string manifest;
+  connection_stream->read(manifest, true);
+  out << manifest << std::endl;
+  return true;
+}
+
 }  // namespace org::apache::nifi::minifi::controller

controller/Controller.h

@@ -36,6 +36,7 @@ bool listComponents(const utils::net::SocketData& socket_data, std::ostream &out
 bool listConnections(const utils::net::SocketData& socket_data, std::ostream &out, bool show_header = true);
 bool printManifest(const utils::net::SocketData& socket_data, std::ostream &out);
 bool getJstacks(const utils::net::SocketData& socket_data, std::ostream &out);
+bool getFlowStatus(const utils::net::SocketData& socket_data, const std::string& status_query, std::ostream &out);
 nonstd::expected<void, std::string> getDebugBundle(const utils::net::SocketData& socket_data, const std::filesystem::path& target_dir);
 
 }  // namespace org::apache::nifi::minifi::controller

controller/MiNiFiController.cpp

@@ -151,6 +151,9 @@ int main(int argc, char **argv) {
   argument_parser.add_argument("--updateflow")
     .metavar("FLOW_CONFIG_PATH")
     .help("Updates the flow of the agent using the provided flow file");
+  argument_parser.add_argument("--flowstatus")
+    .metavar("FLOW_STATUS_QUERY")
+    .help("Returns flow status for the provided query");
 
   auto addFlagOption = [&](std::string_view name, const std::string& help) {
     argument_parser.add_argument(name)
@@ -275,6 +278,12 @@ int main(int argc, char **argv) {
       else
         std::cout << "Debug bundle written to " << std::filesystem::path(*debug_path) / "debug.tar.gz";
     }
+
+    if (const auto& status_query = argument_parser.present("--flowstatus")) {
+      if (!minifi::controller::getFlowStatus(socket_data, *status_query, std::cout)) {
+        std::cout << "Could not connect to remote host " << socket_data.host << ":" << socket_data.port << std::endl;
+      }
+    }
   } catch (const std::exception &exc) {
     // catch anything thrown within try block that derives from std::exception
     std::cerr << exc.what() << std::endl;