martinzink commented on a change in pull request #1116: URL: https://github.com/apache/nifi-minifi-cpp/pull/1116#discussion_r674980681
########## File path: extensions/standard-processors/tests/unit/AppendHostInfoTests.cpp ########## @@ -0,0 +1,79 @@ +/** + * + * 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 "TestBase.h" +#include "AppendHostInfo.h" +#include "LogAttribute.h" + +using AppendHostInfo = org::apache::nifi::minifi::processors::AppendHostInfo; +using LogAttribute = org::apache::nifi::minifi::processors::LogAttribute; + +TEST_CASE("AppendHostInfoTest", "[appendhostinfotest]") { + TestController testController; + std::shared_ptr<TestPlan> plan = testController.createPlan(); + LogTestController::getInstance().setTrace<processors::AppendHostInfo>(); + LogTestController::getInstance().setTrace<processors::LogAttribute>(); + std::shared_ptr<core::Processor> generate_flow_file = plan->addProcessor("GenerateFlowFile", "generate_flow_file"); + std::shared_ptr<core::Processor> append_host_info = plan->addProcessor("AppendHostInfo", "append_host_info", core::Relationship("success", "description"), true); + std::shared_ptr<core::Processor> log_attribute = plan->addProcessor("LogAttribute", "log_attributes", core::Relationship("success", "description"), true); + + plan->runNextProcessor(); // Generate + plan->runNextProcessor(); // AppendHostInfo + plan->runNextProcessor(); // LogAttribute Review comment: I agree that it looks cleaner that way. Changed it in https://github.com/apache/nifi-minifi-cpp/pull/1116/commits/19f9c59dfa2700482fc26364c78cb6c1a0fc5822 ########## File path: extensions/standard-processors/processors/AppendHostInfo.h ########## @@ -37,38 +38,43 @@ namespace nifi { namespace minifi { namespace processors { -// AppendHostInfo Class class AppendHostInfo : public core::Processor { public: - // Constructor - /*! - * Create a new processor - */ - AppendHostInfo(const std::string& name, const utils::Identifier& uuid = {}) // NOLINT + static constexpr const char* REFRESH_POLICY_ON_TRIGGER = "On every trigger"; + static constexpr const char* REFRESH_POLICY_ON_SCHEDULE = "On schedule"; + + explicit AppendHostInfo(const std::string& name, const utils::Identifier& uuid = {}) : core::Processor(name, uuid), - logger_(logging::LoggerFactory<AppendHostInfo>::getLogger()) { + logger_(logging::LoggerFactory<AppendHostInfo>::getLogger()), + refresh_on_trigger_(false) { } - // Destructor virtual ~AppendHostInfo() = default; - // Processor Name static constexpr char const* ProcessorName = "AppendHostInfo"; - // Supported Properties - static core::Property InterfaceName; + + static core::Property InterfaceNameFilter; static core::Property HostAttribute; static core::Property IPAttribute; + static core::Property RefreshPolicy; - // Supported Relationships static core::Relationship Success; public: - // OnTrigger method, implemented by NiFi AppendHostInfo - virtual void onTrigger(core::ProcessContext *context, core::ProcessSession *session); - // Initialize, over write by NiFi AppendHostInfo - virtual void initialize(void); + void onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>& sessionFactory) override; + void onTrigger(core::ProcessContext* context, core::ProcessSession* session) override; + void initialize(void) override; Review comment: sure thing, changed it in https://github.com/apache/nifi-minifi-cpp/pull/1116/commits/19f9c59dfa2700482fc26364c78cb6c1a0fc5822 ########## File path: extensions/standard-processors/processors/AppendHostInfo.cpp ########## @@ -23,90 +23,92 @@ #define __USE_POSIX #endif /* __USE_POSIX */ -#include <limits.h> -#include <string.h> #include <memory> #include <string> -#include <set> +#include <regex> +#include <algorithm> #include "core/ProcessContext.h" #include "core/Property.h" #include "core/ProcessSession.h" #include "core/FlowFile.h" #include "io/ClientSocket.h" +#include "utils/NetworkInterfaceInfo.h" + namespace org { namespace apache { namespace nifi { namespace minifi { namespace processors { -#ifndef WIN32 -#include <netdb.h> -#include <netinet/in.h> -#include <sys/socket.h> -#include <sys/ioctl.h> -#include <net/if.h> -#include <arpa/inet.h> -#endif - -#ifndef HOST_NAME_MAX -#define HOST_NAME_MAX 255 -#endif +core::Property AppendHostInfo::InterfaceNameFilter("Network Interface Filter", "A regular expression to filter ip addresses based on the name of the network interface", ""); +core::Property AppendHostInfo::HostAttribute("Hostname Attribute", "Flowfile attribute used to record the agent's hostname", "source.hostname"); +core::Property AppendHostInfo::IPAttribute("IP Attribute", "Flowfile attribute used to record the agent's IP addresses in a comma separated list", "source.ipv4"); +core::Property AppendHostInfo::RefreshPolicy(core::PropertyBuilder::createProperty("Refresh Policy") + ->withDescription("When to recalculate the host info") + ->withAllowableValues<std::string>({ REFRESH_POLICY_ON_SCHEDULE, REFRESH_POLICY_ON_TRIGGER }) + ->withDefaultValue(REFRESH_POLICY_ON_SCHEDULE)->build()); -core::Property AppendHostInfo::InterfaceName("Network Interface Name", "Network interface from which to read an IP v4 address", "eth0"); -core::Property AppendHostInfo::HostAttribute("Hostname Attribute", "Flowfile attribute to used to record the agent's hostname", "source.hostname"); -core::Property AppendHostInfo::IPAttribute("IP Attribute", "Flowfile attribute to used to record the agent's IP address", "source.ipv4"); core::Relationship AppendHostInfo::Success("success", "success operational on the flow record"); void AppendHostInfo::initialize() { - // Set the supported properties - std::set<core::Property> properties; - properties.insert(InterfaceName); - properties.insert(HostAttribute); - properties.insert(IPAttribute); - setSupportedProperties(properties); + setSupportedProperties({InterfaceNameFilter, HostAttribute, IPAttribute, RefreshPolicy}); + setSupportedRelationships({Success}); +} - // Set the supported relationships - std::set<core::Relationship> relationships; - relationships.insert(Success); - setSupportedRelationships(relationships); +void AppendHostInfo::onSchedule(const std::shared_ptr<core::ProcessContext>& context, const std::shared_ptr<core::ProcessSessionFactory>&) { + context->getProperty(HostAttribute.getName(), hostname_attribute_name_); + context->getProperty(IPAttribute.getName(), ipaddress_attribute_name_); + std::string interface_name_filter_str; + if (context->getProperty(InterfaceNameFilter.getName(), interface_name_filter_str) && !interface_name_filter_str.empty()) + interface_name_filter_.emplace(interface_name_filter_str); + else + interface_name_filter_ = utils::nullopt; + std::string refresh_policy; + context->getProperty(RefreshPolicy.getName(), refresh_policy); + if (refresh_policy == REFRESH_POLICY_ON_TRIGGER) + refresh_on_trigger_ = true; + + if (!refresh_on_trigger_) + refreshHostInfo(); } -void AppendHostInfo::onTrigger(core::ProcessContext *context, core::ProcessSession *session) { +void AppendHostInfo::onTrigger(core::ProcessContext*, core::ProcessSession* session) { std::shared_ptr<core::FlowFile> flow = session->get(); if (!flow) return; - // Get Hostname - - std::string hostAttribute = ""; - context->getProperty(HostAttribute.getName(), hostAttribute); - flow->addAttribute(hostAttribute.c_str(), org::apache::nifi::minifi::io::Socket::getMyHostName()); + if (refresh_on_trigger_) + refreshHostInfo(); - // Get IP address for the specified interface - std::string iface; - context->getProperty(InterfaceName.getName(), iface); - // Confirm the specified interface name exists on this device -#ifndef WIN32 - if (if_nametoindex(iface.c_str()) != 0) { - struct ifreq ifr; - int fd = socket(AF_INET, SOCK_DGRAM, 0); - // Type of address to retrieve - IPv4 IP address - ifr.ifr_addr.sa_family = AF_INET; - // Copy the interface name in the ifreq structure - strncpy(ifr.ifr_name, iface.c_str(), IFNAMSIZ - 1); - ioctl(fd, SIOCGIFADDR, &ifr); - close(fd); - - std::string ipAttribute; - context->getProperty(IPAttribute.getName(), ipAttribute); - flow->addAttribute(ipAttribute.c_str(), inet_ntoa(((struct sockaddr_in *) &ifr.ifr_addr)->sin_addr)); + flow->addAttribute(hostname_attribute_name_, hostname_); + if (ipaddresses_.has_value()) { + flow->addAttribute(ipaddress_attribute_name_, ipaddresses_.value()); } -#endif - // Transfer to the relationship session->transfer(flow, Success); } +void AppendHostInfo::refreshHostInfo() { + hostname_ = org::apache::nifi::minifi::io::Socket::getMyHostName(); + auto filter = [this](const utils::NetworkInterfaceInfo& interface_info) -> bool { + bool has_ipv4_address = interface_info.hasIpV4Address(); + bool matches_regex_or_empty_regex = (!interface_name_filter_.has_value()) || std::regex_match(interface_info.getName(), interface_name_filter_.value()); + return has_ipv4_address && matches_regex_or_empty_regex; + }; + auto network_interface_infos = utils::NetworkInterfaceInfo::getNetworkInterfaceInfos(filter); + std::ostringstream oss; + if (network_interface_infos.size() == 0) { + ipaddresses_ = utils::nullopt; + } else { + for (auto& network_interface_info : network_interface_infos) { + auto& ip_v4_addresses = network_interface_info.getIpV4Addresses(); + std::copy(std::begin(ip_v4_addresses), std::end(ip_v4_addresses), std::ostream_iterator<std::string>(oss, ",")); + } + ipaddresses_ = oss.str(); + ipaddresses_.value().pop_back(); // to remove trailing comma Review comment: That couldnt happen because there is a filter before this which makes sure that only network interface infos with ipv4 addresses are returned. `has_ipv4_address && matches_regex_or_empty_regex;` ########## File path: extensions/pdh/PerformanceDataMonitor.cpp ########## @@ -136,13 +137,9 @@ void PerformanceDataMonitor::initialize() { rapidjson::Value& PerformanceDataMonitor::prepareJSONBody(rapidjson::Document& root) { switch (output_format_) { case OutputFormat::OPENTELEMETRY: { - root.AddMember("Name", "PerformanceData", root.GetAllocator()); - root.AddMember("Timestamp", std::time(0), root.GetAllocator()); - std::string hostname = io::Socket::getMyHostName(); - rapidjson::Value hostname_value; - hostname_value.SetString(hostname.c_str(), hostname.length(), root.GetAllocator()); - root.AddMember("Hostname", hostname_value, root.GetAllocator()); - root.AddMember("Body", rapidjson::Value{ rapidjson::kObjectType }, root.GetAllocator()); + utils::OpenTelemetryLogDataModel::appendEventInformation(root, "PerformanceData"); + utils::OpenTelemetryLogDataModel::appendHostInformation(root); Review comment: Yeah it should have been that in the first place based on https://github.com/open-telemetry/oteps/blob/main/text/logs/0097-log-data-model.md#field-resource -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
