| /** |
| * @file LogAttribute.cpp |
| * LogAttribute 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 "LogAttribute.h" |
| #include <ctime> |
| #include <cstring> |
| #include <memory> |
| #include <string> |
| #include <vector> |
| #include <queue> |
| #include <map> |
| #include <sstream> |
| #include <iostream> |
| #include "utils/TimeUtil.h" |
| #include "utils/StringUtils.h" |
| #include "core/ProcessContext.h" |
| #include "core/ProcessSession.h" |
| #include "core/Resource.h" |
| |
| namespace org::apache::nifi::minifi::processors { |
| |
| void LogAttribute::initialize() { |
| setSupportedProperties(Properties); |
| setSupportedRelationships(Relationships); |
| } |
| |
| void LogAttribute::onSchedule(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSessionFactory>& /*factory*/) { |
| context->getProperty(FlowFilesToLog, flowfiles_to_log_); |
| logger_->log_debug("FlowFiles To Log: %llu", flowfiles_to_log_); |
| |
| context->getProperty(HexencodePayload, hexencode_); |
| |
| context->getProperty(MaxPayloadLineLength, max_line_length_); |
| logger_->log_debug("Maximum Payload Line Length: %u", max_line_length_); |
| } |
| // OnTrigger method, implemented by NiFi LogAttribute |
| void LogAttribute::onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) { |
| logger_->log_trace("enter log attribute, attempting to retrieve %u flow files", flowfiles_to_log_); |
| std::string dashLine = "--------------------------------------------------"; |
| LogAttrLevel level = LogAttrLevelInfo; |
| bool logPayload = false; |
| |
| uint64_t i = 0; |
| const auto max = flowfiles_to_log_ == 0 ? UINT64_MAX : flowfiles_to_log_; |
| for (; i < max; ++i) { |
| std::shared_ptr<core::FlowFile> flow = session->get(); |
| |
| if (!flow) { |
| break; |
| } |
| |
| std::string value; |
| if (context->getProperty(LogLevel, value)) { |
| logLevelStringToEnum(value, level); |
| } |
| if (context->getProperty(LogPrefix, value)) { |
| dashLine = "-----" + value + "-----"; |
| } |
| |
| context->getProperty(LogPayload, logPayload); |
| |
| std::ostringstream message; |
| message << "Logging for flow file " << "\n"; |
| message << dashLine; |
| message << "\nStandard FlowFile Attributes"; |
| message << "\n" << "UUID:" << flow->getUUIDStr(); |
| message << "\n" << "EntryDate:" << utils::timeutils::getTimeStr(flow->getEntryDate()); |
| message << "\n" << "lineageStartDate:" << utils::timeutils::getTimeStr(flow->getlineageStartDate()); |
| message << "\n" << "Size:" << flow->getSize() << " Offset:" << flow->getOffset(); |
| message << "\nFlowFile Attributes Map Content"; |
| std::map<std::string, std::string> attrs = flow->getAttributes(); |
| std::map<std::string, std::string>::iterator it; |
| for (it = attrs.begin(); it != attrs.end(); it++) { |
| message << "\n" << "key:" << it->first << " value:" << it->second; |
| } |
| message << "\nFlowFile Resource Claim Content"; |
| std::shared_ptr<ResourceClaim> claim = flow->getResourceClaim(); |
| if (claim) { |
| message << "\n" << "Content Claim:" << claim->getContentFullPath(); |
| } |
| if (logPayload && flow->getSize() <= 1024 * 1024) { |
| message << "\n" << "Payload:" << "\n"; |
| const auto read_result = session->readBuffer(flow); |
| |
| std::string printable_payload; |
| if (hexencode_) { |
| printable_payload = utils::StringUtils::to_hex(read_result.buffer); |
| } else { |
| printable_payload = to_string(read_result); |
| } |
| |
| if (max_line_length_ == 0U) { |
| message << printable_payload << "\n"; |
| } else { |
| for (size_t i = 0; i < printable_payload.size(); i += max_line_length_) { |
| message << printable_payload.substr(i, max_line_length_) << '\n'; |
| } |
| } |
| } else { |
| message << "\n"; |
| } |
| message << dashLine; |
| std::string output = message.str(); |
| |
| switch (level) { |
| case LogAttrLevelInfo: |
| core::logging::LOG_INFO(logger_) << output; |
| break; |
| case LogAttrLevelDebug: |
| core::logging::LOG_DEBUG(logger_) << output; |
| break; |
| case LogAttrLevelError: |
| core::logging::LOG_ERROR(logger_) << output; |
| break; |
| case LogAttrLevelTrace: |
| core::logging::LOG_TRACE(logger_) << output; |
| break; |
| case LogAttrLevelWarn: |
| core::logging::LOG_WARN(logger_) << output; |
| break; |
| default: |
| break; |
| } |
| session->transfer(flow, Success); |
| } |
| logger_->log_debug("Logged %d flow files", i); |
| } |
| |
| REGISTER_RESOURCE(LogAttribute, Processor); |
| |
| } // namespace org::apache::nifi::minifi::processors |