This is an automated email from the ASF dual-hosted git repository. szaszm pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit bed452e702ed313f57e15ab52720674ad217de46 Author: Gabor Gyimesi <[email protected]> AuthorDate: Tue Aug 8 13:26:51 2023 +0200 MINIFICPP-2166 Remove unused properties and update documentation Closes #1622 Signed-off-by: Marton Szasz <[email protected]> --- C2.md | 26 +- CONFIGURE.md | 186 +++++++++- .../http-curl/tests/C2DescribeManifestTest.cpp | 4 +- .../tests/unit/ConfigurationTests.cpp | 2 +- libminifi/include/FlowControlProtocol.h | 273 --------------- libminifi/include/FlowController.h | 2 - libminifi/include/properties/Configuration.h | 6 - libminifi/src/Configuration.cpp | 6 - libminifi/src/FlowControlProtocol.cpp | 383 --------------------- libminifi/src/FlowController.cpp | 3 - 10 files changed, 198 insertions(+), 693 deletions(-) diff --git a/C2.md b/C2.md index 82e31495b..335f59b87 100644 --- a/C2.md +++ b/C2.md @@ -75,11 +75,17 @@ be requested via C2 DESCRIBE manifest command. nifi.c2.agent.protocol.class=RESTSender # nifi.c2.agent.protocol.class=CoapProtocol + # Coap protocol can be also defined in the flow configuration + # nifi.c2.coap.connector.service=MyCoapProtocol + # control c2 heartbeat interval nifi.c2.agent.heartbeat.period=30 sec # enable reporter classes - nifi.c2.agent.heartbeat.reporter.class=RESTReciver + nifi.c2.agent.heartbeat.reporter.classes=RESTReceiver + # If RESTReceiver is configured its listener port and optional SSL certificate can also be configured + nifi.c2.rest.listener.port=<port> + nifi.c2.rest.listener.cacert=<SSL Cert path> # specify the rest URIs if using RESTSender nifi.c2.rest.url=http://<your-c2-server>/<c2-api-path>/c2-protocol/heartbeat @@ -98,6 +104,24 @@ be requested via C2 DESCRIBE manifest command. # specify encoding strategy for c2 requests (gzip, none) #nifi.c2.rest.request.encoding=none + # minimize REST heartbeat updates + #nifi.c2.rest.heartbeat.minimize.updates=true + +#### Flow Id and URL + +Flow id and URL are usually retrieved from the C2 server. These identify the last updated flow version and where the flow was downloaded from. These properties are persisted in the minifi.properties file. + + # in minifi.properties + nifi.c2.flow.id=8da5de7f-dcdb-4f6b-aa2f-6f162a7f9dc4 + nifi.c2.flow.url=http://localhost:10090/efm/api/flows/8da5de7f-dcdb-4f6b-aa2f-6f162a7f9dc4/content?aid=efmtest + +#### Agent Identifier Fallback + +It is possible to set a persistent fallback agent id. This is needed so that the C2 server can identify the same agent after a restart, even if nifi.c2.agent.identifier is not specified. + + # in minifi.properties + nifi.c2.agent.identifier.fallback=my_fallback_id + ### Metrics Command and Control metrics can be used to send metrics through the heartbeat or via the DESCRIBE diff --git a/CONFIGURE.md b/CONFIGURE.md index 06e66962d..478614e20 100644 --- a/CONFIGURE.md +++ b/CONFIGURE.md @@ -70,31 +70,126 @@ It's recommended to create your configuration in YAML format or configure the ag max concurrent tasks: 1 Properties: +### Configuring flow configuration format + +MiNiFi supports YAML and JSON configuration formats. The desired configuration format can be set in the minifi.properties file, but it is automatically identified by default. The default value is `adaptiveconfiguration`, but we can force to use YAML with the `yamlconfiguration` value. + + # in minifi.properties + nifi.flow.configuration.class.name=adaptiveconfiguration + ### Scheduling strategies Currently Apache NiFi MiNiFi C++ supports TIMER_DRIVEN, EVENT_DRIVEN, and CRON_DRIVEN. TIMER_DRIVEN uses periods to execute your processor(s) at given intervals. The EVENT_DRIVEN strategy awaits for data be available or some other notification mechanism to trigger execution. CRON_DRIVEN executes at the desired intervals based on the CRON periods. Apache NiFi MiNiFi C++ supports standard CRON expressions without intervals ( */5 * * * * ). +### Configuring encryption for flow configuration + +To encrypt flow configuration set the following property to true. + + # in minifi.properties + nifi.flow.configuration.encrypt=true + +### Configuring additional sensitive properties + +It is possible to set a comma seperated list of encrypted configuration options beyond the default sensitive property list. + + # in minifi.properties + nifi.sensitive.props.additional.keys=nifi.flow.configuration.file, nifi.rest.api.password + +### Backup previous flow configuration on flow update + +It is possible to backup the previous flow configuration file with `.bak` extension in case of a flow update (e.g. through C2 or controller socket protocol). + + # in minifi.properties + nifi.flow.configuration.backup.on.update=true + +### Set number of flow threads + +The number of threads used by the flow scheduler can be set in the MiNiFi configuration. The default value is 5. + + # in minifi.properties + nifi.flow.engine.threads=5 + +### OnTrigger runtime alert + +MiNiFi writes warning logs in case a processor has been running for too long. The period for these alerts can be set in the configuration file with the default being 5 seconds. + + # in minifi.properties + nifi.flow.engine.alert.period=5 sec + +### Event driven processor time slice + +The flow scheduler can be configured how much time it should allocate at maximum for event driven processors. The processor is triggered until it has work to do, but no more than the configured time slice. The default value is 500 milliseconds. + + # in minifi.properties + nifi.flow.engine.event.driven.time.slice=500 millis + +### Administrative yield duration + +In case an uncaught exception is thrown while running a processor, the processor will yield for the configured administrative yield time. The default yield duration is 30 seconds. + + # in minifi.properties + nifi.administrative.yield.duration=30 sec + +### Bored yield duration + +If a processor is triggered but has no work available, it will yield for the configured bored yield time. The default yield duration is 100 milliseconds. + + # in minifi.properties + nifi.bored.yield.duration=100 millis + +### Graceful shutdown period + +It is possible to configure a graceful shutdown period, the period the flow controller will wait to unload the flow configuration and stop running processors. + + # in minifi.properties + nifi.flowcontroller.graceful.shutdown.period=30 sec + +### FlowController drain timeout + +Timeout period for finishing processing of flow files in progress when shutting down flow controller. When not set we do not wait for flow files to finish processing. + + # in minifi.properties + nifi.flowcontroller.drain.timeout=500 millis + ### SiteToSite Security Configuration - in minifi.properties + # in minifi.properties - enable tls + # enable tls nifi.remote.input.secure=true - if you want to enable client certificate base authorization + # if you want to enable client certificate base authorization nifi.security.need.ClientAuth=true - setup the client certificate and private key PEM files + # setup the client certificate and private key PEM files nifi.security.client.certificate=./conf/client.pem nifi.security.client.private.key=./conf/client.pem - setup the client private key passphrase file + # setup the client private key passphrase file nifi.security.client.pass.phrase=./conf/password - setup the client CA certificate file + # setup the client CA certificate file nifi.security.client.ca.certificate=./conf/nifi-cert.pem - if you do not want to enable client certificate base authorization + # if you do not want to enable client certificate base authorization nifi.security.need.ClientAuth=false +It can also be configured to use the system certificate store. + + # in minifi.properties + nifi.security.use.system.cert.store=true + +Windows specific certificate options with the following default values: + + # in minifi.properties + nifi.security.windows.cert.store.location=LocalMachine + nifi.security.windows.server.cert.store=ROOT + nifi.security.windows.client.cert.store=MY + + # The CN that the client certificate is required to match; default: use the first available client certificate in the store + # nifi.security.windows.client.cert.cn= + + # Comma-separated list of enhanced key usage values that the client certificate is required to have + nifi.security.windows.client.cert.key.usage=Client Authentication + You have the option of specifying an SSL Context Service definition for the RPGs instead of the properties above. This will link to a corresponding SSL Context service defined in the flow. @@ -130,6 +225,14 @@ for TCP and secure HTTPS communications. Passphrase: <passphrase path or passphrase> CA Certificate: <CA cert path> +If the SSL certificates are not provided with an absolute path or cannot be found on the given relative path, MiNiFi will try to find them on the default path provided in the configuration file. + + # in minifi.properties + + # default minifi resource path + nifi.default.directory=/path/to/cert/files/ + + ### HTTP SiteToSite Configuration To enable HTTPSiteToSite for a remote process group. Remote Processing Groups: @@ -150,6 +253,40 @@ To enable HTTP Proxy for a remote process group. ### Command and Control Configuration Please see the [C2 readme](C2.md) for more informatoin +### State Storage + +State storage is used for keeping the state of stateful processors like TailFile. This is done using RocksDB database, but can be configured to use a different state storage with custom options. + +The default location of the RocksDB local state storage is the `corecomponentstate` directory under the MiNiFi root directory. This can be reconfigured if other directory is preferred. + + # in minifi.properties + nifi.state.storage.local.path=/var/tmp/minifi-state/ + +To have a custom state storage one option is to configure it in the flow configuration file and set the created controller in the minifi.properties file. + + # in config.yml + Controller Services: + - name: testcontroller + id: 2438e3c8-015a-1000-79ca-83af40ec1994 + class: PersistentMapStateStorage + Properties: + Auto Persistence Interval: + - value: 0 sec + Always Persist: + - value: true + File: + - value: state.txt + + # in minifi.properties + nifi.state.storage.local=2438e3c8-015a-1000-79ca-83af40ec1994 + +Another option to define a state storage is to use the following properties in the minifi.properties file. + + # in minifi.properties + nifi.state.storage.local.class.name=PersistentMapStateStorage + nifi.state.storage.local.always.persist=true + nifi.state.storage.local.auto.persistence.interval=0 sec + ### Configuring Repository storage locations Persistent repositories, such as the Flow File repository, use a configurable path to store data. @@ -157,7 +294,7 @@ The repository locations and their defaults are defined below. By default the MI variable is used. If this is not specified we extrapolate the path and use the root installation folder. You may specify your own path in place of these defaults. - in minifi.properties + # in minifi.properties nifi.provenance.repository.directory.default=${MINIFI_HOME}/provenance_repository nifi.flowfile.repository.directory.default=${MINIFI_HOME}/flowfile_repository nifi.database.content.repository.directory.default=${MINIFI_HOME}/content_repository @@ -167,16 +304,15 @@ folder. You may specify your own path in place of these defaults. Rocksdb has an option to set compression type for its database to use less disk space. If content repository or flow file repository is set to use the rocksdb database as their storage, then we have the option to compress those repositories. On Unix operating systems `zlib`, `bzip2`, `zstd`, `lz4` and `lz4hc` compression types and on Windows `xpress` compression type is supported by MiNiFi C++. If the property is set to `auto` then `xpress` will be used on Windows, `zstd` on Unix operating systems. These options can be set in the minifi.properies file with the following pr [...] - in minifi.properties + # in minifi.properties nifi.flowfile.repository.rocksdb.compression=zlib nifi.content.repository.rocksdb.compression=auto - ### Configuring compaction for rocksdb database Rocksdb has an option to run compaction at specific intervals not just when needed. - in minifi.properties + # in minifi.properties nifi.flowfile.repository.rocksdb.compaction.period=2 min nifi.database.content.repository.rocksdb.compaction.period=2 min @@ -189,20 +325,20 @@ created into. E.g. in `minifidb:///home/user/minifi/agent_state/flowfile` a dire `/home/user/minifi/agent_state` populated with rocksdb-specific content, and in that repository a logically separate "subdatabase" is created under the name `"flowfile"`. - in minifi.properties + # in minifi.properties nifi.flowfile.repository.directory.default=minifidb://${MINIFI_HOME}/agent_state/flowfile nifi.database.content.repository.directory.default=minifidb://${MINIFI_HOME}/agent_state/content - nifi.state.management.provider.local.path=minifidb://${MINIFI_HOME}/agent_state/processor_states + nifi.state.storage.local.path=minifidb://${MINIFI_HOME}/agent_state/processor_states We should not simultaneously use the same directory with and without the `minifidb://` scheme. Moreover the `"default"` name is restricted and should not be used. - in minifi.properties + # in minifi.properties nifi.flowfile.repository.directory.default=minifidb://${MINIFI_HOME}/agent_state/flowfile nifi.database.content.repository.directory.default=${MINIFI_HOME}/agent_state ^ error: using the same database directory without the "minifidb://" scheme - nifi.state.management.provider.local.path=minifidb://${MINIFI_HOME}/agent_state/default + nifi.state.storage.local.path=minifidb://${MINIFI_HOME}/agent_state/default ^ error: "default" is restricted ### Configuring Repository encryption @@ -232,7 +368,7 @@ Each of the repositories can be configured to be volatile ( state kept in memory To configure the repositories: - in minifi.properties + # in minifi.properties # For Volatile Repositories: nifi.flowfile.repository.class.name=VolatileFlowFileRepository nifi.provenance.repository.class.name=VolatileProvenanceRepository @@ -265,6 +401,14 @@ Each of the repositories can be configured to be volatile ( state kept in memory The content repository has a default option for "minimal.locking" set to true. This will attempt to use lock free structures. This may or may not be optimal as this requires additional additional searching of the underlying vector. This may be optimal for cases where max.count is not excessively high. In cases where object permanence is low within the repositories, minimal locking will result in better performance. If there are many processors and/or timing is such that the content repo [...] +### Configuring provenance repository storage + +Provenance repository size buffer size and TTL can be configured when used with RocksDB. If not set it uses the available maximum RocksDB values. + + #in minifi.properties + nifi.provenance.repository.max.storage.size=16 MB + nifi.provenance.repository.max.storage.time=30 days + ### Provenance Reporter Add Provenance Reporting to config.yml @@ -391,6 +535,16 @@ The MQTTController Service can be configured for MQTT connectivity and provide t Max Throughput: 1,024,1024 Max Payload: 1,024,1024 +### Disk space watchdog # + +Stops MiNiFi FlowController activity (excluding C2), when the available disk space on either of the repository volumes go below stop.threshold, checked every interval, then restarts when the available space on all repository volumes reach at least restart.threshold. + + # in minifi.properties + minifi.disk.space.watchdog.enable=true + minifi.disk.space.watchdog.interval=15 sec + minifi.disk.space.watchdog.stop.threshold=100 MB + minifi.disk.space.watchdog.restart.threshold=150 MB + ### Extension configuration To notify the agent which extensions it should load see [Loading extensions](Extensions.md#Loading extensions). diff --git a/extensions/http-curl/tests/C2DescribeManifestTest.cpp b/extensions/http-curl/tests/C2DescribeManifestTest.cpp index afa2617a0..23e619946 100644 --- a/extensions/http-curl/tests/C2DescribeManifestTest.cpp +++ b/extensions/http-curl/tests/C2DescribeManifestTest.cpp @@ -64,10 +64,10 @@ int main(int argc, char **argv) { harness.getConfiguration()->set(minifi::Configuration::nifi_rest_api_password, encrypted_value); harness.getConfiguration()->set(std::string(minifi::Configuration::nifi_rest_api_password) + ".protected", utils::crypto::EncryptionType::name()); - harness.getConfiguration()->set(minifi::Configuration::nifi_server_name, "server_name"); + harness.getConfiguration()->set(minifi::Configuration::nifi_c2_agent_identifier_fallback, "c2_id_fallback"); harness.getConfiguration()->set(minifi::Configuration::nifi_framework_dir, "framework_path"); harness.getConfiguration()->set(minifi::Configuration::nifi_sensitive_props_additional_keys, - std::string(minifi::Configuration::nifi_framework_dir) + ", " + std::string(minifi::Configuration::nifi_server_name)); + std::string(minifi::Configuration::nifi_framework_dir) + ", " + std::string(minifi::Configuration::nifi_c2_agent_identifier_fallback)); harness.getConfiguration()->set(minifi::Configuration::nifi_log_appender_rolling_directory, "/var/log/minifi"); harness.setUrl(args.url, &responder); diff --git a/extensions/standard-processors/tests/unit/ConfigurationTests.cpp b/extensions/standard-processors/tests/unit/ConfigurationTests.cpp index 7ffb9b3e0..a6e2c04e7 100644 --- a/extensions/standard-processors/tests/unit/ConfigurationTests.cpp +++ b/extensions/standard-processors/tests/unit/ConfigurationTests.cpp @@ -48,7 +48,7 @@ TEST_CASE("Configuration can merge lists of property names", "[mergeProperties]" } TEST_CASE("Configuration can validate values to be assigned to specific properties", "[validatePropertyValue]") { - REQUIRE(Configuration::validatePropertyValue(Configuration::nifi_server_name, "anything is valid")); + REQUIRE(Configuration::validatePropertyValue(Configuration::nifi_c2_agent_identifier_fallback, "anything is valid")); REQUIRE_FALSE(Configuration::validatePropertyValue(Configuration::nifi_flow_configuration_encrypt, "invalid.value")); REQUIRE(Configuration::validatePropertyValue(Configuration::nifi_flow_configuration_encrypt, "true")); REQUIRE(Configuration::validatePropertyValue("random.property", "random_value")); diff --git a/libminifi/include/FlowControlProtocol.h b/libminifi/include/FlowControlProtocol.h deleted file mode 100644 index b85f16a1e..000000000 --- a/libminifi/include/FlowControlProtocol.h +++ /dev/null @@ -1,273 +0,0 @@ -/** - * @file FlowControlProtocol.h - * FlowControlProtocol class declaration - * - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -#ifndef LIBMINIFI_INCLUDE_FLOWCONTROLPROTOCOL_H_ -#define LIBMINIFI_INCLUDE_FLOWCONTROLPROTOCOL_H_ - -#include <errno.h> -#include <fcntl.h> -#include <stdio.h> -#include <sys/types.h> - -#include <chrono> -#include <cinttypes> -#include <memory> -#include <string> -#include <thread> -#include <vector> - -#include "core/logging/LoggerFactory.h" -#include "core/Property.h" -#include "properties/Configure.h" -#include "utils/file/FileUtils.h" -#include "utils/gsl.h" - -namespace org { -namespace apache { -namespace nifi { -namespace minifi { -class FlowController; - -#define DEFAULT_NIFI_SERVER_PORT 9000 -#define DEFAULT_REPORT_INTERVAL 1000 // 1 sec -#define MAX_READ_TIMEOUT 30000 // 30 seconds - -// FlowControl Protocol Msg Type -typedef enum { - REGISTER_REQ, // Device Register Request from device to server which contain device serial number, current running flow YAML version - REGISTER_RESP, // Device Register Respond from server to device, may contain new flow.YAML from server ask device to apply and also device report interval - REPORT_REQ, // Period Device Report from device to server which contain device serial number, current running flow YAML name/version and other period report info - REPORT_RESP, // Report Respond from server to device, may ask device to update flow YAML or processor property - MAX_FLOW_CONTROL_MSG_TYPE -} FlowControlMsgType; - -// FlowControl Protocol Msg Type String -static const char *FlowControlMsgTypeStr[MAX_FLOW_CONTROL_MSG_TYPE] = { "REGISTER_REQ", "REGISTER_RESP", "REPORT_REQ", "REPORT_RESP" }; - -// Flow Control Msg Type to String -inline const char *FlowControlMsgTypeToStr(FlowControlMsgType type) { - if (type < MAX_FLOW_CONTROL_MSG_TYPE) - return FlowControlMsgTypeStr[type]; - else - return nullptr; -} - -// FlowControll Protocol Msg ID (Some Messages are fix length, Some are variable length (TLV) -typedef enum { - // Fix length 8 bytes: client to server in register request, required field - FLOW_SERIAL_NUMBER, - // Flow YAML name TLV: client to server in register request and report request, required field - FLOW_YML_NAME, - // Flow YAML content, TLV: server to client in register respond, option field in case server want to ask client to load YAML from server - FLOW_YML_CONTENT, - // Fix length, 4 bytes Report interval in msec: server to client in register respond, option field - REPORT_INTERVAL, - // Processor Name TLV: server to client in report respond, option field in case server want to ask client to update processor property - PROCESSOR_NAME, - // Processor Property Name TLV: server to client in report respond, option field in case server want to ask client to update processor property - PROPERTY_NAME, - // Processor Property Value TLV: server to client in report respond, option field in case server want to ask client to update processor property - PROPERTY_VALUE, - // Report Blob TLV: client to server in report request, option field in case client want to pickyback the report blob in report request to server - REPORT_BLOB, - MAX_FLOW_MSG_ID -} FlowControlMsgID; - -// FlowControl Protocol Msg ID String -static const char *FlowControlMsgIDStr[MAX_FLOW_MSG_ID] = { "FLOW_SERIAL_NUMBER", "FLOW_YAML_NAME", "FLOW_YAML_CONTENT", "REPORT_INTERVAL", "PROCESSOR_NAME", - "PROPERTY_NAME", "PROPERTY_VALUE", "REPORT_BLOB" }; - -#define TYPE_HDR_LEN 4 // Fix Hdr Type -#define TLV_HDR_LEN 8 // Type 4 bytes and Len 4 bytes - -// FlowControl Protocol Msg Len -inline int FlowControlMsgIDEncodingLen(FlowControlMsgID id, int payLoadLen) { - if (id == FLOW_SERIAL_NUMBER) - return (TYPE_HDR_LEN + 8); - else if (id == REPORT_INTERVAL) - return (TYPE_HDR_LEN + 4); - else if (id < MAX_FLOW_MSG_ID) - return (TLV_HDR_LEN + payLoadLen); - else - return -1; -} - -// Flow Control Msg Id to String -inline const char *FlowControlMsgIdToStr(FlowControlMsgID id) { - if (id < MAX_FLOW_MSG_ID) - return FlowControlMsgIDStr[id]; - else - return nullptr; -} - -// Flow Control Respond status code -typedef enum { - RESP_SUCCESS, - RESP_TRIGGER_REGISTER, // Server respond to client report to re trigger register - RESP_START_FLOW_CONTROLLER, // Server respond to client to start flow controller - RESP_STOP_FLOW_CONTROLLER, // Server respond to client to stop flow controller - RESP_FAILURE, - MAX_RESP_CODE -} FlowControlRespCode; - -// FlowControl Resp Code str -static const char *FlowControlRespCodeStr[MAX_RESP_CODE] = { "RESP_SUCCESS", "RESP_TRIGGER_REGISTER", "RESP_START_FLOW_CONTROLLER", "RESP_STOP_FLOW_CONTROLLER", "RESP_FAILURE" }; - -// Flow Control Resp Code to String -inline const char *FlowControlRespCodeToStr(FlowControlRespCode code) { - if (code < MAX_RESP_CODE) - return FlowControlRespCodeStr[code]; - else - return nullptr; -} - -// Common FlowControlProtocol Header -typedef struct { - uint32_t msgType; // Msg Type - uint32_t seqNumber; // Seq Number to match Req with Resp - uint32_t status; // Resp Code, see FlowControlRespCode - uint32_t payloadLen; // Msg Payload length -} FlowControlProtocolHeader; - -// FlowControlProtocol Class -class FlowControlProtocol { - public: - // Constructor - /*! - * Create a new control protocol - */ - FlowControlProtocol(FlowController *controller, const std::shared_ptr<Configure> &configure) { - _controller = controller; - _socket = 0; - _serverName = "localhost"; - _serverPort = DEFAULT_NIFI_SERVER_PORT; - _registered = false; - _seqNumber = 0; - _reportInterval = DEFAULT_REPORT_INTERVAL; - running_ = false; - - std::string value; - - if (configure->get(Configure::nifi_server_name, value)) { - _serverName = value; - logger_->log_info("NiFi Server Name %s", _serverName); - } - if (configure->get(Configure::nifi_server_port, value) && core::Property::StringToInt(value, _serverPort)) { - logger_->log_info("NiFi Server Port: [%" PRIu16 "]", _serverPort); - } - if (configure->get(Configure::nifi_server_report_interval, value)) { - if (auto parsed_time = utils::timeutils::StringToDuration<std::chrono::milliseconds>(value)) { - _reportInterval = parsed_time->count(); - logger_->log_info("NiFi server report interval: [%" PRId64 "] ms", _reportInterval); - } - } else { - _reportInterval = 0; - } - } - - FlowControlProtocol(const FlowControlProtocol&) = delete; - FlowControlProtocol& operator=(FlowControlProtocol) = delete; - - // Destructor - virtual ~FlowControlProtocol() { - stop(); - if (_socket) utils::file::FileUtils::close(_socket); - } - - public: - // SendRegisterRequest and Process Register Respond, return 0 for success - int sendRegisterReq(); - // SendReportReq and Process Report Respond, return 0 for success - int sendReportReq(); - // Start the flow control protocol - void start(); - // Stop the flow control protocol - void stop(); - // Set Report BLOB for periodically report - void setReportBlob(char *blob, int len) { - std::lock_guard<std::mutex> lock(mutex_); - _reportBlob.resize(len); - memcpy(_reportBlob.data(), blob, len); - } - // Run function for the thread - static void run(FlowControlProtocol *protocol); - - private: - // Connect to the socket, return sock descriptor if success, 0 for failure - static int connectServer(const char *host, uint16_t port); - // Send Data via the socket, return -1 for failure - static int sendData(uint8_t *buf, int buflen); - // Read length into buf, return -1 for failure and 0 for EOF - int readData(uint8_t *buf, int buflen); - // Select on the socket - int selectClient(int msec) const; - // Read the header - int readHdr(FlowControlProtocolHeader *hdr); - // encode uint32_t - uint8_t *encode(uint8_t *buf, uint32_t value) { - *buf++ = (value & 0xFF000000) >> 24; - *buf++ = (value & 0x00FF0000) >> 16; - *buf++ = (value & 0x0000FF00) >> 8; - *buf++ = (value & 0x000000FF); - return buf; - } - - // encode uint32_t - uint8_t *decode(uint8_t *buf, uint32_t &value) { - value = ((buf[0] << 24) | (buf[1] << 16) | (buf[2] << 8) | (buf[3])); - return (buf + 4); - } - - // encode byte array - uint8_t *encode(uint8_t *buf, uint8_t *bufArray, int size) { - memcpy(buf, bufArray, size); - buf += size; - return buf; - } - - // encode std::string - uint8_t *encode(uint8_t *buf, const std::string& value) { - // add the \0 for size - buf = encode(buf, gsl::narrow<uint32_t>(value.size() + 1)); - buf = encode(buf, const_cast<uint8_t*>(reinterpret_cast<const uint8_t*>(value.c_str())), gsl::narrow<int>(value.size() + 1)); - return buf; - } - - std::mutex mutex_; - std::shared_ptr<core::logging::Logger> logger_ = core::logging::LoggerFactory<FlowControlProtocol>::getLogger(); - // NiFi server Name - std::string _serverName; - // NiFi server port - uint16_t _serverPort; - uint8_t _serialNumber[8] = {0}; - int _socket; // to server - int64_t _reportInterval; // in msec - bool _registered; // whether it was registered to the NiFi server - uint32_t _seqNumber; - FlowController *_controller = nullptr; - std::vector<char> _reportBlob; - std::thread _thread; - bool running_; -}; - -} // namespace minifi -} // namespace nifi -} // namespace apache -} // namespace org -#endif // LIBMINIFI_INCLUDE_FLOWCONTROLPROTOCOL_H_ diff --git a/libminifi/include/FlowController.h b/libminifi/include/FlowController.h index dbd0844f0..1d3cbf9e5 100644 --- a/libminifi/include/FlowController.h +++ b/libminifi/include/FlowController.h @@ -50,7 +50,6 @@ #include "c2/C2Agent.h" #include "CronDrivenSchedulingAgent.h" #include "EventDrivenSchedulingAgent.h" -#include "FlowControlProtocol.h" #include "FlowFileRecord.h" #include "properties/Configure.h" #include "TimerDrivenSchedulingAgent.h" @@ -188,7 +187,6 @@ class FlowController : public core::controller::ForwardingControllerServiceProvi std::unique_ptr<TimerDrivenSchedulingAgent> timer_scheduler_; std::unique_ptr<EventDrivenSchedulingAgent> event_scheduler_; std::unique_ptr<CronDrivenSchedulingAgent> cron_scheduler_; - std::unique_ptr<FlowControlProtocol> protocol_; std::chrono::steady_clock::time_point start_time_; std::shared_ptr<Configure> configuration_; std::shared_ptr<core::Repository> provenance_repo_; diff --git a/libminifi/include/properties/Configuration.h b/libminifi/include/properties/Configuration.h index 1bda9ab9b..a87e0ee78 100644 --- a/libminifi/include/properties/Configuration.h +++ b/libminifi/include/properties/Configuration.h @@ -49,7 +49,6 @@ class Configuration : public Properties { static constexpr const char *nifi_bored_yield_duration = "nifi.bored.yield.duration"; static constexpr const char *nifi_graceful_shutdown_seconds = "nifi.flowcontroller.graceful.shutdown.period"; static constexpr const char *nifi_flowcontroller_drain_timeout = "nifi.flowcontroller.drain.timeout"; - static constexpr const char *nifi_server_name = "nifi.server.name"; static constexpr const char *nifi_configuration_class_name = "nifi.flow.configuration.class.name"; static constexpr const char *nifi_flow_repository_class_name = "nifi.flowfile.repository.class.name"; static constexpr const char *nifi_flow_repository_rocksdb_compression = "nifi.flowfile.repository.rocksdb.compression"; @@ -63,8 +62,6 @@ class Configuration : public Properties { static constexpr const char *nifi_volatile_repository_options_content_max_count = "nifi.volatile.repository.options.content.max.count"; static constexpr const char *nifi_volatile_repository_options_content_max_bytes = "nifi.volatile.repository.options.content.max.bytes"; static constexpr const char *nifi_volatile_repository_options_content_minimal_locking = "nifi.volatile.repository.options.content.minimal.locking"; - static constexpr const char *nifi_server_port = "nifi.server.port"; - static constexpr const char *nifi_server_report_interval = "nifi.server.report.interval"; static constexpr const char *nifi_provenance_repository_max_storage_size = "nifi.provenance.repository.max.storage.size"; static constexpr const char *nifi_provenance_repository_max_storage_time = "nifi.provenance.repository.max.storage.time"; static constexpr const char *nifi_provenance_repository_directory_default = "nifi.provenance.repository.directory.default"; @@ -123,9 +120,6 @@ class Configuration : public Properties { static constexpr const char *nifi_c2_rest_ssl_context_service = "nifi.c2.rest.ssl.context.service"; static constexpr const char *nifi_c2_rest_heartbeat_minimize_updates = "nifi.c2.rest.heartbeat.minimize.updates"; static constexpr const char *nifi_c2_rest_request_encoding = "nifi.c2.rest.request.encoding"; - static constexpr const char *nifi_c2_mqtt_connector_service = "nifi.c2.mqtt.connector.service"; - static constexpr const char *nifi_c2_mqtt_heartbeat_topic = "nifi.c2.mqtt.heartbeat.topic"; - static constexpr const char *nifi_c2_mqtt_update_topic = "nifi.c2.mqtt.update.topic"; // state management options static constexpr const char *nifi_state_storage_local = "nifi.state.storage.local"; diff --git a/libminifi/src/Configuration.cpp b/libminifi/src/Configuration.cpp index 7ac5f12a3..62d4e49d5 100644 --- a/libminifi/src/Configuration.cpp +++ b/libminifi/src/Configuration.cpp @@ -33,7 +33,6 @@ const std::unordered_map<std::string_view, gsl::not_null<const core::PropertyVal {Configuration::nifi_bored_yield_duration, gsl::make_not_null(&core::StandardPropertyTypes::TIME_PERIOD_TYPE)}, {Configuration::nifi_graceful_shutdown_seconds, gsl::make_not_null(&core::StandardPropertyTypes::TIME_PERIOD_TYPE)}, {Configuration::nifi_flowcontroller_drain_timeout, gsl::make_not_null(&core::StandardPropertyTypes::TIME_PERIOD_TYPE)}, - {Configuration::nifi_server_name, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)}, {Configuration::nifi_configuration_class_name, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)}, {Configuration::nifi_flow_repository_class_name, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)}, {Configuration::nifi_flow_repository_rocksdb_compression, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)}, @@ -47,8 +46,6 @@ const std::unordered_map<std::string_view, gsl::not_null<const core::PropertyVal {Configuration::nifi_volatile_repository_options_content_max_count, gsl::make_not_null(&core::StandardPropertyTypes::UNSIGNED_INT_TYPE)}, {Configuration::nifi_volatile_repository_options_content_max_bytes, gsl::make_not_null(&core::StandardPropertyTypes::DATA_SIZE_TYPE)}, {Configuration::nifi_volatile_repository_options_content_minimal_locking, gsl::make_not_null(&core::StandardPropertyTypes::BOOLEAN_TYPE)}, - {Configuration::nifi_server_port, gsl::make_not_null(&core::StandardPropertyTypes::PORT_TYPE)}, - {Configuration::nifi_server_report_interval, gsl::make_not_null(&core::StandardPropertyTypes::TIME_PERIOD_TYPE)}, {Configuration::nifi_provenance_repository_max_storage_size, gsl::make_not_null(&core::StandardPropertyTypes::DATA_SIZE_TYPE)}, {Configuration::nifi_provenance_repository_max_storage_time, gsl::make_not_null(&core::StandardPropertyTypes::TIME_PERIOD_TYPE)}, {Configuration::nifi_provenance_repository_directory_default, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)}, @@ -98,9 +95,6 @@ const std::unordered_map<std::string_view, gsl::not_null<const core::PropertyVal {Configuration::nifi_c2_rest_ssl_context_service, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)}, {Configuration::nifi_c2_rest_request_encoding, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)}, {Configuration::nifi_c2_rest_heartbeat_minimize_updates, gsl::make_not_null(&core::StandardPropertyTypes::BOOLEAN_TYPE)}, - {Configuration::nifi_c2_mqtt_connector_service, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)}, - {Configuration::nifi_c2_mqtt_heartbeat_topic, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)}, - {Configuration::nifi_c2_mqtt_update_topic, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)}, {Configuration::nifi_state_storage_local, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)}, {Configuration::nifi_state_storage_local_old, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)}, {Configuration::nifi_state_storage_local_class_name, gsl::make_not_null(&core::StandardPropertyTypes::VALID_TYPE)}, diff --git a/libminifi/src/FlowControlProtocol.cpp b/libminifi/src/FlowControlProtocol.cpp deleted file mode 100644 index 7f6e38db1..000000000 --- a/libminifi/src/FlowControlProtocol.cpp +++ /dev/null @@ -1,383 +0,0 @@ -/** - * @file FlowControlProtocol.cpp - * FlowControlProtocol class implementation - * - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -#include "FlowControlProtocol.h" - -#include <chrono> -#include <cstdio> -#include <ctime> -#include <thread> -#include <string> -#include <random> -#include <iostream> -#include <cinttypes> - -#include "FlowController.h" -#include "core/Core.h" -#include "utils/gsl.h" - -namespace org::apache::nifi::minifi { - -int FlowControlProtocol::connectServer(const char* /*host*/, uint16_t /*port*/) { - return 0; -} - -int FlowControlProtocol::sendData(uint8_t* /*buf*/, int /*buflen*/) { - return 0; -} - -int FlowControlProtocol::selectClient(int msec) const { - fd_set fds; - struct timeval tv{}; - int retval; - int fd = _socket; - - FD_ZERO(&fds); - FD_SET(fd, &fds); - - tv.tv_sec = msec / 1000; - tv.tv_usec = (msec % 1000) * 1000; - - if (msec > 0) - retval = select(fd + 1, &fds, nullptr, nullptr, &tv); - else - retval = select(fd + 1, &fds, nullptr, nullptr, nullptr); - - if (retval <= 0) - return retval; - if (FD_ISSET(fd, &fds)) - return retval; - else - return 0; -} - -int FlowControlProtocol::readData(uint8_t *buf, int buflen) { - gsl_Expects(buflen >= 0); - int sendSize = buflen; - - while (buflen) { - const auto selectstatus = selectClient(MAX_READ_TIMEOUT); - if (selectstatus <= 0) { - return selectstatus; - } -#ifdef WIN32 - const auto readstatus = _read(_socket, buf, buflen); -#elif !defined(__MACH__) - const auto readstatus = read(_socket, buf, gsl::narrow<size_t>(buflen)); -#else - const auto readstatus = recv(_socket, buf, buflen, 0); -#endif - if (readstatus <= 0) { - return gsl::narrow<int>(readstatus); - } - buflen -= readstatus; - buf += readstatus; - } - - return sendSize; -} - -int FlowControlProtocol::readHdr(FlowControlProtocolHeader *hdr) { - uint8_t buffer[sizeof(FlowControlProtocolHeader)]; - - uint8_t *data = buffer; - - int status = readData(buffer, sizeof(FlowControlProtocolHeader)); - if (status <= 0) - return status; - - uint32_t value; - data = this->decode(data, value); - hdr->msgType = value; - - data = this->decode(data, value); - hdr->seqNumber = value; - - data = this->decode(data, value); - hdr->status = value; - - this->decode(data, value); - hdr->payloadLen = value; - - return sizeof(FlowControlProtocolHeader); -} - -void FlowControlProtocol::start() { - if (_reportInterval <= 0) - return; - if (running_) - return; - running_ = true; - logger_->log_trace("FlowControl Protocol Start"); - _thread = std::thread(run, this); - _thread.detach(); -} - -void FlowControlProtocol::stop() { - if (!running_) - return; - running_ = false; - logger_->log_info("FlowControl Protocol Stop"); -} - -void FlowControlProtocol::run(FlowControlProtocol *protocol) { - while (protocol->running_) { - std::this_thread::sleep_for(std::chrono::milliseconds(protocol->_reportInterval)); - if (!protocol->_registered) { - // if it is not register yet - protocol->sendRegisterReq(); - } else { - protocol->sendReportReq(); - } - } -} - -int FlowControlProtocol::sendRegisterReq() { - if (_registered) { - logger_->log_debug("Already registered"); - return -1; - } - - uint16_t port = this->_serverPort; - - if (this->_socket <= 0) - this->_socket = connectServer(_serverName.c_str(), port); - - if (this->_socket <= 0) - return -1; - - // Calculate the total payload msg size - const auto payloadSize = FlowControlMsgIDEncodingLen(FLOW_SERIAL_NUMBER, 0) + FlowControlMsgIDEncodingLen(FLOW_YML_NAME, gsl::narrow<int>(this->_controller->getName().size() + 1)); - const size_t size = sizeof(FlowControlProtocolHeader) + payloadSize; - - std::vector<uint8_t> buffer; - buffer.resize(size); - uint8_t *data = buffer.data(); - - // encode the HDR - FlowControlProtocolHeader hdr; - hdr.msgType = REGISTER_REQ; - hdr.payloadLen = payloadSize; - hdr.seqNumber = this->_seqNumber; - hdr.status = RESP_SUCCESS; - data = this->encode(data, hdr.msgType); - data = this->encode(data, hdr.seqNumber); - data = this->encode(data, hdr.status); - data = this->encode(data, hdr.payloadLen); - - // encode the serial number - data = this->encode(data, FLOW_SERIAL_NUMBER); - data = this->encode(data, this->_serialNumber, 8); - - // encode the YAML name - data = this->encode(data, FLOW_YML_NAME); - this->encode(data, this->_controller->getName()); - - // send it - int status = sendData(buffer.data(), gsl::narrow<int>(size)); - buffer.clear(); - if (status <= 0) { - utils::file::FileUtils::close(_socket); - _socket = 0; - logger_->log_error("Flow Control Protocol Send Register Req failed"); - return -1; - } - - // Looking for register respond - status = readHdr(&hdr); - - if (status <= 0) { - utils::file::FileUtils::close(_socket); - _socket = 0; - logger_->log_error("Flow Control Protocol Read Register Resp header failed"); - return -1; - } - logger_->log_debug("Flow Control Protocol receive MsgType %s", FlowControlMsgTypeToStr((FlowControlMsgType) hdr.msgType)); - logger_->log_debug("Flow Control Protocol receive Seq Num %" PRIu32, hdr.seqNumber); - logger_->log_debug("Flow Control Protocol receive Resp Code %s", FlowControlRespCodeToStr((FlowControlRespCode) hdr.status)); - logger_->log_debug("Flow Control Protocol receive Payload len %" PRIu32, hdr.payloadLen); - - if (hdr.status == RESP_SUCCESS && hdr.seqNumber == this->_seqNumber) { - this->_registered = true; - this->_seqNumber++; - logger_->log_trace("Flow Control Protocol Register success"); - std::vector<uint8_t> payload; - payload.resize(hdr.payloadLen); - uint8_t *payloadPtr = payload.data(); - status = readData(payload.data(), hdr.payloadLen); - if (status <= 0) { - logger_->log_warn("Flow Control Protocol Register Read Payload fail"); - utils::file::FileUtils::close(_socket); - _socket = 0; - return -1; - } - while (payloadPtr < (payload.data() + hdr.payloadLen)) { - uint32_t msgID; - payloadPtr = this->decode(payloadPtr, msgID); - if (((FlowControlMsgID) msgID) == REPORT_INTERVAL) { - // Fixed 4 bytes - uint32_t reportInterval; - payloadPtr = this->decode(payloadPtr, reportInterval); - logger_->log_debug("Flow Control Protocol receive report interval %" PRIu32 " ms", reportInterval); - this->_reportInterval = reportInterval; - } else { - break; - } - } - utils::file::FileUtils::close(_socket); - _socket = 0; - return 0; - } else { - logger_->log_warn("Flow Control Protocol Register fail"); - utils::file::FileUtils::close(_socket); - _socket = 0; - return -1; - } -} - -int FlowControlProtocol::sendReportReq() { - uint16_t port = this->_serverPort; - - if (this->_socket <= 0) - this->_socket = connectServer(_serverName.c_str(), port); - - if (this->_socket <= 0) - return -1; - - // Calculate the total payload msg size - uint32_t payloadSize = FlowControlMsgIDEncodingLen(FLOW_YML_NAME, gsl::narrow<int>(this->_controller->getName().size() + 1)); - const size_t size = sizeof(FlowControlProtocolHeader) + payloadSize; - - std::vector<uint8_t> buffer; - buffer.resize(size); - auto* data = buffer.data(); - - // encode the HDR - FlowControlProtocolHeader hdr; - hdr.msgType = REPORT_REQ; - hdr.payloadLen = payloadSize; - hdr.seqNumber = this->_seqNumber; - hdr.status = RESP_SUCCESS; - data = this->encode(data, hdr.msgType); - data = this->encode(data, hdr.seqNumber); - data = this->encode(data, hdr.status); - data = this->encode(data, hdr.payloadLen); - - // encode the YAML name - data = this->encode(data, FLOW_YML_NAME); - this->encode(data, this->_controller->getName()); - - // send it - int status = sendData(buffer.data(), gsl::narrow<int>(size)); - buffer.clear(); - if (status <= 0) { - utils::file::FileUtils::close(_socket); - _socket = 0; - logger_->log_error("Flow Control Protocol Send Report Req failed"); - return -1; - } - - // Looking for report respond - status = readHdr(&hdr); - - if (status <= 0) { - utils::file::FileUtils::close(_socket); - _socket = 0; - logger_->log_error("Flow Control Protocol Read Report Resp header failed"); - return -1; - } - logger_->log_debug("Flow Control Protocol receive MsgType %s", FlowControlMsgTypeToStr((FlowControlMsgType) hdr.msgType)); - logger_->log_debug("Flow Control Protocol receive Seq Num %" PRIu32, hdr.seqNumber); - logger_->log_debug("Flow Control Protocol receive Resp Code %s", FlowControlRespCodeToStr((FlowControlRespCode) hdr.status)); - logger_->log_debug("Flow Control Protocol receive Payload len %" PRIu32, hdr.payloadLen); - - if (hdr.status == RESP_SUCCESS && hdr.seqNumber == this->_seqNumber) { - this->_seqNumber++; - std::vector<uint8_t> payload; - payload.resize(hdr.payloadLen); - uint8_t *payloadPtr = payload.data(); - status = readData(payload.data(), hdr.payloadLen); - if (status <= 0) { - logger_->log_warn("Flow Control Protocol Report Resp Read Payload fail"); - utils::file::FileUtils::close(_socket); - _socket = 0; - return -1; - } - std::string processor; - std::string propertyName; - std::string propertyValue; - while (payloadPtr < (payload.data() + hdr.payloadLen)) { - uint32_t msgID; - payloadPtr = this->decode(payloadPtr, msgID); - if (((FlowControlMsgID) msgID) == PROCESSOR_NAME) { - uint32_t len; - payloadPtr = this->decode(payloadPtr, len); - processor = (const char *) payloadPtr; - payloadPtr += len; - logger_->log_debug("Flow Control Protocol receive report resp processor %s", processor); - } else if (((FlowControlMsgID) msgID) == PROPERTY_NAME) { - uint32_t len; - payloadPtr = this->decode(payloadPtr, len); - propertyName = (const char *) payloadPtr; - payloadPtr += len; - logger_->log_debug("Flow Control Protocol receive report resp property name %s", propertyName); - } else if (((FlowControlMsgID) msgID) == PROPERTY_VALUE) { - uint32_t len; - payloadPtr = this->decode(payloadPtr, len); - propertyValue = (const char *) payloadPtr; - payloadPtr += len; - logger_->log_debug("Flow Control Protocol receive report resp property value %s", propertyValue); - this->_controller->updatePropertyValue(processor, propertyName, propertyValue); - } else { - break; - } - } - utils::file::FileUtils::close(_socket); - _socket = 0; - return 0; - } else if (hdr.status == RESP_TRIGGER_REGISTER && hdr.seqNumber == this->_seqNumber) { - logger_->log_trace("Flow Control Protocol trigger reregister"); - this->_registered = false; - this->_seqNumber++; - utils::file::FileUtils::close(_socket); - _socket = 0; - return 0; - } else if (hdr.status == RESP_STOP_FLOW_CONTROLLER && hdr.seqNumber == this->_seqNumber) { - logger_->log_trace("Flow Control Protocol stop flow controller"); - this->_controller->stop(); - this->_seqNumber++; - utils::file::FileUtils::close(_socket); - _socket = 0; - return 0; - } else if (hdr.status == RESP_START_FLOW_CONTROLLER && hdr.seqNumber == this->_seqNumber) { - logger_->log_trace("Flow Control Protocol start flow controller"); - this->_controller->start(); - this->_seqNumber++; - utils::file::FileUtils::close(_socket); - _socket = 0; - return 0; - } else { - logger_->log_trace("Flow Control Protocol Report fail"); - utils::file::FileUtils::close(_socket); - _socket = 0; - return -1; - } -} - -} // namespace org::apache::nifi::minifi diff --git a/libminifi/src/FlowController.cpp b/libminifi/src/FlowController.cpp index d8f19a6c0..5f26a9562 100644 --- a/libminifi/src/FlowController.cpp +++ b/libminifi/src/FlowController.cpp @@ -73,7 +73,6 @@ FlowController::FlowController(std::shared_ptr<core::Repository> provenance_repo if (flow_configuration_) { controller_service_provider_impl_ = flow_configuration_->getControllerServiceProvider(); } - protocol_ = std::make_unique<FlowControlProtocol>(this, configuration_); if (metrics_publisher_store_) { metrics_publisher_store_->initialize(this, this); } @@ -101,7 +100,6 @@ FlowController::~FlowController() { } stop(); // TODO(adebreceni): are these here on purpose, so they are destroyed first? - protocol_ = nullptr; flow_file_repo_ = nullptr; provenance_repo_ = nullptr; logger_->log_trace("Destroying FlowController"); @@ -346,7 +344,6 @@ int16_t FlowController::start() { core::logging::LoggerConfiguration::getConfiguration().initializeAlertSinks(this, configuration_); running_ = true; - protocol_->start(); content_repo_->start(); provenance_repo_->start(); flow_file_repo_->start();
