MINIFICPP-2448 Add minifi.sh flowStatus command

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.
     ./minificontroller --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

@@ -503,6 +503,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 3 ]; 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

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

controller/Controller.cpp

@@ -266,4 +266,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) {
+  std::unique_ptr<io::BaseStream> 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

@@ -147,6 +147,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)
@@ -271,6 +274,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;

controller/tests/ControllerTests.cpp

@@ -200,6 +200,16 @@ class TestControllerSocketReporter : public c2::ControllerSocketReporter {
   std::string getAgentManifest() override {
     return "testAgentManifest";
   }
+
+  void setRoot(core::ProcessGroup* /*root*/) override {
+  }
+
+  void setFlowStatusDependencies(core::BulletinStore* /*bulletin_store*/, const std::filesystem::path& /*flowile_repo_dir*/, const std::filesystem::path& /*flowile_repo_dir*/) override {
+  }
+
+  std::string getFlowStatus(const std::vector<c2::FlowStatusRequest>& /*requests*/) override {
+    return "";
+  }
 };
 
 class TestControllerServiceProvider : public core::controller::ControllerServiceProviderImpl {
@@ -570,4 +580,29 @@ TEST_CASE_METHOD(ControllerTestFixture, "Debug bundle retrieval fails if target
   REQUIRE(result.error() == "Object specified as the target directory already exists and it is not a directory");
 }
 
+TEST_CASE_METHOD(ControllerTestFixture, "Test flow status getter", "[controllerTests]") {
+  SECTION("With SSL from service provider") {
+    setConnectionType(ControllerTestFixture::ConnectionType::SSL_FROM_SERVICE_PROVIDER);
+  }
+
+  SECTION("With SSL from properties") {
+    setConnectionType(ControllerTestFixture::ConnectionType::SSL_FROM_CONFIGURATION);
+  }
+
+  SECTION("Without SSL") {
+    setConnectionType(ControllerTestFixture::ConnectionType::UNSECURE);
+  }
+
+  auto reporter = std::make_shared<minifi::c2::ControllerSocketMetricsPublisher>("ControllerSocketMetricsPublisher");
+  auto response_node_loader = std::make_shared<minifi::state::response::ResponseNodeLoaderImpl>(configuration_, std::vector<std::shared_ptr<core::RepositoryMetricsSource>>{}, nullptr);
+  reporter->initialize(configuration_, response_node_loader);
+  initalizeControllerSocket(reporter);
+
+  std::stringstream flow_status_stream;
+  minifi::controller::getFlowStatus(controller_socket_data_, "processor:TailFile:health", flow_status_stream);
+  std::string expected_status = "{\"controllerServiceStatusList\":null,\"connectionStatusList\":null,\"remoteProcessGroupStatusList\":null,\"instanceStatus\":null,\"systemDiagnosticsStatus\":null,"
+                                "\"processorStatusList\":[],\"errorsGeneratingReport\":[\"Unable to get processorStatus: No processor with key 'TailFile' to report status on\"]}\n";
+  REQUIRE(flow_status_stream.str() == expected_status);
+}
+
 }  // namespace org::apache::nifi::minifi::test

libminifi/include/FlowController.h

@@ -61,6 +61,7 @@
 #include "core/state/MetricsPublisherStore.h"
 #include "RootProcessGroupWrapper.h"
 #include "c2/ControllerSocketProtocol.h"
+#include "core/BulletinStore.h"
 
 namespace org::apache::nifi::minifi {
 
@@ -74,7 +75,7 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi
                  std::shared_ptr<Configure> configure, std::shared_ptr<core::FlowConfiguration> flow_configuration,
                  std::shared_ptr<core::ContentRepository> content_repo, std::unique_ptr<state::MetricsPublisherStore> metrics_publisher_store = nullptr,
                  std::shared_ptr<utils::file::FileSystem> filesystem = std::make_shared<utils::file::FileSystem>(), std::function<void()> request_restart = []{},
-                 utils::file::AssetManager* asset_manager = {});
+                 utils::file::AssetManager* asset_manager = {}, core::BulletinStore* bulletin_store = {});
 
   ~FlowController() override;
 

libminifi/include/RootProcessGroupWrapper.h

@@ -32,6 +32,7 @@
 #include "core/state/MetricsPublisherStore.h"
 #include "utils/gsl.h"
 #include "core/logging/Logger.h"
+#include "c2/ControllerSocketProtocol.h"
 
 namespace org::apache::nifi::minifi {
 
@@ -77,6 +78,10 @@ class RootProcessGroupWrapper {
   std::optional<std::vector<state::StateController*>> getAllProcessorControllers(
           const std::function<gsl::not_null<std::unique_ptr<state::ProcessorController>>(core::Processor&)>& controllerFactory);
 
+  void setControllerSocketProtocol(c2::ControllerSocketProtocol* controller_socket_protocol) {
+    controller_socket_protocol_ = controller_socket_protocol;
+  }
+
  private:
   std::optional<std::chrono::milliseconds> loadShutdownTimeoutFromConfiguration();
 
@@ -86,6 +91,7 @@ class RootProcessGroupWrapper {
   state::MetricsPublisherStore* metrics_publisher_store_{};
   std::unordered_map<utils::Identifier, std::unique_ptr<state::ProcessorController>> processor_to_controller_;
   std::chrono::milliseconds shutdown_check_interval_{1000};
+  c2::ControllerSocketProtocol* controller_socket_protocol_{};
   std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<RootProcessGroupWrapper>::getLogger();
 };
 

libminifi/include/c2/ControllerSocketMetricsPublisher.h

@@ -26,6 +26,7 @@
 #include "core/state/MetricsPublisher.h"
 #include "core/logging/LoggerFactory.h"
 #include "c2/HeartbeatJsonSerializer.h"
+#include "FlowStatusBuilder.h"
 
 namespace org::apache::nifi::minifi::c2 {
 
@@ -42,10 +43,15 @@ class ControllerSocketMetricsPublisher : public state::MetricsPublisherImpl, pub
   std::unordered_set<std::string> getConnections() override;
   std::string getAgentManifest() override;
 
+  void setRoot(core::ProcessGroup* root) override;
+  void setFlowStatusDependencies(core::BulletinStore* bulletin_store, const std::filesystem::path& flowile_repo_dir, const std::filesystem::path& content_repo_dir) override;
+  std::string getFlowStatus(const std::vector<FlowStatusRequest>& requests) override;
+
  protected:
   c2::HeartbeatJsonSerializer heartbeat_json_serializer_;
   std::mutex queue_metrics_node_mutex_;
   std::shared_ptr<state::response::ResponseNode> queue_metrics_node_;
+  FlowStatusBuilder flow_status_builder_;
   std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<ControllerSocketMetricsPublisher>::getLogger();
 };
 

libminifi/include/c2/ControllerSocketProtocol.h

@@ -46,6 +46,7 @@ class ControllerSocketProtocol {
     std::shared_ptr<Configure> configuration, const std::shared_ptr<ControllerSocketReporter>& controller_socket_reporter);
   ~ControllerSocketProtocol();
   void initialize();
+  void setRoot(core::ProcessGroup* root);
 
  private:
   void handleStart(io::BaseStream &stream);
@@ -59,6 +60,7 @@ class ControllerSocketProtocol {
   void writeGetFullResponse(io::BaseStream &stream);
   void writeManifestResponse(io::BaseStream &stream);
   void writeJstackResponse(io::BaseStream &stream);
+  void writeFlowStatusResponse(io::BaseStream &stream);
   void writeDebugBundleResponse(io::BaseStream &stream);
   void handleDescribe(io::BaseStream &stream);
   asio::awaitable<void> handleCommand(std::unique_ptr<io::BaseStream> stream);