You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by al...@apache.org on 2018/01/20 04:39:20 UTC
[1/3] nifi-minifi-cpp git commit: MINIFICPP-365 Adjusting log levels.
Repository: nifi-minifi-cpp
Updated Branches:
refs/heads/master 8207e959d -> 233d1d44c
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/sitetosite/SiteToSiteClient.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/sitetosite/SiteToSiteClient.cpp b/libminifi/src/sitetosite/SiteToSiteClient.cpp
index 1f98655..2b103a7 100644
--- a/libminifi/src/sitetosite/SiteToSiteClient.cpp
+++ b/libminifi/src/sitetosite/SiteToSiteClient.cpp
@@ -99,7 +99,7 @@ void SiteToSiteClient::deleteTransaction(std::string transactionID) {
transaction = it->second;
}
- logger_->log_debug("Site2Site delete transaction %s", transaction->getUUIDStr().c_str());
+ logger_->log_debug("Site2Site delete transaction %s", transaction->getUUIDStr());
known_transactions_.erase(transactionID);
}
@@ -188,7 +188,7 @@ bool SiteToSiteClient::transferFlowFiles(const std::shared_ptr<core::ProcessCont
throw Exception(SITE2SITE_EXCEPTION, "Send Failed");
}
- logger_->log_debug("Site2Site transaction %s send flow record %s", transactionID.c_str(), flow->getUUIDStr().c_str());
+ logger_->log_debug("Site2Site transaction %s send flow record %s", transactionID, flow->getUUIDStr());
if (resp == 0) {
uint64_t endTime = getTimeMillis();
std::string transitUri = peer_->getURL() + "/" + flow->getUUIDStr();
@@ -218,7 +218,7 @@ bool SiteToSiteClient::transferFlowFiles(const std::shared_ptr<core::ProcessCont
ss << "Complete Failed for " << transactionID;
throw Exception(SITE2SITE_EXCEPTION, ss.str().c_str());
}
- logger_->log_debug("Site2Site transaction %s successfully send flow record %d, content bytes %llu", transactionID.c_str(), transaction->total_transfers_, transaction->_bytes);
+ logger_->log_debug("Site2Site transaction %s successfully send flow record %d, content bytes %llu", transactionID, transaction->total_transfers_, transaction->_bytes);
} catch (std::exception &exception) {
if (transaction)
deleteTransaction(transactionID);
@@ -280,7 +280,7 @@ bool SiteToSiteClient::confirm(std::string transactionID) {
// time window involved in the entire transaction, it is reduced to a simple round-trip conversation.
int64_t crcValue = transaction->getCRC();
std::string crc = std::to_string(crcValue);
- logger_->log_debug("Site2Site Send confirm with CRC %d to transaction %s", transaction->getCRC(), transactionID.c_str());
+ logger_->log_debug("Site2Site Send confirm with CRC %d to transaction %s", transaction->getCRC(), transactionID);
ret = writeResponse(transaction, CONFIRM_TRANSACTION, crc);
if (ret <= 0)
return false;
@@ -291,18 +291,18 @@ bool SiteToSiteClient::confirm(std::string transactionID) {
return false;
if (code == CONFIRM_TRANSACTION) {
- logger_->log_debug("Site2Site transaction %s peer confirm transaction", transactionID.c_str());
+ logger_->log_debug("Site2Site transaction %s peer confirm transaction", transactionID);
transaction->_state = TRANSACTION_CONFIRMED;
return true;
} else if (code == BAD_CHECKSUM) {
- logger_->log_debug("Site2Site transaction %s peer indicate bad checksum", transactionID.c_str());
+ logger_->log_debug("Site2Site transaction %s peer indicate bad checksum", transactionID);
return false;
} else {
- logger_->log_debug("Site2Site transaction %s peer unknown respond code %d", transactionID.c_str(), code);
+ logger_->log_debug("Site2Site transaction %s peer unknown respond code %d", transactionID, code);
return false;
}
} else {
- logger_->log_debug("Site2Site Send FINISH TRANSACTION for transaction %s", transactionID.c_str());
+ logger_->log_debug("Site2Site Send FINISH TRANSACTION for transaction %s", transactionID);
ret = writeResponse(transaction, FINISH_TRANSACTION, "FINISH_TRANSACTION");
if (ret <= 0)
return false;
@@ -312,19 +312,19 @@ bool SiteToSiteClient::confirm(std::string transactionID) {
// we've sent a FINISH_TRANSACTION. Now we'll wait for the peer to send a 'Confirm Transaction' response
if (code == CONFIRM_TRANSACTION) {
- logger_->log_debug("Site2Site transaction %s peer confirm transaction with CRC %s", transactionID.c_str(), message.c_str());
+ logger_->log_debug("Site2Site transaction %s peer confirm transaction with CRC %s", transactionID, message);
if (this->_currentVersion > 3) {
int64_t crcValue = transaction->getCRC();
std::string crc = std::to_string(crcValue);
if (message == crc) {
- logger_->log_debug("Site2Site transaction %s CRC matched", transactionID.c_str());
+ logger_->log_debug("Site2Site transaction %s CRC matched", transactionID);
ret = writeResponse(transaction, CONFIRM_TRANSACTION, "CONFIRM_TRANSACTION");
if (ret <= 0)
return false;
transaction->_state = TRANSACTION_CONFIRMED;
return true;
} else {
- logger_->log_debug("Site2Site transaction %s CRC not matched %s", transactionID.c_str(), crc.c_str());
+ logger_->log_debug("Site2Site transaction %s CRC not matched %s", transactionID, crc);
ret = writeResponse(transaction, BAD_CHECKSUM, "BAD_CHECKSUM");
return false;
}
@@ -335,7 +335,7 @@ bool SiteToSiteClient::confirm(std::string transactionID) {
transaction->_state = TRANSACTION_CONFIRMED;
return true;
} else {
- logger_->log_debug("Site2Site transaction %s peer unknown respond code %d", transactionID.c_str(), code);
+ logger_->log_debug("Site2Site transaction %s peer unknown respond code %d", transactionID, code);
return false;
}
return false;
@@ -413,7 +413,7 @@ bool SiteToSiteClient::complete(std::string transactionID) {
transaction->_state = TRANSACTION_COMPLETED;
return true;
} else {
- logger_->log_debug("Site2Site transaction %s send finished", transactionID.c_str());
+ logger_->log_debug("Site2Site transaction %s send finished", transactionID);
ret = this->writeResponse(transaction, TRANSACTION_FINISHED, "Finished");
if (ret <= 0) {
return false;
@@ -433,11 +433,11 @@ bool SiteToSiteClient::complete(std::string transactionID) {
return false;
if (code == TRANSACTION_FINISHED) {
- logger_->log_debug("Site2Site transaction %s peer finished transaction", transactionID.c_str());
+ logger_->log_info("Site2Site transaction %s peer finished transaction", transactionID);
transaction->_state = TRANSACTION_COMPLETED;
return true;
} else {
- logger_->log_debug("Site2Site transaction %s peer unknown respond code %d", transactionID.c_str(), code);
+ logger_->log_warn("Site2Site transaction %s peer unknown respond code %d", transactionID, code);
return false;
}
}
@@ -467,12 +467,12 @@ int16_t SiteToSiteClient::send(std::string transactionID, DataPacket *packet, co
}
if (transaction->getState() != TRANSACTION_STARTED && transaction->getState() != DATA_EXCHANGED) {
- logger_->log_debug("Site2Site transaction %s is not at started or exchanged state", transactionID.c_str());
+ logger_->log_warn("Site2Site transaction %s is not at started or exchanged state", transactionID);
return -1;
}
if (transaction->getDirection() != SEND) {
- logger_->log_debug("Site2Site transaction %s direction is wrong", transactionID.c_str());
+ logger_->log_warn("Site2Site transaction %s direction is wrong", transactionID);
return -1;
}
@@ -500,7 +500,7 @@ int16_t SiteToSiteClient::send(std::string transactionID, DataPacket *packet, co
if (ret <= 0) {
return -1;
}
- logger_->log_debug("Site2Site transaction %s send attribute key %s value %s", transactionID.c_str(), itAttribute->first.c_str(), itAttribute->second.c_str());
+ logger_->log_debug("Site2Site transaction %s send attribute key %s value %s", transactionID, itAttribute->first, itAttribute->second);
}
uint64_t len = 0;
@@ -515,15 +515,15 @@ int16_t SiteToSiteClient::send(std::string transactionID, DataPacket *packet, co
sitetosite::ReadCallback callback(packet);
session->read(flowFile, &callback);
if (flowFile->getSize() != packet->_size) {
- logger_->log_debug("MisMatched sizes %llu %llu", flowFile->getSize(), packet->_size);
+ logger_->log_debug("Mismatched sizes %llu %llu", flowFile->getSize(), packet->_size);
return -2;
}
}
if (packet->payload_.length() == 0 && len == 0) {
if (flowFile->getResourceClaim() == nullptr)
- logger_->log_debug("no claim");
+ logger_->log_trace("no claim");
else
- logger_->log_debug("Flowfile empty %s", flowFile->getResourceClaim()->getContentFullPath());
+ logger_->log_trace("Flowfile empty %s", flowFile->getResourceClaim()->getContentFullPath());
}
} else if (packet->payload_.length() > 0) {
len = packet->payload_.length();
@@ -545,7 +545,9 @@ int16_t SiteToSiteClient::send(std::string transactionID, DataPacket *packet, co
transaction->total_transfers_++;
transaction->_state = DATA_EXCHANGED;
transaction->_bytes += len;
- logger_->log_debug("Site2Site transaction %s send flow record %d, total length %llu, added %llu", transactionID.c_str(), transaction->total_transfers_, transaction->_bytes, len);
+
+ logging::LOG_INFO(logger_) << "Site to Site transaction " << transactionID << " sent flow " << transaction->total_transfers_
+ << "flow records, with total size " << transaction->_bytes;
return 0;
}
@@ -571,12 +573,12 @@ bool SiteToSiteClient::receive(std::string transactionID, DataPacket *packet, bo
}
if (transaction->getState() != TRANSACTION_STARTED && transaction->getState() != DATA_EXCHANGED) {
- logger_->log_debug("Site2Site transaction %s is not at started or exchanged state", transactionID.c_str());
+ logger_->log_warn("Site2Site transaction %s is not at started or exchanged state", transactionID);
return false;
}
if (transaction->getDirection() != RECEIVE) {
- logger_->log_debug("Site2Site transaction %s direction is wrong", transactionID.c_str());
+ logger_->log_warn("Site2Site transaction %s direction is wrong", transactionID);
return false;
}
@@ -596,15 +598,15 @@ bool SiteToSiteClient::receive(std::string transactionID, DataPacket *packet, bo
return false;
}
if (code == CONTINUE_TRANSACTION) {
- logger_->log_debug("Site2Site transaction %s peer indicate continue transaction", transactionID.c_str());
+ logger_->log_debug("Site2Site transaction %s peer indicate continue transaction", transactionID);
transaction->_dataAvailable = true;
} else if (code == FINISH_TRANSACTION) {
- logger_->log_debug("Site2Site transaction %s peer indicate finish transaction", transactionID.c_str());
+ logger_->log_debug("Site2Site transaction %s peer indicate finish transaction", transactionID);
transaction->_dataAvailable = false;
eof = true;
return true;
} else {
- logger_->log_debug("Site2Site transaction %s peer indicate wrong respond code %d", transactionID.c_str(), code);
+ logger_->log_debug("Site2Site transaction %s peer indicate wrong respond code %d", transactionID, code);
return false;
}
}
@@ -623,7 +625,7 @@ bool SiteToSiteClient::receive(std::string transactionID, DataPacket *packet, bo
}
// read the attributes
- logger_->log_debug("Site2Site transaction %s receives attribute key %d", transactionID.c_str(), numAttributes);
+ logger_->log_debug("Site2Site transaction %s receives attribute key %d", transactionID, numAttributes);
for (unsigned int i = 0; i < numAttributes; i++) {
std::string key;
std::string value;
@@ -636,7 +638,7 @@ bool SiteToSiteClient::receive(std::string transactionID, DataPacket *packet, bo
return false;
}
packet->_attributes[key] = value;
- logger_->log_debug("Site2Site transaction %s receives attribute key %s value %s", transactionID.c_str(), key.c_str(), value.c_str());
+ logger_->log_debug("Site2Site transaction %s receives attribute key %s value %s", transactionID, key, value);
}
uint64_t len;
@@ -657,7 +659,8 @@ bool SiteToSiteClient::receive(std::string transactionID, DataPacket *packet, bo
}
transaction->_state = DATA_EXCHANGED;
transaction->_bytes += len;
- logger_->log_debug("Site2Site transaction %s receives flow record %d, total length %llu, added %llu", transactionID.c_str(), transaction->total_transfers_, transaction->_bytes, len);
+ logging::LOG_INFO(logger_) << "Site to Site transaction " << transactionID << " received flow record " << transaction->total_transfers_
+ << ", total length " << transaction->_bytes << ", added " << len;
return true;
}
@@ -747,7 +750,9 @@ bool SiteToSiteClient::receiveFlowFiles(const std::shared_ptr<core::ProcessConte
transaction_str << "Complete Transaction " << transactionID << " Failed";
throw Exception(SITE2SITE_EXCEPTION, transaction_str.str().c_str());
}
- logger_->log_info("Site2Site transaction %s successfully receive flow record %d, content bytes %llu", transactionID.c_str(), transfers, bytes);
+ logging::LOG_INFO(logger_) << "Site to Site transaction " << transactionID << " received flow record " << transfers
+ << ", with content size " << bytes << " bytes";
+
// we yield the receive if we did not get anything
if (transfers == 0)
context->yield();
@@ -756,14 +761,14 @@ bool SiteToSiteClient::receiveFlowFiles(const std::shared_ptr<core::ProcessConte
deleteTransaction(transactionID);
context->yield();
tearDown();
- logger_->log_debug("Caught Exception %s", exception.what());
+ logger_->log_warn("Caught Exception %s", exception.what());
throw;
} catch (...) {
if (transaction)
deleteTransaction(transactionID);
context->yield();
tearDown();
- logger_->log_debug("Caught Exception during RawSiteToSiteClient::receiveFlowFiles");
+ logger_->log_warn("Caught Exception during RawSiteToSiteClient::receiveFlowFiles");
throw;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/main/MiNiFiMain.cpp
----------------------------------------------------------------------
diff --git a/main/MiNiFiMain.cpp b/main/MiNiFiMain.cpp
index 7ebe972..780c4c9 100644
--- a/main/MiNiFiMain.cpp
+++ b/main/MiNiFiMain.cpp
@@ -110,7 +110,7 @@ int main(int argc, char **argv) {
}
if (signal(SIGINT, sigHandler) == SIG_ERR || signal(SIGTERM, sigHandler) == SIG_ERR || signal(SIGPIPE, SIG_IGN) == SIG_ERR) {
- logger->log_error("Can not install signal handler");
+ std::cerr << "Cannot install signal handler" << std::endl;
return -1;
}
[2/3] nifi-minifi-cpp git commit: MINIFICPP-365 Adjusting log levels.
Posted by al...@apache.org.
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/FlowControlProtocol.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/FlowControlProtocol.cpp b/libminifi/src/FlowControlProtocol.cpp
index b5fc928..60122b0 100644
--- a/libminifi/src/FlowControlProtocol.cpp
+++ b/libminifi/src/FlowControlProtocol.cpp
@@ -107,7 +107,7 @@ int FlowControlProtocol::connectServer(const char *host, uint16_t port) {
return 0;
}
- logger_->log_info("Flow Control Protocol socket %ll connect to server %s port %ll success", sock, host, port);
+ logger_->log_debug("Flow Control Protocol socket %ll connect to server %s port %ll success", sock, host, port);
return sock;
}
@@ -207,7 +207,7 @@ void FlowControlProtocol::start() {
if (running_)
return;
running_ = true;
- logger_->log_info("FlowControl Protocol Start");
+ logger_->log_trace("FlowControl Protocol Start");
_thread = new std::thread(run, this);
_thread->detach();
}
@@ -234,7 +234,7 @@ void FlowControlProtocol::run(FlowControlProtocol *protocol) {
int FlowControlProtocol::sendRegisterReq() {
if (_registered) {
- logger_->log_info("Already registered");
+ logger_->log_debug("Already registered");
return -1;
}
@@ -291,21 +291,21 @@ int FlowControlProtocol::sendRegisterReq() {
logger_->log_error("Flow Control Protocol Read Register Resp header failed");
return -1;
}
- logger_->log_info("Flow Control Protocol receive MsgType %s", FlowControlMsgTypeToStr((FlowControlMsgType) hdr.msgType));
- logger_->log_info("Flow Control Protocol receive Seq Num %ll", hdr.seqNumber);
- logger_->log_info("Flow Control Protocol receive Resp Code %s", FlowControlRespCodeToStr((FlowControlRespCode) hdr.status));
- logger_->log_info("Flow Control Protocol receive Payload len %ll", hdr.payloadLen);
+ logger_->log_debug("Flow Control Protocol receive MsgType %s", FlowControlMsgTypeToStr((FlowControlMsgType) hdr.msgType));
+ logger_->log_debug("Flow Control Protocol receive Seq Num %ll", 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 %ll", hdr.payloadLen);
if (hdr.status == RESP_SUCCESS && hdr.seqNumber == this->_seqNumber) {
this->_registered = true;
this->_seqNumber++;
- logger_->log_info("Flow Control Protocol Register success");
+ logger_->log_trace("Flow Control Protocol Register success");
uint8_t *payload = new uint8_t[hdr.payloadLen];
uint8_t *payloadPtr = payload;
status = readData(payload, hdr.payloadLen);
if (status <= 0) {
delete[] payload;
- logger_->log_info("Flow Control Protocol Register Read Payload fail");
+ logger_->log_warn("Flow Control Protocol Register Read Payload fail");
close(_socket);
_socket = 0;
return -1;
@@ -317,7 +317,7 @@ int FlowControlProtocol::sendRegisterReq() {
// Fixed 4 bytes
uint32_t reportInterval;
payloadPtr = this->decode(payloadPtr, reportInterval);
- logger_->log_info("Flow Control Protocol receive report interval %ll ms", reportInterval);
+ logger_->log_debug("Flow Control Protocol receive report interval %ll ms", reportInterval);
this->_reportInterval = reportInterval;
} else {
break;
@@ -328,7 +328,7 @@ int FlowControlProtocol::sendRegisterReq() {
_socket = 0;
return 0;
} else {
- logger_->log_info("Flow Control Protocol Register fail");
+ logger_->log_warn("Flow Control Protocol Register fail");
close(_socket);
_socket = 0;
return -1;
@@ -385,10 +385,10 @@ int FlowControlProtocol::sendReportReq() {
logger_->log_error("Flow Control Protocol Read Report Resp header failed");
return -1;
}
- logger_->log_info("Flow Control Protocol receive MsgType %s", FlowControlMsgTypeToStr((FlowControlMsgType) hdr.msgType));
- logger_->log_info("Flow Control Protocol receive Seq Num %ll", hdr.seqNumber);
- logger_->log_info("Flow Control Protocol receive Resp Code %s", FlowControlRespCodeToStr((FlowControlRespCode) hdr.status));
- logger_->log_info("Flow Control Protocol receive Payload len %ll", hdr.payloadLen);
+ logger_->log_debug("Flow Control Protocol receive MsgType %s", FlowControlMsgTypeToStr((FlowControlMsgType) hdr.msgType));
+ logger_->log_debug("Flow Control Protocol receive Seq Num %ll", 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 %ll", hdr.payloadLen);
if (hdr.status == RESP_SUCCESS && hdr.seqNumber == this->_seqNumber) {
this->_seqNumber++;
@@ -397,7 +397,7 @@ int FlowControlProtocol::sendReportReq() {
status = readData(payload, hdr.payloadLen);
if (status <= 0) {
delete[] payload;
- logger_->log_info("Flow Control Protocol Report Resp Read Payload fail");
+ logger_->log_warn("Flow Control Protocol Report Resp Read Payload fail");
close(_socket);
_socket = 0;
return -1;
@@ -413,19 +413,19 @@ int FlowControlProtocol::sendReportReq() {
payloadPtr = this->decode(payloadPtr, len);
processor = (const char *) payloadPtr;
payloadPtr += len;
- logger_->log_info("Flow Control Protocol receive report resp processor %s", processor.c_str());
+ 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_info("Flow Control Protocol receive report resp property name %s", propertyName.c_str());
+ 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_info("Flow Control Protocol receive report resp property value %s", propertyValue.c_str());
+ logger_->log_debug("Flow Control Protocol receive report resp property value %s", propertyValue);
this->_controller->updatePropertyValue(processor, propertyName, propertyValue);
} else {
break;
@@ -436,28 +436,28 @@ int FlowControlProtocol::sendReportReq() {
_socket = 0;
return 0;
} else if (hdr.status == RESP_TRIGGER_REGISTER && hdr.seqNumber == this->_seqNumber) {
- logger_->log_info("Flow Control Protocol trigger reregister");
+ logger_->log_trace("Flow Control Protocol trigger reregister");
this->_registered = false;
this->_seqNumber++;
close(_socket);
_socket = 0;
return 0;
} else if (hdr.status == RESP_STOP_FLOW_CONTROLLER && hdr.seqNumber == this->_seqNumber) {
- logger_->log_info("Flow Control Protocol stop flow controller");
+ logger_->log_trace("Flow Control Protocol stop flow controller");
this->_controller->stop(true);
this->_seqNumber++;
close(_socket);
_socket = 0;
return 0;
} else if (hdr.status == RESP_START_FLOW_CONTROLLER && hdr.seqNumber == this->_seqNumber) {
- logger_->log_info("Flow Control Protocol start flow controller");
+ logger_->log_trace("Flow Control Protocol start flow controller");
this->_controller->start();
this->_seqNumber++;
close(_socket);
_socket = 0;
return 0;
} else {
- logger_->log_info("Flow Control Protocol Report fail");
+ logger_->log_trace("Flow Control Protocol Report fail");
close(_socket);
_socket = 0;
return -1;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/FlowController.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/FlowController.cpp b/libminifi/src/FlowController.cpp
index 4d2e1b7..e6118e6 100644
--- a/libminifi/src/FlowController.cpp
+++ b/libminifi/src/FlowController.cpp
@@ -135,7 +135,7 @@ void FlowController::initializePaths(const std::string &adjustedFilename) {
}
std::string pathString(path);
configuration_filename_ = pathString;
- logger_->log_info("FlowController NiFi Configuration file %s", pathString.c_str());
+ logger_->log_info("FlowController NiFi Configuration file %s", pathString);
// Create the content repo directory if needed
struct stat contentDirStat;
@@ -181,7 +181,7 @@ bool FlowController::applyConfiguration(const std::string &configurePayload) {
if (newRoot == nullptr)
return false;
- logger_->log_info("Starting to reload Flow Controller with flow control name %s, version %d", newRoot->getName().c_str(), newRoot->getVersion());
+ logger_->log_info("Starting to reload Flow Controller with flow control name %s, version %d", newRoot->getName(), newRoot->getVersion());
std::lock_guard<std::recursive_mutex> flow_lock(mutex_);
stop(true);
@@ -287,7 +287,7 @@ void FlowController::load() {
void FlowController::reload(std::string yamlFile) {
std::lock_guard<std::recursive_mutex> flow_lock(mutex_);
- logger_->log_info("Starting to reload Flow Controller with yaml %s", yamlFile.c_str());
+ logger_->log_info("Starting to reload Flow Controller with yaml %s", yamlFile);
stop(true);
unload();
std::string oldYamlFile = this->configuration_filename_;
@@ -296,7 +296,7 @@ void FlowController::reload(std::string yamlFile) {
start();
if (this->root_ != nullptr) {
this->configuration_filename_ = oldYamlFile;
- logger_->log_info("Rollback Flow Controller to YAML %s", oldYamlFile.c_str());
+ logger_->log_info("Rollback Flow Controller to YAML %s", oldYamlFile);
stop(true);
unload();
load();
@@ -405,7 +405,7 @@ void FlowController::initializeC2() {
auto ptr = core::ClassLoader::getDefaultClassLoader().instantiate(clazz, clazz);
if (nullptr == ptr) {
- logger_->log_error("No metric defined for %s", clazz.c_str());
+ logger_->log_error("No metric defined for %s", clazz);
continue;
}
@@ -454,7 +454,7 @@ void FlowController::initializeC2() {
ret = metrics_[clazz];
}
if (nullptr == ret) {
- logger_->log_error("No metric defined for %s", clazz.c_str());
+ logger_->log_error("No metric defined for %s", clazz);
continue;
}
component_metrics_by_id_[id].push_back(ret);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/FlowFileRecord.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/FlowFileRecord.cpp b/libminifi/src/FlowFileRecord.cpp
index 3816485..8775de7 100644
--- a/libminifi/src/FlowFileRecord.cpp
+++ b/libminifi/src/FlowFileRecord.cpp
@@ -98,11 +98,11 @@ FlowFileRecord::FlowFileRecord(std::shared_ptr<core::Repository> flow_repository
}
FlowFileRecord::~FlowFileRecord() {
- logger_->log_debug("Destroying flow file record, UUID %s", uuidStr_.c_str());
+ logger_->log_debug("Destroying flow file record, UUID %s", uuidStr_);
if (!snapshot_)
- logger_->log_debug("Delete FlowFile UUID %s", uuidStr_.c_str());
+ logger_->log_debug("Delete FlowFile UUID %s", uuidStr_);
else
- logger_->log_debug("Delete SnapShot FlowFile UUID %s", uuidStr_.c_str());
+ logger_->log_debug("Delete SnapShot FlowFile UUID %s", uuidStr_);
if (claim_) {
releaseClaim(claim_);
} else {
@@ -119,11 +119,11 @@ void FlowFileRecord::releaseClaim(std::shared_ptr<ResourceClaim> claim) {
// Decrease the flow file record owned count for the resource claim
claim_->decreaseFlowFileRecordOwnedCount();
std::string value;
- logger_->log_debug("Delete Resource Claim %s, %s, attempt %llu", getUUIDStr(), claim_->getContentFullPath().c_str(), claim_->getFlowFileRecordOwnedCount());
+ logger_->log_debug("Delete Resource Claim %s, %s, attempt %llu", getUUIDStr(), claim_->getContentFullPath(), claim_->getFlowFileRecordOwnedCount());
if (claim_->getFlowFileRecordOwnedCount() <= 0) {
// we cannot rely on the stored variable here since we aren't guaranteed atomicity
if (flow_repository_ != nullptr && !flow_repository_->Get(uuidStr_, value)) {
- logger_->log_debug("Delete Resource Claim %s", claim_->getContentFullPath().c_str());
+ logger_->log_debug("Delete Resource Claim %s", claim_->getContentFullPath());
content_repo_->remove(claim_);
}
}
@@ -184,7 +184,7 @@ bool FlowFileRecord::DeSerialize(std::string key) {
ret = flow_repository_->Get(key, value);
if (!ret) {
- logger_->log_error("NiFi FlowFile Store event %s can not found", key.c_str());
+ logger_->log_error("NiFi FlowFile Store event %s can not found", key);
return false;
}
io::DataStream stream((const uint8_t*) value.data(), value.length());
@@ -192,9 +192,9 @@ bool FlowFileRecord::DeSerialize(std::string key) {
ret = DeSerialize(stream);
if (ret) {
- logger_->log_debug("NiFi FlowFile retrieve uuid %s size %llu connection %s success", uuidStr_.c_str(), stream.getSize(), uuid_connection_.c_str());
+ logger_->log_debug("NiFi FlowFile retrieve uuid %s size %llu connection %s success", uuidStr_, stream.getSize(), uuid_connection_);
} else {
- logger_->log_debug("NiFi FlowFile retrieve uuid %s size %llu connection %s fail", uuidStr_.c_str(), stream.getSize(), uuid_connection_.c_str());
+ logger_->log_debug("NiFi FlowFile retrieve uuid %s size %llu connection %s fail", uuidStr_, stream.getSize(), uuid_connection_);
}
return ret;
@@ -263,10 +263,10 @@ bool FlowFileRecord::Serialize() {
}
if (flow_repository_->Put(uuidStr_, const_cast<uint8_t*>(outStream.getBuffer()), outStream.getSize())) {
- logger_->log_debug("NiFi FlowFile Store event %s size %llu success", uuidStr_.c_str(), outStream.getSize());
+ logger_->log_debug("NiFi FlowFile Store event %s size %llu success", uuidStr_, outStream.getSize());
return true;
} else {
- logger_->log_error("NiFi FlowFile Store event %s size %llu fail", uuidStr_.c_str(), outStream.getSize());
+ logger_->log_error("NiFi FlowFile Store event %s size %llu fail", uuidStr_, outStream.getSize());
return false;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/RemoteProcessorGroupPort.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/RemoteProcessorGroupPort.cpp b/libminifi/src/RemoteProcessorGroupPort.cpp
index caddc32..31b6c89 100644
--- a/libminifi/src/RemoteProcessorGroupPort.cpp
+++ b/libminifi/src/RemoteProcessorGroupPort.cpp
@@ -71,7 +71,7 @@ std::unique_ptr<sitetosite::SiteToSiteClient> RemoteProcessorGroupPort::getNextP
nextProtocol = sitetosite::createClient(config);
} else if (peer_index_ >= 0) {
std::lock_guard<std::mutex> lock(peer_mutex_);
- logger_->log_info("Creating client from peer %ll", peer_index_.load());
+ logger_->log_debug("Creating client from peer %ll", peer_index_.load());
sitetosite::SiteToSiteClientConfiguration config(stream_factory_, peers_[this->peer_index_].getPeer(), client_type_);
peer_index_++;
@@ -81,12 +81,12 @@ std::unique_ptr<sitetosite::SiteToSiteClient> RemoteProcessorGroupPort::getNextP
nextProtocol = sitetosite::createClient(config);
} else {
- logger_->log_info("Refreshing the peer list since there are none configured.");
+ logger_->log_debug("Refreshing the peer list since there are none configured.");
refreshPeerList();
}
}
}
- logger_->log_info("Obtained protocol from available_protocols_");
+ logger_->log_debug("Obtained protocol from available_protocols_");
return nextProtocol;
}
@@ -95,11 +95,11 @@ void RemoteProcessorGroupPort::returnProtocol(std::unique_ptr<sitetosite::SiteTo
if (max_concurrent_tasks_ > count)
count = max_concurrent_tasks_;
if (available_protocols_.size_approx() >= count) {
- logger_->log_info("not enqueueing protocol %s", getUUIDStr());
+ logger_->log_debug("not enqueueing protocol %s", getUUIDStr());
// let the memory be freed
return;
}
- logger_->log_info("enqueueing protocol %s, have a total of %lu", getUUIDStr(), available_protocols_.size_approx());
+ logger_->log_debug("enqueueing protocol %s, have a total of %lu", getUUIDStr(), available_protocols_.size_approx());
available_protocols_.enqueue(std::move(return_protocol));
}
@@ -142,13 +142,13 @@ void RemoteProcessorGroupPort::initialize() {
if (peer_index_ >= static_cast<int>(peers_.size())) {
peer_index_ = 0;
}
- logger_->log_info("Creating client");
+ logger_->log_trace("Creating client");
nextProtocol = sitetosite::createClient(config);
- logger_->log_info("Created client, moving into available protocols");
+ logger_->log_trace("Created client, moving into available protocols");
returnProtocol(std::move(nextProtocol));
}
}
- logger_->log_info("Finished initialization");
+ logger_->log_trace("Finished initialization");
}
void RemoteProcessorGroupPort::onSchedule(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSessionFactory> &sessionFactory) {
@@ -168,14 +168,14 @@ void RemoteProcessorGroupPort::onSchedule(const std::shared_ptr<core::ProcessCon
}
void RemoteProcessorGroupPort::onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) {
- logger_->log_info("On trigger %s", getUUIDStr());
+ logger_->log_trace("On trigger %s", getUUIDStr());
if (!transmitting_) {
return;
}
std::string value;
- logger_->log_info("On trigger %s", getUUIDStr());
+ logger_->log_trace("On trigger %s", getUUIDStr());
if (url_.empty()) {
if (context->getProperty(hostName.getName(), value) && !value.empty()) {
host_ = value;
@@ -193,7 +193,7 @@ void RemoteProcessorGroupPort::onTrigger(const std::shared_ptr<core::ProcessCont
std::unique_ptr<sitetosite::SiteToSiteClient> protocol_ = nullptr;
try {
- logger_->log_info("get protocol in on trigger");
+ logger_->log_trace("get protocol in on trigger");
protocol_ = getNextProtocol();
if (!protocol_) {
@@ -203,7 +203,7 @@ void RemoteProcessorGroupPort::onTrigger(const std::shared_ptr<core::ProcessCont
}
if (!protocol_->transfer(direction_, context, session)) {
- logger_->log_info("protocol transmission failed, yielding");
+ logger_->log_warn("protocol transmission failed, yielding");
context->yield();
}
@@ -264,7 +264,7 @@ void RemoteProcessorGroupPort::refreshRemoteSite2SiteInfo() {
const std::vector<char> &response_body = client->getResponseBody();
if (!response_body.empty()) {
std::string controller = std::string(response_body.begin(), response_body.end());
- logger_->log_debug("controller config %s", controller.c_str());
+ logger_->log_debug("controller config %s", controller);
Json::Value value;
Json::Reader reader;
bool parsingSuccessful = reader.parse(controller, value);
@@ -280,7 +280,7 @@ void RemoteProcessorGroupPort::refreshRemoteSite2SiteInfo() {
if (!secure.empty())
this->site2site_secure_ = secure.asBool();
}
- logger_->log_info("process group remote site2site port %d, is secure %d", site2site_port_, site2site_secure_);
+ logger_->log_debug("process group remote site2site port %d, is secure %d", site2site_port_, site2site_secure_);
}
} else {
logger_->log_error("Cannot output body to content for ProcessGroup::refreshRemoteSite2SiteInfo: received HTTP code %ll from %s", client->getResponseCode(), fullUrl);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/ThreadedSchedulingAgent.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/ThreadedSchedulingAgent.cpp b/libminifi/src/ThreadedSchedulingAgent.cpp
index 8864ba0..61dd3fe 100644
--- a/libminifi/src/ThreadedSchedulingAgent.cpp
+++ b/libminifi/src/ThreadedSchedulingAgent.cpp
@@ -58,12 +58,12 @@ void ThreadedSchedulingAgent::schedule(std::shared_ptr<core::Processor> processo
}
if (processor->getScheduledState() != core::RUNNING) {
- logger_->log_info("Can not schedule threads for processor %s because it is not running", processor->getName().c_str());
+ logger_->log_debug("Can not schedule threads for processor %s because it is not running", processor->getName());
return;
}
if (thread_pool_.isRunning(processor->getUUIDStr())) {
- logger_->log_info("Can not schedule threads for processor %s because there are existing threads running");
+ logger_->log_warn("Can not schedule threads for processor %s because there are existing threads running");
return;
}
@@ -93,7 +93,7 @@ void ThreadedSchedulingAgent::schedule(std::shared_ptr<core::Processor> processo
std::future<uint64_t> future;
thread_pool_.execute(std::move(functor), future);
}
- logger_->log_info("Scheduled thread %d concurrent workers for for process %s", processor->getMaxConcurrentTasks(), processor->getName().c_str());
+ logger_->log_debug("Scheduled thread %d concurrent workers for for process %s", processor->getMaxConcurrentTasks(), processor->getName());
return;
}
@@ -104,10 +104,10 @@ void ThreadedSchedulingAgent::stop() {
void ThreadedSchedulingAgent::unschedule(std::shared_ptr<core::Processor> processor) {
std::lock_guard<std::mutex> lock(mutex_);
- logger_->log_info("Shutting down threads for processor %s/%s", processor->getName().c_str(), processor->getUUIDStr().c_str());
+ logger_->log_debug("Shutting down threads for processor %s/%s", processor->getName(), processor->getUUIDStr());
if (processor->getScheduledState() != core::RUNNING) {
- logger_->log_info("Cannot unschedule threads for processor %s because it is not running", processor->getName().c_str());
+ logger_->log_warn("Cannot unschedule threads for processor %s because it is not running", processor->getName());
return;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/capi/Plan.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/capi/Plan.cpp b/libminifi/src/capi/Plan.cpp
index 53133d1..9775c80 100644
--- a/libminifi/src/capi/Plan.cpp
+++ b/libminifi/src/capi/Plan.cpp
@@ -115,7 +115,7 @@ bool linkToPrevious) {
bool ExecutionPlan::setProperty(const std::shared_ptr<core::Processor> proc, const std::string &prop, const std::string &value) {
uint32_t i = 0;
- logger_->log_info("Attempting to set property %s %s for %s", prop, value, proc->getName());
+ logger_->log_debug("Attempting to set property %s %s for %s", prop, value, proc->getName());
for (i = 0; i < processor_queue_.size(); i++) {
if (processor_queue_.at(i) == proc) {
break;
@@ -162,7 +162,7 @@ bool ExecutionPlan::runNextProcessor(std::function<void(const std::shared_ptr<co
if (verify != nullptr) {
verify(context, current_session);
} else {
- logger_->log_info("Running %s", processor->getName());
+ logger_->log_debug("Running %s", processor->getName());
processor->onTrigger(context, current_session);
}
current_session->commit();
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/controllers/SSLContextService.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/controllers/SSLContextService.cpp b/libminifi/src/controllers/SSLContextService.cpp
index e8dc520..1c3fd71 100644
--- a/libminifi/src/controllers/SSLContextService.cpp
+++ b/libminifi/src/controllers/SSLContextService.cpp
@@ -134,7 +134,7 @@ void SSLContextService::onEnable() {
certificate = test_cert;
logger_->log_debug("%s now good", certificate);
} else {
- logger_->log_debug("%s still not good", test_cert);
+ logger_->log_warn("%s still not good", test_cert);
valid_ = false;
}
cert_file_test.close();
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/core/ConfigurableComponent.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/core/ConfigurableComponent.cpp b/libminifi/src/core/ConfigurableComponent.cpp
index 0273aa2..a8a0b0d 100644
--- a/libminifi/src/core/ConfigurableComponent.cpp
+++ b/libminifi/src/core/ConfigurableComponent.cpp
@@ -88,7 +88,7 @@ bool ConfigurableComponent::setProperty(const std::string name, std::string valu
Property item = it->second;
item.setValue(value);
properties_[item.getName()] = item;
- logger_->log_debug("Component %s property name %s value %s", name.c_str(), item.getName().c_str(), value.c_str());
+ logger_->log_debug("Component %s property name %s value %s", name, item.getName(), value);
return true;
} else {
return false;
@@ -109,7 +109,7 @@ bool ConfigurableComponent::updateProperty(const std::string &name, const std::s
Property item = it->second;
item.addValue(value);
properties_[item.getName()] = item;
- logger_->log_debug("Component %s property name %s value %s", name.c_str(), item.getName().c_str(), value.c_str());
+ logger_->log_debug("Component %s property name %s value %s", name, item.getName(), value);
return true;
} else {
return false;
@@ -130,7 +130,7 @@ bool ConfigurableComponent::setProperty(Property &prop, std::string value) {
Property item = it->second;
item.setValue(value);
properties_[item.getName()] = item;
- logger_->log_debug("property name %s value %s", prop.getName().c_str(), item.getName().c_str(), value.c_str());
+ logger_->log_debug("property name %s value %s", prop.getName(), item.getName(), value);
return true;
} else {
Property newProp(prop);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/core/Connectable.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/core/Connectable.cpp b/libminifi/src/core/Connectable.cpp
index e2f033e..29ee411 100644
--- a/libminifi/src/core/Connectable.cpp
+++ b/libminifi/src/core/Connectable.cpp
@@ -49,7 +49,7 @@ Connectable::~Connectable() {
bool Connectable::setSupportedRelationships(std::set<core::Relationship> relationships) {
if (isRunning()) {
- logger_->log_info("Can not set processor supported relationship while the process %s is running", name_.c_str());
+ logger_->log_warn("Can not set processor supported relationship while the process %s is running", name_);
return false;
}
@@ -58,7 +58,7 @@ bool Connectable::setSupportedRelationships(std::set<core::Relationship> relatio
relationships_.clear();
for (auto item : relationships) {
relationships_[item.getName()] = item;
- logger_->log_info("Processor %s supported relationship name %s", name_.c_str(), item.getName().c_str());
+ logger_->log_debug("Processor %s supported relationship name %s", name_, item.getName());
}
return true;
}
@@ -81,7 +81,7 @@ bool Connectable::isSupportedRelationship(core::Relationship relationship) {
bool Connectable::setAutoTerminatedRelationships(std::set<Relationship> relationships) {
if (isRunning()) {
- logger_->log_info("Can not set processor auto terminated relationship while the process %s is running", name_.c_str());
+ logger_->log_warn("Can not set processor auto terminated relationship while the process %s is running", name_);
return false;
}
@@ -90,7 +90,7 @@ bool Connectable::setAutoTerminatedRelationships(std::set<Relationship> relation
auto_terminated_relationships_.clear();
for (auto item : relationships) {
auto_terminated_relationships_[item.getName()] = item;
- logger_->log_info("Processor %s auto terminated relationship name %s", name_.c_str(), item.getName().c_str());
+ logger_->log_debug("Processor %s auto terminated relationship name %s", name_, item.getName());
}
return true;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/core/FlowConfiguration.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/core/FlowConfiguration.cpp b/libminifi/src/core/FlowConfiguration.cpp
index 22d737a..b082dce 100644
--- a/libminifi/src/core/FlowConfiguration.cpp
+++ b/libminifi/src/core/FlowConfiguration.cpp
@@ -37,7 +37,7 @@ FlowConfiguration::~FlowConfiguration() {
std::shared_ptr<core::Processor> FlowConfiguration::createProcessor(std::string name, uuid_t uuid) {
auto ptr = core::ClassLoader::getDefaultClassLoader().instantiate(name, uuid);
if (nullptr == ptr) {
- logger_->log_error("No Processor defined for %s", name.c_str());
+ logger_->log_error("No Processor defined for %s", name);
return nullptr;
}
std::shared_ptr<core::Processor> processor = std::static_pointer_cast<core::Processor>(ptr);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/core/ProcessGroup.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/core/ProcessGroup.cpp b/libminifi/src/core/ProcessGroup.cpp
index 5915089..023ca9d 100644
--- a/libminifi/src/core/ProcessGroup.cpp
+++ b/libminifi/src/core/ProcessGroup.cpp
@@ -54,7 +54,7 @@ ProcessGroup::ProcessGroup(ProcessGroupType type, std::string name, uuid_t uuid,
yield_period_msec_ = 0;
transmitting_ = false;
- logger_->log_info("ProcessGroup %s created", name_);
+ logger_->log_debug("ProcessGroup %s created", name_);
}
ProcessGroup::~ProcessGroup() {
@@ -79,7 +79,7 @@ void ProcessGroup::addProcessor(std::shared_ptr<Processor> processor) {
if (processors_.find(processor) == processors_.end()) {
// We do not have the same processor in this process group yet
processors_.insert(processor);
- logger_->log_info("Add processor %s into process group %s", processor->getName().c_str(), name_.c_str());
+ logger_->log_debug("Add processor %s into process group %s", processor->getName(), name_);
}
}
@@ -89,7 +89,7 @@ void ProcessGroup::removeProcessor(std::shared_ptr<Processor> processor) {
if (processors_.find(processor) != processors_.end()) {
// We do have the same processor in this process group yet
processors_.erase(processor);
- logger_->log_info("Remove processor %s from process group %s", processor->getName().c_str(), name_.c_str());
+ logger_->log_debug("Remove processor %s from process group %s", processor->getName(), name_);
}
}
@@ -99,7 +99,7 @@ void ProcessGroup::addProcessGroup(ProcessGroup *child) {
if (child_process_groups_.find(child) == child_process_groups_.end()) {
// We do not have the same child process group in this process group yet
child_process_groups_.insert(child);
- logger_->log_info("Add child process group %s into process group %s", child->getName().c_str(), name_.c_str());
+ logger_->log_debug("Add child process group %s into process group %s", child->getName(), name_);
}
}
@@ -109,7 +109,7 @@ void ProcessGroup::removeProcessGroup(ProcessGroup *child) {
if (child_process_groups_.find(child) != child_process_groups_.end()) {
// We do have the same child process group in this process group yet
child_process_groups_.erase(child);
- logger_->log_info("Remove child process group %s from process group %s", child->getName().c_str(), name_.c_str());
+ logger_->log_debug("Remove child process group %s from process group %s", child->getName(), name_);
}
}
@@ -119,7 +119,7 @@ void ProcessGroup::startProcessing(TimerDrivenSchedulingAgent *timeScheduler, Ev
try {
// Start all the processor node, input and output ports
for (auto processor : processors_) {
- logger_->log_debug("Starting %s", processor->getName().c_str());
+ logger_->log_debug("Starting %s", processor->getName());
if (!processor->isRunning() && processor->getScheduledState() != DISABLED) {
if (processor->getSchedulingStrategy() == TIMER_DRIVEN)
@@ -171,7 +171,7 @@ std::shared_ptr<Processor> ProcessGroup::findProcessor(uuid_t uuid) {
std::lock_guard<std::recursive_mutex> lock(mutex_);
std::shared_ptr<Processor> ret = NULL;
for (auto processor : processors_) {
- logger_->log_info("find processor %s", processor->getName().c_str());
+ logger_->log_debug("find processor %s", processor->getName());
uuid_t processorUUID;
if (processor->getUUID(processorUUID)) {
@@ -186,7 +186,7 @@ std::shared_ptr<Processor> ProcessGroup::findProcessor(uuid_t uuid) {
}
}
for (auto processGroup : child_process_groups_) {
- logger_->log_info("find processor child %s", processGroup->getName().c_str());
+ logger_->log_debug("find processor child %s", processGroup->getName());
std::shared_ptr<Processor> processor = processGroup->findProcessor(uuid);
if (processor)
return processor;
@@ -212,7 +212,7 @@ void ProcessGroup::getAllProcessors(std::vector<std::shared_ptr<Processor>> &pro
std::shared_ptr<Processor> ret = NULL;
for (auto processor : processors_) {
- logger_->log_debug("Current processor is %s", processor->getName().c_str());
+ logger_->log_debug("Current processor is %s", processor->getName());
processor_vec.push_back(processor);
}
for (auto processGroup : child_process_groups_) {
@@ -224,7 +224,7 @@ std::shared_ptr<Processor> ProcessGroup::findProcessor(const std::string &proces
std::lock_guard<std::recursive_mutex> lock(mutex_);
std::shared_ptr<Processor> ret = NULL;
for (auto processor : processors_) {
- logger_->log_debug("Current processor is %s", processor->getName().c_str());
+ logger_->log_debug("Current processor is %s", processor->getName());
if (processor->getName() == processorName)
return processor;
}
@@ -275,7 +275,7 @@ void ProcessGroup::addConnection(std::shared_ptr<Connection> connection) {
if (connections_.find(connection) == connections_.end()) {
// We do not have the same connection in this process group yet
connections_.insert(connection);
- logger_->log_info("Add connection %s into process group %s", connection->getName().c_str(), name_.c_str());
+ logger_->log_debug("Add connection %s into process group %s", connection->getName(), name_);
uuid_t sourceUUID;
std::shared_ptr<Processor> source = NULL;
connection->getSourceUUID(sourceUUID);
@@ -297,7 +297,7 @@ void ProcessGroup::removeConnection(std::shared_ptr<Connection> connection) {
if (connections_.find(connection) != connections_.end()) {
// We do not have the same connection in this process group yet
connections_.erase(connection);
- logger_->log_info("Remove connection %s into process group %s", connection->getName().c_str(), name_.c_str());
+ logger_->log_debug("Remove connection %s into process group %s", connection->getName(), name_);
uuid_t sourceUUID;
std::shared_ptr<Processor> source = NULL;
connection->getSourceUUID(sourceUUID);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/core/ProcessSession.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/core/ProcessSession.cpp b/libminifi/src/core/ProcessSession.cpp
index bdfddc5..bd98db6 100644
--- a/libminifi/src/core/ProcessSession.cpp
+++ b/libminifi/src/core/ProcessSession.cpp
@@ -46,7 +46,7 @@ std::shared_ptr<core::FlowFile> ProcessSession::create() {
std::shared_ptr<core::FlowFile> record = std::make_shared<FlowFileRecord>(process_context_->getFlowFileRepository(), process_context_->getContentRepository(), empty);
_addedFlowFiles[record->getUUIDStr()] = record;
- logger_->log_debug("Create FlowFile with UUID %s", record->getUUIDStr().c_str());
+ logger_->log_debug("Create FlowFile with UUID %s", record->getUUIDStr());
std::stringstream details;
details << process_context_->getProcessorNode()->getName() << " creates flow record " << record->getUUIDStr();
provenance_report_->create(record, details.str());
@@ -64,7 +64,7 @@ std::shared_ptr<core::FlowFile> ProcessSession::create(const std::shared_ptr<cor
if (record) {
_addedFlowFiles[record->getUUIDStr()] = record;
- logger_->log_debug("Create FlowFile with UUID %s", record->getUUIDStr().c_str());
+ logger_->log_debug("Create FlowFile with UUID %s", record->getUUIDStr());
}
if (record) {
@@ -106,7 +106,7 @@ std::shared_ptr<core::FlowFile> ProcessSession::cloneDuringTransfer(std::shared_
if (record) {
this->_clonedFlowFiles[record->getUUIDStr()] = record;
- logger_->log_debug("Clone FlowFile with UUID %s during transfer", record->getUUIDStr().c_str());
+ logger_->log_debug("Clone FlowFile with UUID %s during transfer", record->getUUIDStr());
// Copy attributes
std::map<std::string, std::string> parentAttributes = parent->getAttributes();
std::map<std::string, std::string>::iterator it;
@@ -191,10 +191,13 @@ void ProcessSession::removeAttribute(const std::shared_ptr<core::FlowFile> &flow
}
void ProcessSession::penalize(const std::shared_ptr<core::FlowFile> &flow) {
- flow->setPenaltyExpiration(getTimeMillis() + process_context_->getProcessorNode()->getPenalizationPeriodMsec());
+ uint64_t penalization_period = process_context_->getProcessorNode()->getPenalizationPeriodMsec();
+ logging::LOG_INFO(logger_) << "Penalizing " << flow->getUUIDStr() << " for " << penalization_period << "ms at " << process_context_->getProcessorNode()->getName();
+ flow->setPenaltyExpiration(getTimeMillis() + penalization_period);
}
void ProcessSession::transfer(const std::shared_ptr<core::FlowFile> &flow, Relationship relationship) {
+ logging::LOG_INFO(logger_) << "Transferring " << flow->getUUIDStr() << " from " << process_context_->getProcessorNode()->getName() << " to relationship " << relationship.getName();
_transferRelationship[flow->getUUIDStr()] = relationship;
}
@@ -371,8 +374,8 @@ void ProcessSession::importFrom(io::DataStream &stream, const std::shared_ptr<co
}
flow->setResourceClaim(claim);
- logger_->log_debug("Import offset %llu length %llu into content %s for FlowFile UUID %s", flow->getOffset(), flow->getSize(), flow->getResourceClaim()->getContentFullPath().c_str(),
- flow->getUUIDStr().c_str());
+ logger_->log_debug("Import offset %llu length %llu into content %s for FlowFile UUID %s", flow->getOffset(), flow->getSize(), flow->getResourceClaim()->getContentFullPath(),
+ flow->getUUIDStr());
content_stream->closeStream();
std::stringstream details;
@@ -443,8 +446,8 @@ void ProcessSession::import(std::string source, const std::shared_ptr<core::Flow
}
flow->setResourceClaim(claim);
- logger_->log_debug("Import offset %llu length %llu into content %s for FlowFile UUID %s", flow->getOffset(), flow->getSize(), flow->getResourceClaim()->getContentFullPath().c_str(),
- flow->getUUIDStr().c_str());
+ logger_->log_debug("Import offset %llu length %llu into content %s for FlowFile UUID %s", flow->getOffset(), flow->getSize(), flow->getResourceClaim()->getContentFullPath(),
+ flow->getUUIDStr());
stream->closeStream();
input.close();
@@ -531,7 +534,7 @@ void ProcessSession::import(std::string source, std::vector<std::shared_ptr<Flow
flowFile->setResourceClaim(claim);
claim->increaseFlowFileRecordOwnedCount();
logger_->log_debug("Import offset %llu length %llu into content %s for FlowFile UUID %s", flowFile->getOffset(), flowFile->getSize(),
- flowFile->getResourceClaim()->getContentFullPath().c_str(), flowFile->getUUIDStr().c_str());
+ flowFile->getResourceClaim()->getContentFullPath(), flowFile->getUUIDStr());
stream->closeStream();
std::string details = process_context_->getProcessorNode()->getName() + " modify flow record content " + flowFile->getUUIDStr();
uint64_t endTime = getTimeMillis();
@@ -568,18 +571,18 @@ void ProcessSession::import(std::string source, std::vector<std::shared_ptr<Flow
}
bool ProcessSession::exportContent(const std::string &destination, const std::string &tmpFile, const std::shared_ptr<core::FlowFile> &flow, bool keepContent) {
- logger_->log_info("Exporting content of %s to %s", flow->getUUIDStr().c_str(), destination.c_str());
+ logger_->log_debug("Exporting content of %s to %s", flow->getUUIDStr(), destination);
ProcessSessionReadCallback cb(tmpFile, destination, logger_);
read(flow, &cb);
- logger_->log_info("Committing %s", destination.c_str());
+ logger_->log_info("Committing %s", destination);
bool commit_ok = cb.commit();
if (commit_ok) {
logger_->log_info("Commit OK.");
} else {
- logger_->log_error("Commit of %s to %s failed!", flow->getUUIDStr().c_str(), destination.c_str());
+ logger_->log_error("Commit of %s to %s failed!", flow->getUUIDStr(), destination);
}
return commit_ok;
}
@@ -597,12 +600,12 @@ bool ProcessSession::exportContent(const std::string &destination, const std::sh
}
void ProcessSession::stash(const std::string &key, const std::shared_ptr<core::FlowFile> &flow) {
- logger_->log_info("Stashing content from %s to key %s", flow->getUUIDStr().c_str(), key.c_str());
+ logger_->log_debug("Stashing content from %s to key %s", flow->getUUIDStr(), key);
if (!flow->getResourceClaim()) {
logger_->log_warn("Attempted to stash content of record %s when "
"there is no resource claim",
- flow->getUUIDStr().c_str());
+ flow->getUUIDStr());
return;
}
@@ -615,11 +618,11 @@ void ProcessSession::stash(const std::string &key, const std::shared_ptr<core::F
}
void ProcessSession::restore(const std::string &key, const std::shared_ptr<core::FlowFile> &flow) {
- logger_->log_info("Restoring content to %s from key %s", flow->getUUIDStr().c_str(), key.c_str());
+ logger_->log_info("Restoring content to %s from key %s", flow->getUUIDStr(), key);
// Restore the claim
if (!flow->hasStashClaim(key)) {
- logger_->log_warn("Requested restore to record %s from unknown key %s", flow->getUUIDStr().c_str(), key.c_str());
+ logger_->log_warn("Requested restore to record %s from unknown key %s", flow->getUUIDStr(), key);
return;
}
@@ -627,7 +630,7 @@ void ProcessSession::restore(const std::string &key, const std::shared_ptr<core:
if (flow->getResourceClaim()) {
logger_->log_warn("Restoring stashed content of record %s from key %s when there is "
"existing content; existing content will be overwritten",
- flow->getUUIDStr().c_str(), key.c_str());
+ flow->getUUIDStr(), key);
flow->releaseClaim(flow->getResourceClaim());
}
@@ -768,7 +771,7 @@ void ProcessSession::commit() {
_originalFlowFiles.clear();
// persistent the provenance report
this->provenance_report_->commit();
- logger_->log_trace("ProcessSession committed for %s", process_context_->getProcessorNode()->getName().c_str());
+ logger_->log_trace("ProcessSession committed for %s", process_context_->getProcessorNode()->getName());
} catch (std::exception &exception) {
logger_->log_debug("Caught Exception %s", exception.what());
throw;
@@ -797,7 +800,7 @@ void ProcessSession::rollback() {
_addedFlowFiles.clear();
_updatedFlowFiles.clear();
_deletedFlowFiles.clear();
- logger_->log_debug("ProcessSession rollback for %s", process_context_->getProcessorNode()->getName().c_str());
+ logger_->log_debug("ProcessSession rollback for %s", process_context_->getProcessorNode()->getName());
} catch (std::exception &exception) {
logger_->log_debug("Caught Exception %s", exception.what());
throw;
@@ -835,7 +838,7 @@ std::shared_ptr<core::FlowFile> ProcessSession::get() {
_updatedFlowFiles[ret->getUUIDStr()] = ret;
std::map<std::string, std::string> empty;
std::shared_ptr<core::FlowFile> snapshot = std::make_shared<FlowFileRecord>(process_context_->getFlowFileRepository(), process_context_->getContentRepository(), empty);
- logger_->log_debug("Create Snapshot FlowFile with UUID %s", snapshot->getUUIDStr().c_str());
+ logger_->log_debug("Create Snapshot FlowFile with UUID %s", snapshot->getUUIDStr());
snapshot = ret;
// save a snapshot
_originalFlowFiles[snapshot->getUUIDStr()] = snapshot;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/core/ProcessSessionReadCallback.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/core/ProcessSessionReadCallback.cpp b/libminifi/src/core/ProcessSessionReadCallback.cpp
index bc72e9f..65fd26a 100644
--- a/libminifi/src/core/ProcessSessionReadCallback.cpp
+++ b/libminifi/src/core/ProcessSessionReadCallback.cpp
@@ -64,19 +64,19 @@ int64_t ProcessSessionReadCallback::process(std::shared_ptr<io::BaseStream> stre
bool ProcessSessionReadCallback::commit() {
bool success = false;
- logger_->log_info("committing export operation to %s", _destFile.c_str());
+ logger_->log_debug("committing export operation to %s", _destFile);
if (_writeSucceeded) {
_tmpFileOs.close();
if (rename(_tmpFile.c_str(), _destFile.c_str())) {
- logger_->log_info("commit export operation to %s failed because rename() call failed", _destFile.c_str());
+ logger_->log_warn("commit export operation to %s failed because rename() call failed", _destFile);
} else {
success = true;
- logger_->log_info("commit export operation to %s succeeded", _destFile.c_str());
+ logger_->log_debug("commit export operation to %s succeeded", _destFile);
}
} else {
- logger_->log_error("commit export operation to %s failed because write failed", _destFile.c_str());
+ logger_->log_error("commit export operation to %s failed because write failed", _destFile);
}
return success;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/core/Processor.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/core/Processor.cpp b/libminifi/src/core/Processor.cpp
index 2bceac5..7adf718 100644
--- a/libminifi/src/core/Processor.cpp
+++ b/libminifi/src/core/Processor.cpp
@@ -62,7 +62,7 @@ Processor::Processor(std::string name, uuid_t uuid)
active_tasks_ = 0;
yield_expiration_ = 0;
incoming_connections_Iter = this->_incomingConnections.begin();
- logger_->log_info("Processor %s created UUID %s", name_, uuidStr_);
+ logger_->log_debug("Processor %s created UUID %s", name_, uuidStr_);
}
bool Processor::isRunning() {
@@ -80,7 +80,7 @@ bool Processor::addConnection(std::shared_ptr<Connectable> conn) {
bool ret = false;
if (isRunning()) {
- logger_->log_info("Can not add connection while the process %s is running", name_.c_str());
+ logger_->log_warn("Can not add connection while the process %s is running", name_);
return false;
}
std::shared_ptr<Connection> connection = std::static_pointer_cast<Connection>(conn);
@@ -102,7 +102,7 @@ bool Processor::addConnection(std::shared_ptr<Connectable> conn) {
if (_incomingConnections.find(connection) == _incomingConnections.end()) {
_incomingConnections.insert(connection);
connection->setDestination(shared_from_this());
- logger_->log_info("Add connection %s into Processor %s incoming connection", connection->getName().c_str(), name_.c_str());
+ logger_->log_debug("Add connection %s into Processor %s incoming connection", connection->getName(), name_);
incoming_connections_Iter = this->_incomingConnections.begin();
ret = true;
}
@@ -121,7 +121,7 @@ bool Processor::addConnection(std::shared_ptr<Connectable> conn) {
existedConnection.insert(connection);
connection->setSource(shared_from_this());
out_going_connections_[relationship] = existedConnection;
- logger_->log_info("Add connection %s into Processor %s outgoing connection for relationship %s", connection->getName().c_str(), name_.c_str(), relationship.c_str());
+ logger_->log_debug("Add connection %s into Processor %s outgoing connection for relationship %s", connection->getName(), name_, relationship);
ret = true;
}
} else {
@@ -130,7 +130,7 @@ bool Processor::addConnection(std::shared_ptr<Connectable> conn) {
newConnection.insert(connection);
connection->setSource(shared_from_this());
out_going_connections_[relationship] = newConnection;
- logger_->log_info("Add connection %s into Processor %s outgoing connection for relationship %s", connection->getName().c_str(), name_.c_str(), relationship.c_str());
+ logger_->log_debug("Add connection %s into Processor %s outgoing connection for relationship %s", connection->getName(), name_, relationship);
ret = true;
}
}
@@ -140,7 +140,7 @@ bool Processor::addConnection(std::shared_ptr<Connectable> conn) {
void Processor::removeConnection(std::shared_ptr<Connectable> conn) {
if (isRunning()) {
- logger_->log_info("Can not remove connection while the process %s is running", name_.c_str());
+ logger_->log_warn("Can not remove connection while the process %s is running", name_);
return;
}
@@ -159,7 +159,7 @@ void Processor::removeConnection(std::shared_ptr<Connectable> conn) {
if (_incomingConnections.find(connection) != _incomingConnections.end()) {
_incomingConnections.erase(connection);
connection->setDestination(NULL);
- logger_->log_info("Remove connection %s into Processor %s incoming connection", connection->getName().c_str(), name_.c_str());
+ logger_->log_debug("Remove connection %s into Processor %s incoming connection", connection->getName(), name_);
incoming_connections_Iter = this->_incomingConnections.begin();
}
}
@@ -174,7 +174,7 @@ void Processor::removeConnection(std::shared_ptr<Connectable> conn) {
if (out_going_connections_[relationship].find(connection) != out_going_connections_[relationship].end()) {
out_going_connections_[relationship].erase(connection);
connection->setSource(NULL);
- logger_->log_info("Remove connection %s into Processor %s outgoing connection for relationship %s", connection->getName().c_str(), name_.c_str(), relationship.c_str());
+ logger_->log_debug("Remove connection %s into Processor %s outgoing connection for relationship %s", connection->getName(), name_, relationship);
}
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/core/Repository.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/core/Repository.cpp b/libminifi/src/core/Repository.cpp
index 575f694..1432d9e 100644
--- a/libminifi/src/core/Repository.cpp
+++ b/libminifi/src/core/Repository.cpp
@@ -40,7 +40,7 @@ void Repository::start() {
return;
running_ = true;
thread_ = std::thread(&Repository::threadExecutor, this);
- logger_->log_info("%s Repository Monitor Thread Start", name_.c_str());
+ logger_->log_debug("%s Repository Monitor Thread Start", name_);
}
void Repository::stop() {
@@ -49,7 +49,7 @@ void Repository::stop() {
running_ = false;
if (thread_.joinable())
thread_.join();
- logger_->log_info("%s Repository Monitor Thread Stop", name_.c_str());
+ logger_->log_debug("%s Repository Monitor Thread Stop", name_);
}
// repoSize
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/core/repository/VolatileContentRepository.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/core/repository/VolatileContentRepository.cpp b/libminifi/src/core/repository/VolatileContentRepository.cpp
index 7c9aad9..60f538d 100644
--- a/libminifi/src/core/repository/VolatileContentRepository.cpp
+++ b/libminifi/src/core/repository/VolatileContentRepository.cpp
@@ -83,7 +83,7 @@ void VolatileContentRepository::start() {
thread_ = std::thread(&VolatileContentRepository::run, shared_from_parent<VolatileContentRepository>());
thread_.detach();
running_ = true;
- logger_->log_info("%s Repository Monitor Thread Start", getName());
+ logger_->log_debug("%s Repository Monitor Thread Start", getName());
}
std::shared_ptr<io::BaseStream> VolatileContentRepository::write(const std::shared_ptr<minifi::ResourceClaim> &claim) {
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/core/yaml/YamlConfiguration.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/core/yaml/YamlConfiguration.cpp b/libminifi/src/core/yaml/YamlConfiguration.cpp
index 5b7e2ff..8b93c51 100644
--- a/libminifi/src/core/yaml/YamlConfiguration.cpp
+++ b/libminifi/src/core/yaml/YamlConfiguration.cpp
@@ -309,8 +309,6 @@ void YamlConfiguration::parseRemoteProcessGroupYaml(YAML::Node *rpgNode, core::P
YAML::Node inputPorts = currRpgNode["Input Ports"].as<YAML::Node>();
if (inputPorts && inputPorts.IsSequence()) {
for (YAML::const_iterator portIter = inputPorts.begin(); portIter != inputPorts.end(); ++portIter) {
- logger_->log_debug("Got a current port, iterating...");
-
YAML::Node currPort = portIter->as<YAML::Node>();
this->parsePortYaml(&currPort, group, sitetosite::SEND);
@@ -688,7 +686,7 @@ void YamlConfiguration::parsePropertiesNodeYaml(YAML::Node *propertiesNode,
YAML::Node propertiesNode = nodeVal["value"];
// must insert the sequence in differently.
std::string rawValueString = propertiesNode.as<std::string>();
- logger_->log_info("Found %s=%s", propertyName, rawValueString);
+ logger_->log_debug("Found %s=%s", propertyName, rawValueString);
if (!processor->updateProperty(propertyName, rawValueString)) {
std::shared_ptr<core::Connectable> proc = std::dynamic_pointer_cast<core::Connectable>(processor);
if (proc != 0) {
@@ -757,7 +755,8 @@ void YamlConfiguration::checkRequiredField(YAML::Node *yamlNode,
errMsg += " [in '" + yamlSection + "' section of configuration file]";
}
}
- logger_->log_error(errMsg.c_str());
+ logging::LOG_ERROR(logger_) << errMsg;
+
throw std::invalid_argument(errMsg);
}
}
@@ -783,7 +782,7 @@ YAML::Node YamlConfiguration::getOptionalField(YAML::Node *yamlNode,
infoMessage += defaultValue.as<std::string>();
}
- logger_->log_info(infoMessage.c_str());
+ logging::LOG_ERROR(logger_) << infoMessage;
result = defaultValue;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/io/FileStream.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/io/FileStream.cpp b/libminifi/src/io/FileStream.cpp
index e903d95..78916d6 100644
--- a/libminifi/src/io/FileStream.cpp
+++ b/libminifi/src/io/FileStream.cpp
@@ -142,9 +142,7 @@ int FileStream::readData(uint8_t *buf, int buflen) {
size_t ret = len - offset_;
offset_ = len;
length_ = len;
- std::stringstream str;
- str << path_ << " eof bit, ended at " << offset_;
- logger_->log_info(str.str().c_str());
+ logging::LOG_DEBUG(logger_) << path_ << " eof bit, ended at " << offset_;
return ret;
} else {
offset_ += buflen;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/io/tls/TLSSocket.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/io/tls/TLSSocket.cpp b/libminifi/src/io/tls/TLSSocket.cpp
index 40cf1e9..d51fa76 100644
--- a/libminifi/src/io/tls/TLSSocket.cpp
+++ b/libminifi/src/io/tls/TLSSocket.cpp
@@ -94,7 +94,7 @@ int16_t TLSContext::initialize() {
int retp = SSL_CTX_use_PrivateKey_file(ctx, privatekey.c_str(), SSL_FILETYPE_PEM);
if (retp != 1) {
- logger_->log_error("Could not create load private key,%i on %s error : %s", retp, privatekey.c_str(), std::strerror(errno));
+ logger_->log_error("Could not create load private key,%i on %s error : %s", retp, privatekey, std::strerror(errno));
error_value = TLS_ERROR_KEY_ERROR;
return error_value;
}
@@ -114,7 +114,7 @@ int16_t TLSContext::initialize() {
}
}
- logger_->log_info("Load/Verify Client Certificate OK.");
+ logger_->log_debug("Load/Verify Client Certificate OK.");
}
return 0;
}
@@ -159,13 +159,13 @@ int16_t TLSSocket::initialize() {
ssl = SSL_new(context_->getContext());
SSL_set_fd(ssl, socket_file_descriptor_);
if (SSL_connect(ssl) == -1) {
- logger_->log_error("SSL socket connect failed to %s %d", requested_hostname_.c_str(), port_);
+ logger_->log_error("SSL socket connect failed to %s %d", requested_hostname_, port_);
SSL_free(ssl);
ssl = NULL;
close(socket_file_descriptor_);
return -1;
} else {
- logger_->log_info("SSL socket connect success to %s %d", requested_hostname_.c_str(), port_);
+ logger_->log_debug("SSL socket connect success to %s %d", requested_hostname_, port_);
return 0;
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/processors/ExecuteProcess.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/processors/ExecuteProcess.cpp b/libminifi/src/processors/ExecuteProcess.cpp
index 13d52c8..f88f5c7 100644
--- a/libminifi/src/processors/ExecuteProcess.cpp
+++ b/libminifi/src/processors/ExecuteProcess.cpp
@@ -85,7 +85,7 @@ void ExecuteProcess::onTrigger(core::ProcessContext *context, core::ProcessSessi
if (context->getProperty(BatchDuration.getName(), value)) {
core::TimeUnit unit;
if (core::Property::StringToTime(value, _batchDuration, unit) && core::Property::ConvertTimeUnitToMS(_batchDuration, unit, _batchDuration)) {
- logger_->log_info("Setting _batchDuration");
+ logger_->log_debug("Setting _batchDuration");
}
}
if (context->getProperty(RedirectErrorStream.getName(), value)) {
@@ -99,12 +99,12 @@ void ExecuteProcess::onTrigger(core::ProcessContext *context, core::ProcessSessi
if (_workingDir.length() > 0 && _workingDir != ".") {
// change to working directory
if (chdir(_workingDir.c_str()) != 0) {
- logger_->log_error("Execute Command can not chdir %s", _workingDir.c_str());
+ logger_->log_error("Execute Command can not chdir %s", _workingDir);
yield();
return;
}
}
- logger_->log_info("Execute Command %s", _fullCommand.c_str());
+ logger_->log_info("Execute Command %s", _fullCommand);
// split the command into array
char *p = std::strtok(const_cast<char*>(_fullCommand.c_str()), " ");
int argc = 0;
@@ -153,13 +153,13 @@ void ExecuteProcess::onTrigger(core::ProcessContext *context, core::ProcessSessi
int numRead = read(_pipefd[0], buffer, sizeof(buffer));
if (numRead <= 0)
break;
- logger_->log_info("Execute Command Respond %d", numRead);
+ logger_->log_debug("Execute Command Respond %d", numRead);
ExecuteProcess::WriteCallback callback(buffer, numRead);
std::shared_ptr<FlowFileRecord> flowFile = std::static_pointer_cast<FlowFileRecord>(session->create());
if (!flowFile)
continue;
- flowFile->addAttribute("command", _command.c_str());
- flowFile->addAttribute("command.arguments", _commandArgument.c_str());
+ flowFile->addAttribute("command", _command);
+ flowFile->addAttribute("command.arguments", _commandArgument);
session->write(flowFile, &callback);
session->transfer(flowFile, Success);
session->commit();
@@ -173,15 +173,15 @@ void ExecuteProcess::onTrigger(core::ProcessContext *context, core::ProcessSessi
int numRead = read(_pipefd[0], bufPtr, (sizeof(buffer) - totalRead));
if (numRead <= 0) {
if (totalRead > 0) {
- logger_->log_info("Execute Command Respond %d", totalRead);
+ logger_->log_debug("Execute Command Respond %d", totalRead);
// child exits and close the pipe
ExecuteProcess::WriteCallback callback(buffer, totalRead);
if (!flowFile) {
flowFile = std::static_pointer_cast<FlowFileRecord>(session->create());
if (!flowFile)
break;
- flowFile->addAttribute("command", _command.c_str());
- flowFile->addAttribute("command.arguments", _commandArgument.c_str());
+ flowFile->addAttribute("command", _command);
+ flowFile->addAttribute("command.arguments", _commandArgument);
session->write(flowFile, &callback);
} else {
session->append(flowFile, &callback);
@@ -192,14 +192,14 @@ void ExecuteProcess::onTrigger(core::ProcessContext *context, core::ProcessSessi
} else {
if (numRead == static_cast<int>((sizeof(buffer) - totalRead))) {
// we reach the max buffer size
- logger_->log_info("Execute Command Max Respond %d", sizeof(buffer));
+ logger_->log_debug("Execute Command Max Respond %d", sizeof(buffer));
ExecuteProcess::WriteCallback callback(buffer, sizeof(buffer));
if (!flowFile) {
flowFile = std::static_pointer_cast<FlowFileRecord>(session->create());
if (!flowFile)
continue;
- flowFile->addAttribute("command", _command.c_str());
- flowFile->addAttribute("command.arguments", _commandArgument.c_str());
+ flowFile->addAttribute("command", _command);
+ flowFile->addAttribute("command.arguments", _commandArgument);
session->write(flowFile, &callback);
} else {
session->append(flowFile, &callback);
@@ -217,9 +217,9 @@ void ExecuteProcess::onTrigger(core::ProcessContext *context, core::ProcessSessi
wait(&status);
if (WIFEXITED(status)) {
- logger_->log_info("Execute Command Complete %s status %d pid %d", _fullCommand.c_str(), WEXITSTATUS(status), _pid);
+ logger_->log_info("Execute Command Complete %s status %d pid %d", _fullCommand, WEXITSTATUS(status), _pid);
} else {
- logger_->log_info("Execute Command Complete %s status %d pid %d", _fullCommand.c_str(), WTERMSIG(status), _pid);
+ logger_->log_info("Execute Command Complete %s status %d pid %d", _fullCommand, WTERMSIG(status), _pid);
}
close(_pipefd[0]);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/processors/GetFile.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/processors/GetFile.cpp b/libminifi/src/processors/GetFile.cpp
index 0472c55..5d80b42 100644
--- a/libminifi/src/processors/GetFile.cpp
+++ b/libminifi/src/processors/GetFile.cpp
@@ -136,14 +136,14 @@ void GetFile::onTrigger(core::ProcessContext *context, core::ProcessSession *ses
metrics_->iterations_++;
- logger_->log_info("Is listing empty %i", isListingEmpty());
+ logger_->log_debug("Is listing empty %i", isListingEmpty());
if (isListingEmpty()) {
if (request_.pollInterval == 0 || (getTimeMillis() - last_listing_time_) > request_.pollInterval) {
performListing(request_.directory, request_);
last_listing_time_.store(getTimeMillis());
}
}
- logger_->log_info("Is listing empty %i", isListingEmpty());
+ logger_->log_debug("Is listing empty %i", isListingEmpty());
if (!isListingEmpty()) {
try {
@@ -152,7 +152,7 @@ void GetFile::onTrigger(core::ProcessContext *context, core::ProcessSession *ses
while (!list.empty()) {
std::string fileName = list.front();
list.pop();
- logger_->log_info("GetFile process %s", fileName.c_str());
+ logger_->log_info("GetFile process %s", fileName);
std::shared_ptr<FlowFileRecord> flowFile = std::static_pointer_cast<FlowFileRecord>(session->create());
if (flowFile == nullptr)
return;
@@ -241,13 +241,12 @@ bool GetFile::acceptFile(std::string fullName, std::string name, const GetFileRe
}
void GetFile::performListing(std::string dir, const GetFileRequest &request) {
- logger_->log_info("Performing file listing against %s", dir.c_str());
DIR *d;
d = opendir(dir.c_str());
if (!d)
return;
// only perform a listing while we are not empty
- logger_->log_info("Performing file listing against %s", dir.c_str());
+ logger_->log_debug("Performing file listing against %s", dir);
while (isRunning()) {
struct dirent *entry;
entry = readdir(d);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/processors/GetTCP.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/processors/GetTCP.cpp b/libminifi/src/processors/GetTCP.cpp
index bfb9a3c..3bf15b2 100644
--- a/libminifi/src/processors/GetTCP.cpp
+++ b/libminifi/src/processors/GetTCP.cpp
@@ -168,12 +168,12 @@ void GetTCP::onSchedule(const std::shared_ptr<core::ProcessContext> &context, co
}
}
if (startLoc > 0) {
- logger_->log_info("Starting at %i, ending at %i", startLoc, size_read);
+ logger_->log_trace("Starting at %i, ending at %i", startLoc, size_read);
if (size_read-startLoc > 0) {
handler_->handle(socket_ptr->getHostname(), buffer.data()+startLoc, (size_read-startLoc), true);
}
} else {
- logger_->log_info("Handling at %i, ending at %i", startLoc, size_read);
+ logger_->log_trace("Handling at %i, ending at %i", startLoc, size_read);
if (size_read > 0) {
handler_->handle(socket_ptr->getHostname(), buffer.data(), size_read, false);
}
@@ -202,7 +202,7 @@ void GetTCP::onSchedule(const std::shared_ptr<core::ProcessContext> &context, co
return -1;
}
}while (running_);
- logger_->log_info("Ending private thread");
+ logger_->log_debug("Ending private thread");
return 0;
};
@@ -268,12 +268,12 @@ void GetTCP::onTrigger(const std::shared_ptr<core::ProcessContext> &context, con
live_clients_[endpoint] = future;
}
} else {
- logger_->log_info("Thread still running for %s", endPointFuture->first);
+ logger_->log_debug("Thread still running for %s", endPointFuture->first);
// we have a thread corresponding to this.
}
}
}
- logger_->log_info("Updating endpoint");
+ logger_->log_debug("Updating endpoint");
context->yield();
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/processors/ListenSyslog.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/processors/ListenSyslog.cpp b/libminifi/src/processors/ListenSyslog.cpp
index a71b69f..a3236a6 100644
--- a/libminifi/src/processors/ListenSyslog.cpp
+++ b/libminifi/src/processors/ListenSyslog.cpp
@@ -71,7 +71,7 @@ void ListenSyslog::startSocketThread() {
if (_thread != NULL)
return;
- logger_->log_info("ListenSysLog Socket Thread Start");
+ logger_->log_trace("ListenSysLog Socket Thread Start");
_serverTheadRunning = true;
_thread = new std::thread(run, this);
_thread->detach();
@@ -107,7 +107,7 @@ void ListenSyslog::runThread() {
else
sockfd = socket(AF_INET, SOCK_DGRAM, 0);
if (sockfd < 0) {
- logger_->log_info("ListenSysLog Server socket creation failed");
+ logger_->log_error("ListenSysLog Server socket creation failed");
break;
}
bzero(reinterpret_cast<char *>(&serv_addr), sizeof(serv_addr));
@@ -180,7 +180,7 @@ void ListenSyslog::runThread() {
int recvlen = readline(clientSocket, _buffer, sizeof(_buffer));
if (recvlen <= 0) {
close(clientSocket);
- logger_->log_info("ListenSysLog client socket %d close", clientSocket);
+ logger_->log_debug("ListenSysLog client socket %d close", clientSocket);
it = _clientSockets.erase(it);
} else {
if ((uint64_t)(recvlen + getEventQueueByteSize()) <= _recvBufSize) {
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/processors/LogAttribute.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/processors/LogAttribute.cpp b/libminifi/src/processors/LogAttribute.cpp
index 0c2f64e..65f45c6 100644
--- a/libminifi/src/processors/LogAttribute.cpp
+++ b/libminifi/src/processors/LogAttribute.cpp
@@ -64,7 +64,7 @@ void LogAttribute::initialize() {
}
void LogAttribute::onTrigger(core::ProcessContext *context, core::ProcessSession *session) {
- logger_->log_info("enter log attribute");
+ logger_->log_trace("enter log attribute");
std::string dashLine = "--------------------------------------------------";
LogAttrLevel level = LogAttrLevelInfo;
bool logPayload = false;
@@ -123,19 +123,19 @@ void LogAttribute::onTrigger(core::ProcessContext *context, core::ProcessSession
switch (level) {
case LogAttrLevelInfo:
- logger_->log_info("%s", output.c_str());
+ logger_->log_info("%s", output);
break;
case LogAttrLevelDebug:
- logger_->log_debug("%s", output.c_str());
+ logger_->log_debug("%s", output);
break;
case LogAttrLevelError:
- logger_->log_error("%s", output.c_str());
+ logger_->log_error("%s", output);
break;
case LogAttrLevelTrace:
- logger_->log_trace("%s", output.c_str());
+ logger_->log_trace("%s", output);
break;
case LogAttrLevelWarn:
- logger_->log_warn("%s", output.c_str());
+ logger_->log_warn("%s", output);
break;
default:
break;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/processors/PutFile.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/processors/PutFile.cpp b/libminifi/src/processors/PutFile.cpp
index 9ffbbd7..f073025 100644
--- a/libminifi/src/processors/PutFile.cpp
+++ b/libminifi/src/processors/PutFile.cpp
@@ -123,14 +123,14 @@ void PutFile::onTrigger(core::ProcessContext *context, core::ProcessSession *ses
flowFile->getKeyedAttribute(FILENAME, filename);
std::string tmpFile = tmpWritePath(filename, directory);
- logger_->log_info("PutFile using temporary file %s", tmpFile.c_str());
+ logger_->log_debug("PutFile using temporary file %s", tmpFile);
// Determine dest full file paths
std::stringstream destFileSs;
destFileSs << directory << "/" << filename;
std::string destFile = destFileSs.str();
- logger_->log_info("PutFile writing file %s into directory %s", filename.c_str(), directory.c_str());
+ logger_->log_debug("PutFile writing file %s into directory %s", filename, directory);
// If file exists, apply conflict resolution strategy
struct stat statResult;
@@ -141,7 +141,7 @@ void PutFile::onTrigger(core::ProcessContext *context, core::ProcessSession *ses
// it's a directory, count the files
DIR *myDir = opendir(directory.c_str());
if (!myDir) {
- logger_->log_warn("Could not open %s", directory.c_str());
+ logger_->log_warn("Could not open %s", directory);
session->transfer(flowFile, Failure);
return;
}
@@ -153,7 +153,7 @@ void PutFile::onTrigger(core::ProcessContext *context, core::ProcessSession *ses
ct++;
if (ct >= max_dest_files_) {
logger_->log_warn("Routing to failure because the output directory %s has at least %u files, which exceeds the "
- "configured max number of files", directory.c_str(), max_dest_files_);
+ "configured max number of files", directory, max_dest_files_);
session->transfer(flowFile, Failure);
closedir(myDir);
return;
@@ -165,9 +165,9 @@ void PutFile::onTrigger(core::ProcessContext *context, core::ProcessSession *ses
}
if (stat(destFile.c_str(), &statResult) == 0) {
- logger_->log_info("Destination file %s exists; applying Conflict Resolution Strategy: %s",
- destFile.c_str(),
- conflict_resolution_.c_str());
+ logger_->log_warn("Destination file %s exists; applying Conflict Resolution Strategy: %s",
+ destFile,
+ conflict_resolution_);
if (conflict_resolution_ == CONFLICT_RESOLUTION_STRATEGY_REPLACE) {
putFile(session, flowFile, tmpFile, destFile, directory);
@@ -215,7 +215,7 @@ bool PutFile::putFile(core::ProcessSession *session,
// Attempt to create directories in file's path
std::stringstream dir_path_stream;
- logger_->log_warn("Destination directory does not exist; will attempt to create: ", destDir);
+ logger_->log_debug("Destination directory does not exist; will attempt to create: ", destDir);
size_t i = 0;
auto pos = destFile.find('/');
@@ -225,7 +225,7 @@ bool PutFile::putFile(core::ProcessSession *session,
auto dir_path = dir_path_stream.str();
if (!dir_path_component.empty()) {
- logger_->log_info("Attempting to create directory if it does not already exist: %s", dir_path);
+ logger_->log_debug("Attempting to create directory if it does not already exist: %s", dir_path);
mkdir(dir_path.c_str(), 0700);
dir_path_stream << '/';
} else if (pos == 0) {
@@ -241,7 +241,7 @@ bool PutFile::putFile(core::ProcessSession *session,
ReadCallback cb(tmpFile, destFile);
session->read(flowFile, &cb);
- logger_->log_info("Committing %s", destFile);
+ logger_->log_debug("Committing %s", destFile);
if (cb.commit()) {
session->transfer(flowFile, Success);
return true;
@@ -296,18 +296,18 @@ int64_t PutFile::ReadCallback::process(std::shared_ptr<io::BaseStream> stream) {
bool PutFile::ReadCallback::commit() {
bool success = false;
- logger_->log_info("PutFile committing put file operation to %s", dest_file_.c_str());
+ logger_->log_info("PutFile committing put file operation to %s", dest_file_);
if (write_succeeded_) {
if (rename(tmp_file_.c_str(), dest_file_.c_str())) {
logger_->log_info("PutFile commit put file operation to %s failed because rename() call failed",
- dest_file_.c_str());
+ dest_file_);
} else {
success = true;
- logger_->log_info("PutFile commit put file operation to %s succeeded", dest_file_.c_str());
+ logger_->log_info("PutFile commit put file operation to %s succeeded", dest_file_);
}
} else {
- logger_->log_error("PutFile commit put file operation to %s failed because write failed", dest_file_.c_str());
+ logger_->log_error("PutFile commit put file operation to %s failed because write failed", dest_file_);
}
return success;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/processors/TailFile.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/processors/TailFile.cpp b/libminifi/src/processors/TailFile.cpp
index 490efae..61aa86b 100644
--- a/libminifi/src/processors/TailFile.cpp
+++ b/libminifi/src/processors/TailFile.cpp
@@ -135,7 +135,7 @@ void TailFile::parseStateFileLine(char *buf) {
void TailFile::recoverState() {
std::ifstream file(_stateFile.c_str(), std::ifstream::in);
if (!file.good()) {
- logger_->log_error("load state file failed %s", _stateFile.c_str());
+ logger_->log_error("load state file failed %s", _stateFile);
return;
}
char buf[BUFFER_SIZE];
@@ -147,7 +147,7 @@ void TailFile::recoverState() {
void TailFile::storeState() {
std::ofstream file(_stateFile.c_str());
if (!file.is_open()) {
- logger_->log_error("store state file failed %s", _stateFile.c_str());
+ logger_->log_error("store state file failed %s", _stateFile);
return;
}
file << "FILENAME=" << this->_currentTailFileName << "\n";
@@ -206,7 +206,7 @@ void TailFile::checkRollOver(const std::string &fileLocation, const std::string
++it;
if (it != matchedFiles.end()) {
TailMatchedFileItem nextItem = *it;
- logger_->log_info("TailFile File Roll Over from %s to %s", _currentTailFileName.c_str(), nextItem.fileName.c_str());
+ logger_->log_info("TailFile File Roll Over from %s to %s", _currentTailFileName, nextItem.fileName);
_currentTailFileName = nextItem.fileName;
_currentTailFilePosition = 0;
storeState();
@@ -286,7 +286,7 @@ void TailFile::onTrigger(core::ProcessContext *context, core::ProcessSession *se
}
} else {
- logger_->log_warn("Unable to stat file %s", fullPath.c_str());
+ logger_->log_warn("Unable to stat file %s", fullPath);
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/sitetosite/RawSocketProtocol.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/sitetosite/RawSocketProtocol.cpp b/libminifi/src/sitetosite/RawSocketProtocol.cpp
index df063ff..2bccb0b 100644
--- a/libminifi/src/sitetosite/RawSocketProtocol.cpp
+++ b/libminifi/src/sitetosite/RawSocketProtocol.cpp
@@ -99,7 +99,7 @@ bool RawSiteToSiteClient::establish() {
return false;
}
- logger_->log_info("Site2Site socket established");
+ logger_->log_debug("Site2Site socket established");
peer_state_ = ESTABLISHED;
return true;
@@ -112,11 +112,11 @@ bool RawSiteToSiteClient::initiateResourceNegotiation() {
return false;
}
- logger_->log_info("Negotiate protocol version with destination port %s current version %d", port_id_str_, _currentVersion);
+ logger_->log_debug("Negotiate protocol version with destination port %s current version %d", port_id_str_, _currentVersion);
int ret = peer_->writeUTF(getResourceName());
- logger_->log_info("result of writing resource name is %i", ret);
+ logger_->log_trace("result of writing resource name is %i", ret);
if (ret <= 0) {
logger_->log_debug("result of writing resource name is %i", ret);
// tearDown();
@@ -126,7 +126,7 @@ bool RawSiteToSiteClient::initiateResourceNegotiation() {
ret = peer_->write(_currentVersion);
if (ret <= 0) {
- logger_->log_info("result of writing version is %i", ret);
+ logger_->log_debug("result of writing version is %i", ret);
return false;
}
@@ -134,13 +134,13 @@ bool RawSiteToSiteClient::initiateResourceNegotiation() {
ret = peer_->read(statusCode);
if (ret <= 0) {
- logger_->log_info("result of writing version status code %i", ret);
+ logger_->log_debug("result of writing version status code %i", ret);
return false;
}
- logger_->log_info("status code is %i", statusCode);
+ logger_->log_debug("status code is %i", statusCode);
switch (statusCode) {
case RESOURCE_OK:
- logger_->log_info("Site2Site Protocol Negotiate protocol version OK");
+ logger_->log_debug("Site2Site Protocol Negotiate protocol version OK");
return true;
case DIFFERENT_RESOURCE_VERSION:
uint32_t serverVersion;
@@ -161,11 +161,11 @@ bool RawSiteToSiteClient::initiateResourceNegotiation() {
ret = -1;
return false;
case NEGOTIATED_ABORT:
- logger_->log_info("Site2Site Negotiate protocol response ABORT");
+ logger_->log_warn("Site2Site Negotiate protocol response ABORT");
ret = -1;
return false;
default:
- logger_->log_info("Negotiate protocol response unknown code %d", statusCode);
+ logger_->log_warn("Negotiate protocol response unknown code %d", statusCode);
return true;
}
@@ -179,7 +179,7 @@ bool RawSiteToSiteClient::initiateCodecResourceNegotiation() {
return false;
}
- logger_->log_info("Negotiate Codec version with destination port %s current version %d", port_id_str_, _currentCodecVersion);
+ logger_->log_trace("Negotiate Codec version with destination port %s current version %d", port_id_str_, _currentCodecVersion);
int ret = peer_->writeUTF(getCodecResourceName());
@@ -204,7 +204,7 @@ bool RawSiteToSiteClient::initiateCodecResourceNegotiation() {
switch (statusCode) {
case RESOURCE_OK:
- logger_->log_info("Site2Site Codec Negotiate version OK");
+ logger_->log_trace("Site2Site Codec Negotiate version OK");
return true;
case DIFFERENT_RESOURCE_VERSION:
uint32_t serverVersion;
@@ -224,11 +224,11 @@ bool RawSiteToSiteClient::initiateCodecResourceNegotiation() {
ret = -1;
return false;
case NEGOTIATED_ABORT:
- logger_->log_info("Site2Site Codec Negotiate response ABORT");
+ logger_->log_error("Site2Site Codec Negotiate response ABORT");
ret = -1;
return false;
default:
- logger_->log_info("Negotiate Codec response unknown code %d", statusCode);
+ logger_->log_error("Negotiate Codec response unknown code %d", statusCode);
return true;
}
@@ -240,7 +240,7 @@ bool RawSiteToSiteClient::handShake() {
logger_->log_error("Site2Site peer state is not established while handshake");
return false;
}
- logger_->log_info("Site2Site Protocol Perform hand shake with destination port %s", port_id_str_);
+ logger_->log_debug("Site2Site Protocol Perform hand shake with destination port %s", port_id_str_);
uuid_t uuid;
// Generate the global UUID for the com identify
id_generator_->generate(uuid);
@@ -290,7 +290,7 @@ bool RawSiteToSiteClient::handShake() {
if (ret <= 0) {
return false;
}
- logger_->log_info("Site2Site Protocol Send handshake properties %s %s", it->first.c_str(), it->second.c_str());
+ logger_->log_debug("Site2Site Protocol Send handshake properties %s %s", it->first, it->second);
}
RespondCode code;
@@ -304,7 +304,7 @@ bool RawSiteToSiteClient::handShake() {
switch (code) {
case PROPERTIES_OK:
- logger_->log_info("Site2Site HandShake Completed");
+ logger_->log_debug("Site2Site HandShake Completed");
peer_state_ = HANDSHAKED;
return true;
case PORT_NOT_IN_VALID_STATE:
@@ -314,7 +314,7 @@ bool RawSiteToSiteClient::handShake() {
ret = -1;
return false;
default:
- logger_->log_info("HandShake Failed because of unknown respond code %d", code);
+ logger_->log_error("HandShake Failed because of unknown respond code %d", code);
ret = -1;
return false;
}
@@ -324,7 +324,7 @@ bool RawSiteToSiteClient::handShake() {
void RawSiteToSiteClient::tearDown() {
if (peer_state_ >= ESTABLISHED) {
- logger_->log_info("Site2Site Protocol tearDown");
+ logger_->log_trace("Site2Site Protocol tearDown");
// need to write shutdown request
writeRequestType(SHUTDOWN);
}
@@ -378,9 +378,7 @@ bool RawSiteToSiteClient::getPeerList(std::vector<PeerStatus> &peers) {
}
PeerStatus status(std::make_shared<Peer>(port_id_, host, port, secure), count, true);
peers.push_back(std::move(status));
- std::stringstream str;
- str << "Site2Site Peer host " << host << " port " << port << " Secure " << secure;
- logger_->log_info(str.str().c_str());
+ logging::LOG_TRACE(logger_) << "Site2Site Peer host " << host << " port " << port << " Secure " << secure;
}
tearDown();
@@ -490,7 +488,7 @@ bool RawSiteToSiteClient::negotiateCodec() {
return false;
}
- logger_->log_info("Site2Site Protocol Negotiate Codec with destination port %s", port_id_str_);
+ logger_->log_trace("Site2Site Protocol Negotiate Codec with destination port %s", port_id_str_);
int status = writeRequestType(NEGOTIATE_FLOWFILE_CODEC);
@@ -506,7 +504,7 @@ bool RawSiteToSiteClient::negotiateCodec() {
return false;
}
- logger_->log_info("Site2Site Codec Completed and move to READY state for data transfer");
+ logger_->log_trace("Site2Site Codec Completed and move to READY state for data transfer");
peer_state_ = READY;
return true;
@@ -519,7 +517,7 @@ bool RawSiteToSiteClient::bootstrap() {
tearDown();
if (establish() && handShake() && negotiateCodec()) {
- logger_->log_info("Site2Site Ready For data transaction");
+ logger_->log_debug("Site to Site ready for data transaction");
return true;
} else {
peer_->yield();
@@ -561,24 +559,24 @@ std::shared_ptr<Transaction> RawSiteToSiteClient::createTransaction(std::string
switch (code) {
case MORE_DATA:
dataAvailable = true;
- logger_->log_info("Site2Site peer indicates that data is available");
+ logger_->log_trace("Site2Site peer indicates that data is available");
transaction = std::make_shared<Transaction>(direction, crcstream);
known_transactions_[transaction->getUUIDStr()] = transaction;
transactionID = transaction->getUUIDStr();
transaction->setDataAvailable(dataAvailable);
- logger_->log_info("Site2Site create transaction %s", transaction->getUUIDStr().c_str());
+ logger_->log_trace("Site2Site create transaction %s", transaction->getUUIDStr());
return transaction;
case NO_MORE_DATA:
dataAvailable = false;
- logger_->log_info("Site2Site peer indicates that no data is available");
+ logger_->log_trace("Site2Site peer indicates that no data is available");
transaction = std::make_shared<Transaction>(direction, crcstream);
known_transactions_[transaction->getUUIDStr()] = transaction;
transactionID = transaction->getUUIDStr();
transaction->setDataAvailable(dataAvailable);
- logger_->log_info("Site2Site create transaction %s", transaction->getUUIDStr().c_str());
+ logger_->log_trace("Site2Site create transaction %s", transaction->getUUIDStr());
return transaction;
default:
- logger_->log_info("Site2Site got unexpected response %d when asking for data", code);
+ logger_->log_warn("Site2Site got unexpected response %d when asking for data", code);
return NULL;
}
} else {
@@ -591,7 +589,7 @@ std::shared_ptr<Transaction> RawSiteToSiteClient::createTransaction(std::string
transaction = std::make_shared<Transaction>(direction, crcstream);
known_transactions_[transaction->getUUIDStr()] = transaction;
transactionID = transaction->getUUIDStr();
- logger_->log_info("Site2Site create transaction %s", transaction->getUUIDStr().c_str());
+ logger_->log_trace("Site2Site create transaction %s", transaction->getUUIDStr());
return transaction;
}
}
[3/3] nifi-minifi-cpp git commit: MINIFICPP-365 Adjusting log levels.
Posted by al...@apache.org.
MINIFICPP-365 Adjusting log levels.
Cleaning up c_str calls, adding logging for session events and adjusting Site to Site logs.
Project: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/repo
Commit: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/commit/233d1d44
Tree: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/tree/233d1d44
Diff: http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/diff/233d1d44
Branch: refs/heads/master
Commit: 233d1d44cbe92c4bf19648b8c3fb684f1792c03f
Parents: 8207e95
Author: Aldrin Piri <al...@apache.org>
Authored: Wed Jan 17 15:11:45 2018 -0500
Committer: Aldrin Piri <al...@apache.org>
Committed: Fri Jan 19 23:38:27 2018 -0500
----------------------------------------------------------------------
bootstrap.sh | 2 +-
extensions/civetweb/processors/ListenHTTP.cpp | 30 ++++-----
.../expression-language/ProcessContextExpr.cpp | 2 +-
extensions/gps/GetGPS.cpp | 6 +-
extensions/http-curl/client/HTTPClient.cpp | 10 +--
extensions/http-curl/processors/InvokeHTTP.cpp | 48 ++++++-------
extensions/http-curl/protocols/RESTSender.cpp | 2 +-
.../http-curl/sitetosite/HTTPProtocol.cpp | 18 ++---
extensions/libarchive/BinFiles.cpp | 32 ++++-----
extensions/libarchive/BinFiles.h | 6 +-
extensions/libarchive/FocusArchiveEntry.cpp | 8 +--
extensions/libarchive/MergeContent.cpp | 4 +-
extensions/libarchive/MergeContent.h | 2 +-
extensions/libarchive/UnfocusArchiveEntry.cpp | 8 +--
extensions/librdkafka/PublishKafka.cpp | 36 +++++-----
extensions/mqtt/AbstractMQTTProcessor.cpp | 18 ++---
extensions/mqtt/ConsumeMQTT.cpp | 2 +-
extensions/mqtt/PublishMQTT.cpp | 4 +-
.../rocksdb-repos/DatabaseContentRepository.cpp | 4 +-
extensions/rocksdb-repos/FlowFileRepository.cpp | 10 +--
extensions/rocksdb-repos/FlowFileRepository.h | 14 ++--
.../rocksdb-repos/ProvenanceRepository.cpp | 6 +-
extensions/rocksdb-repos/ProvenanceRepository.h | 14 ++--
extensions/usb-camera/GetUSBCamera.cpp | 18 ++---
libminifi/include/FlowControlProtocol.h | 2 +-
.../StandardControllerServiceProvider.h | 2 +-
.../core/repository/VolatileRepository.h | 2 +-
libminifi/include/processors/ListenSyslog.h | 2 +-
libminifi/src/Connection.cpp | 12 ++--
libminifi/src/FlowControlProtocol.cpp | 46 ++++++-------
libminifi/src/FlowController.cpp | 12 ++--
libminifi/src/FlowFileRecord.cpp | 20 +++---
libminifi/src/RemoteProcessorGroupPort.cpp | 28 ++++----
libminifi/src/ThreadedSchedulingAgent.cpp | 10 +--
libminifi/src/capi/Plan.cpp | 4 +-
libminifi/src/controllers/SSLContextService.cpp | 2 +-
libminifi/src/core/ConfigurableComponent.cpp | 6 +-
libminifi/src/core/Connectable.cpp | 8 +--
libminifi/src/core/FlowConfiguration.cpp | 2 +-
libminifi/src/core/ProcessGroup.cpp | 24 +++----
libminifi/src/core/ProcessSession.cpp | 43 ++++++------
.../src/core/ProcessSessionReadCallback.cpp | 8 +--
libminifi/src/core/Processor.cpp | 16 ++---
libminifi/src/core/Repository.cpp | 4 +-
.../repository/VolatileContentRepository.cpp | 2 +-
libminifi/src/core/yaml/YamlConfiguration.cpp | 9 ++-
libminifi/src/io/FileStream.cpp | 4 +-
libminifi/src/io/tls/TLSSocket.cpp | 8 +--
libminifi/src/processors/ExecuteProcess.cpp | 28 ++++----
libminifi/src/processors/GetFile.cpp | 9 ++-
libminifi/src/processors/GetTCP.cpp | 10 +--
libminifi/src/processors/ListenSyslog.cpp | 6 +-
libminifi/src/processors/LogAttribute.cpp | 12 ++--
libminifi/src/processors/PutFile.cpp | 28 ++++----
libminifi/src/processors/TailFile.cpp | 8 +--
libminifi/src/sitetosite/RawSocketProtocol.cpp | 58 ++++++++--------
libminifi/src/sitetosite/SiteToSiteClient.cpp | 71 +++++++++++---------
main/MiNiFiMain.cpp | 2 +-
58 files changed, 407 insertions(+), 405 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/bootstrap.sh
----------------------------------------------------------------------
diff --git a/bootstrap.sh b/bootstrap.sh
index bc8b1de..0bef12b 100755
--- a/bootstrap.sh
+++ b/bootstrap.sh
@@ -355,7 +355,7 @@ show_supported_features() {
echo "B. Lib Curl Features ...........$(print_feature_status HTTP_CURL_ENABLED)"
echo "C. Lib Archive Features ........$(print_feature_status LIBARCHIVE_ENABLED)"
echo "D. Execute Script support ......$(print_feature_status EXECUTE_SCRIPT_ENABLED)"
- echo "E. Expression Langauge support .$(print_feature_status EXPRESSION_LANGAUGE_ENABLED)"
+ echo "E. Expression Language support .$(print_feature_status EXPRESSION_LANGAUGE_ENABLED)"
echo "F. Kafka support ...............$(print_feature_status KAFKA_ENABLED)"
echo "G. PCAP support ................$(print_feature_status PCAP_ENABLED)"
echo "H. USB Camera support ..........$(print_feature_status USB_ENABLED)"
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/civetweb/processors/ListenHTTP.cpp
----------------------------------------------------------------------
diff --git a/extensions/civetweb/processors/ListenHTTP.cpp b/extensions/civetweb/processors/ListenHTTP.cpp
index 73ade40..d94df9b 100644
--- a/extensions/civetweb/processors/ListenHTTP.cpp
+++ b/extensions/civetweb/processors/ListenHTTP.cpp
@@ -58,7 +58,7 @@ core::Property ListenHTTP::HeadersAsAttributesRegex("HTTP Headers to receive as
core::Relationship ListenHTTP::Success("success", "All files are routed to success");
void ListenHTTP::initialize() {
- logger_->log_info("Initializing ListenHTTP");
+ logger_->log_trace("Initializing ListenHTTP");
// Set the supported properties
std::set<core::Property> properties;
@@ -81,7 +81,7 @@ void ListenHTTP::onSchedule(core::ProcessContext *context, core::ProcessSessionF
std::string basePath;
if (!context->getProperty(BasePath.getName(), basePath)) {
- logger_->log_info("%s attribute is missing, so default value of %s will be used", BasePath.getName().c_str(), BasePath.getValue().c_str());
+ logger_->log_info("%s attribute is missing, so default value of %s will be used", BasePath.getName(), BasePath.getValue());
basePath = BasePath.getValue();
}
@@ -90,20 +90,20 @@ void ListenHTTP::onSchedule(core::ProcessContext *context, core::ProcessSessionF
std::string listeningPort;
if (!context->getProperty(Port.getName(), listeningPort)) {
- logger_->log_error("%s attribute is missing or invalid", Port.getName().c_str());
+ logger_->log_error("%s attribute is missing or invalid", Port.getName());
return;
}
std::string authDNPattern;
if (context->getProperty(AuthorizedDNPattern.getName(), authDNPattern) && !authDNPattern.empty()) {
- logger_->log_info("ListenHTTP using %s: %s", AuthorizedDNPattern.getName().c_str(), authDNPattern.c_str());
+ logger_->log_debug("ListenHTTP using %s: %s", AuthorizedDNPattern.getName(), authDNPattern);
}
std::string sslCertFile;
if (context->getProperty(SSLCertificate.getName(), sslCertFile) && !sslCertFile.empty()) {
- logger_->log_info("ListenHTTP using %s: %s", SSLCertificate.getName().c_str(), sslCertFile.c_str());
+ logger_->log_debug("ListenHTTP using %s: %s", SSLCertificate.getName(), sslCertFile);
}
// Read further TLS/SSL options only if TLS/SSL usage is implied by virtue of certificate value being set
@@ -113,33 +113,33 @@ void ListenHTTP::onSchedule(core::ProcessContext *context, core::ProcessSessionF
if (!sslCertFile.empty()) {
if (context->getProperty(SSLCertificateAuthority.getName(), sslCertAuthorityFile) && !sslCertAuthorityFile.empty()) {
- logger_->log_info("ListenHTTP using %s: %s", SSLCertificateAuthority.getName().c_str(), sslCertAuthorityFile.c_str());
+ logger_->log_debug("ListenHTTP using %s: %s", SSLCertificateAuthority.getName(), sslCertAuthorityFile);
}
if (context->getProperty(SSLVerifyPeer.getName(), sslVerifyPeer)) {
if (sslVerifyPeer.empty() || sslVerifyPeer.compare("no") == 0) {
- logger_->log_info("ListenHTTP will not verify peers");
+ logger_->log_debug("ListenHTTP will not verify peers");
} else {
- logger_->log_info("ListenHTTP will verify peers");
+ logger_->log_debug("ListenHTTP will verify peers");
}
} else {
- logger_->log_info("ListenHTTP will not verify peers");
+ logger_->log_debug("ListenHTTP will not verify peers");
}
if (context->getProperty(SSLMinimumVersion.getName(), sslMinVer)) {
- logger_->log_info("ListenHTTP using %s: %s", SSLMinimumVersion.getName().c_str(), sslMinVer.c_str());
+ logger_->log_debug("ListenHTTP using %s: %s", SSLMinimumVersion.getName(), sslMinVer);
}
}
std::string headersAsAttributesPattern;
if (context->getProperty(HeadersAsAttributesRegex.getName(), headersAsAttributesPattern) && !headersAsAttributesPattern.empty()) {
- logger_->log_info("ListenHTTP using %s: %s", HeadersAsAttributesRegex.getName().c_str(), headersAsAttributesPattern.c_str());
+ logger_->log_debug("ListenHTTP using %s: %s", HeadersAsAttributesRegex.getName(), headersAsAttributesPattern);
}
auto numThreads = getMaxConcurrentTasks();
- logger_->log_info("ListenHTTP starting HTTP server on port %s and path %s with %d threads", listeningPort.c_str(), basePath.c_str(), numThreads);
+ logger_->log_info("ListenHTTP starting HTTP server on port %s and path %s with %d threads", listeningPort, basePath, numThreads);
// Initialize web server
std::vector<std::string> options;
@@ -225,7 +225,7 @@ void ListenHTTP::Handler::sendErrorResponse(struct mg_connection *conn) {
bool ListenHTTP::Handler::handlePost(CivetServer *server, struct mg_connection *conn) {
auto req_info = mg_get_request_info(conn);
- logger_->log_info("ListenHTTP handling POST request of length %ll", req_info->content_length);
+ logger_->log_debug("ListenHTTP handling POST request of length %ll", req_info->content_length);
// If this is a two-way TLS connection, authorize the peer against the configured pattern
if (req_info->is_ssl && req_info->client_cert != nullptr) {
@@ -271,12 +271,12 @@ bool ListenHTTP::Handler::handlePost(CivetServer *server, struct mg_connection *
session->transfer(flowFile, Success);
session->commit();
} catch (std::exception &exception) {
- logger_->log_debug("ListenHTTP Caught Exception %s", exception.what());
+ logger_->log_error("ListenHTTP Caught Exception %s", exception.what());
sendErrorResponse(conn);
session->rollback();
throw;
} catch (...) {
- logger_->log_debug("ListenHTTP Caught Exception Processor::onTrigger");
+ logger_->log_error("ListenHTTP Caught Exception Processor::onTrigger");
sendErrorResponse(conn);
session->rollback();
throw;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/expression-language/ProcessContextExpr.cpp
----------------------------------------------------------------------
diff --git a/extensions/expression-language/ProcessContextExpr.cpp b/extensions/expression-language/ProcessContextExpr.cpp
index d442978..0748e2d 100644
--- a/extensions/expression-language/ProcessContextExpr.cpp
+++ b/extensions/expression-language/ProcessContextExpr.cpp
@@ -28,7 +28,7 @@ bool ProcessContext::getProperty(const std::string &name, std::string &value,
if (expressions_.find(name) == expressions_.end()) {
std::string expression_str;
getProperty(name, expression_str);
- logger_->log_info("Compiling expression for %s/%s: %s", getProcessorNode()->getName(), name, expression_str);
+ logger_->log_debug("Compiling expression for %s/%s: %s", getProcessorNode()->getName(), name, expression_str);
expressions_.emplace(name, expression::compile(expression_str));
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/gps/GetGPS.cpp
----------------------------------------------------------------------
diff --git a/extensions/gps/GetGPS.cpp b/extensions/gps/GetGPS.cpp
index 64e8618..08a5d16 100644
--- a/extensions/gps/GetGPS.cpp
+++ b/extensions/gps/GetGPS.cpp
@@ -80,7 +80,7 @@ void GetGPS::onSchedule(const std::shared_ptr<core::ProcessContext> &context, co
if (context->getProperty(GPSDWaitTime.getName(), value)) {
core::Property::StringToInt(value, gpsdWaitTime_);
}
- logger_->log_info("GPSD client scheduled");
+ logger_->log_trace("GPSD client scheduled");
}
void GetGPS::onTrigger(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) {
@@ -106,7 +106,7 @@ void GetGPS::onTrigger(const std::shared_ptr<core::ProcessContext> &context, con
if (gpsdata->status > 0) {
if (gpsdata->fix.longitude != gpsdata->fix.longitude || gpsdata->fix.altitude != gpsdata->fix.altitude) {
- logger_->log_info("No GPS fix.\n");
+ logger_->log_info("No GPS fix.");
continue;
}
@@ -146,7 +146,7 @@ void GetGPS::onTrigger(const std::shared_ptr<core::ProcessContext> &context, con
}
} catch (std::exception &exception) {
- logger_->log_debug("GetGPS Caught Exception %s", exception.what());
+ logger_->log_error("GetGPS Caught Exception %s", exception.what());
throw;
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/http-curl/client/HTTPClient.cpp
----------------------------------------------------------------------
diff --git a/extensions/http-curl/client/HTTPClient.cpp b/extensions/http-curl/client/HTTPClient.cpp
index 707d301..31214cc 100644
--- a/extensions/http-curl/client/HTTPClient.cpp
+++ b/extensions/http-curl/client/HTTPClient.cpp
@@ -98,7 +98,7 @@ HTTPClient::~HTTPClient() {
curl_easy_cleanup(http_session_);
http_session_ = nullptr;
}
- logger_->log_info("Closing HTTPClient for %s", url_);
+ logger_->log_trace("Closing HTTPClient for %s", url_);
}
void HTTPClient::forceClose() {
@@ -132,7 +132,7 @@ void HTTPClient::initialize(const std::string &method, const std::string url, co
}
void HTTPClient::setDisablePeerVerification() {
- logger_->log_info("Disabling peer verification");
+ logger_->log_debug("Disabling peer verification");
curl_easy_setopt(http_session_, CURLOPT_SSL_VERIFYPEER, 0L);
}
@@ -152,7 +152,7 @@ void HTTPClient::setReadCallback(HTTPReadCallback *callbackObj) {
}
void HTTPClient::setUploadCallback(HTTPUploadCallback *callbackObj) {
- logger_->log_info("Setting callback for %s", url_);
+ logger_->log_debug("Setting callback for %s", url_);
write_callback_ = callbackObj;
if (method_ == "put" || method_ == "PUT") {
curl_easy_setopt(http_session_, CURLOPT_INFILESIZE_LARGE, (curl_off_t ) callbackObj->ptr->getBufferSize());
@@ -218,7 +218,7 @@ bool HTTPClient::submit() {
}
curl_easy_setopt(http_session_, CURLOPT_URL, url_.c_str());
- logger_->log_info("Submitting to %s", url_);
+ logger_->log_debug("Submitting to %s", url_);
if (callback == nullptr) {
content_.ptr = &read_callback_;
curl_easy_setopt(http_session_, CURLOPT_WRITEFUNCTION, &utils::HTTPRequestResponse::recieve_write);
@@ -240,7 +240,7 @@ bool HTTPClient::submit() {
return false;
}
- logger_->log_info("Finished with %s", url_);
+ logger_->log_debug("Finished with %s", url_);
std::string key = "";
for (auto header_line : header_response_.header_tokens_) {
unsigned int i = 0;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/http-curl/processors/InvokeHTTP.cpp
----------------------------------------------------------------------
diff --git a/extensions/http-curl/processors/InvokeHTTP.cpp b/extensions/http-curl/processors/InvokeHTTP.cpp
index a8bcbab..6b501a8 100644
--- a/extensions/http-curl/processors/InvokeHTTP.cpp
+++ b/extensions/http-curl/processors/InvokeHTTP.cpp
@@ -112,7 +112,7 @@ core::Relationship InvokeHTTP::RelFailure("failure", "The original FlowFile will
"timeout or general exception. It will have new attributes detailing the request.");
void InvokeHTTP::initialize() {
- logger_->log_info("Initializing InvokeHTTP");
+ logger_->log_trace("Initializing InvokeHTTP");
// Set the supported properties
std::set<core::Property> properties;
@@ -142,12 +142,12 @@ void InvokeHTTP::initialize() {
void InvokeHTTP::onSchedule(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSessionFactory> &sessionFactory) {
if (!context->getProperty(Method.getName(), method_)) {
- logger_->log_info("%s attribute is missing, so default value of %s will be used", Method.getName().c_str(), Method.getValue().c_str());
+ logger_->log_debug("%s attribute is missing, so default value of %s will be used", Method.getName(), Method.getValue());
return;
}
if (!context->getProperty(URL.getName(), url_)) {
- logger_->log_info("%s attribute is missing, so default value of %s will be used", URL.getName().c_str(), URL.getValue().c_str());
+ logger_->log_debug("%s attribute is missing, so default value of %s will be used", URL.getName(), URL.getValue());
return;
}
@@ -158,7 +158,7 @@ void InvokeHTTP::onSchedule(const std::shared_ptr<core::ProcessContext> &context
// set the timeout in curl options.
} else {
- logger_->log_info("%s attribute is missing, so default value of %s will be used", ConnectTimeout.getName().c_str(), ConnectTimeout.getValue().c_str());
+ logger_->log_debug("%s attribute is missing, so default value of %s will be used", ConnectTimeout.getName(), ConnectTimeout.getValue());
return;
}
@@ -172,34 +172,34 @@ void InvokeHTTP::onSchedule(const std::shared_ptr<core::ProcessContext> &context
core::Property::StringToInt(timeoutStr, read_timeout_);
} else {
- logger_->log_info("%s attribute is missing, so default value of %s will be used", ReadTimeout.getName().c_str(), ReadTimeout.getValue().c_str());
+ logger_->log_debug("%s attribute is missing, so default value of %s will be used", ReadTimeout.getName(), ReadTimeout.getValue());
}
std::string dateHeaderStr;
if (!context->getProperty(DateHeader.getName(), dateHeaderStr)) {
- logger_->log_info("%s attribute is missing, so default value of %s will be used", DateHeader.getName().c_str(), DateHeader.getValue().c_str());
+ logger_->log_debug("%s attribute is missing, so default value of %s will be used", DateHeader.getName(), DateHeader.getValue());
}
date_header_include_ = utils::StringUtils::StringToBool(dateHeaderStr, date_header_include_);
if (!context->getProperty(PropPutOutputAttributes.getName(), put_attribute_name_)) {
- logger_->log_info("%s attribute is missing, so default value of %s will be used", PropPutOutputAttributes.getName().c_str(), PropPutOutputAttributes.getValue().c_str());
+ logger_->log_debug("%s attribute is missing, so default value of %s will be used", PropPutOutputAttributes.getName(), PropPutOutputAttributes.getValue());
}
if (!context->getProperty(AttributesToSend.getName(), attribute_to_send_regex_)) {
- logger_->log_info("%s attribute is missing, so default value of %s will be used", AttributesToSend.getName().c_str(), AttributesToSend.getValue().c_str());
+ logger_->log_debug("%s attribute is missing, so default value of %s will be used", AttributesToSend.getName(), AttributesToSend.getValue());
}
std::string always_output_response = "false";
if (!context->getProperty(AlwaysOutputResponse.getName(), always_output_response)) {
- logger_->log_info("%s attribute is missing, so default value of %s will be used", AlwaysOutputResponse.getName().c_str(), AlwaysOutputResponse.getValue().c_str());
+ logger_->log_debug("%s attribute is missing, so default value of %s will be used", AlwaysOutputResponse.getName(), AlwaysOutputResponse.getValue());
}
utils::StringUtils::StringToBool(always_output_response, always_output_response_);
std::string penalize_no_retry = "false";
if (!context->getProperty(PenalizeOnNoRetry.getName(), penalize_no_retry)) {
- logger_->log_info("%s attribute is missing, so default value of %s will be used", PenalizeOnNoRetry.getName().c_str(), PenalizeOnNoRetry.getValue().c_str());
+ logger_->log_debug("%s attribute is missing, so default value of %s will be used", PenalizeOnNoRetry.getName(), PenalizeOnNoRetry.getValue());
}
utils::StringUtils::StringToBool(penalize_no_retry, penalize_no_retry_);
@@ -214,7 +214,7 @@ void InvokeHTTP::onSchedule(const std::shared_ptr<core::ProcessContext> &context
std::string useChunkedEncoding = "false";
if (!context->getProperty(UseChunkedEncoding.getName(), useChunkedEncoding)) {
- logger_->log_info("%s attribute is missing, so default value of %s will be used", UseChunkedEncoding.getName().c_str(), UseChunkedEncoding.getValue().c_str());
+ logger_->log_debug("%s attribute is missing, so default value of %s will be used", UseChunkedEncoding.getName(), UseChunkedEncoding.getValue());
}
utils::StringUtils::StringToBool(useChunkedEncoding, use_chunked_encoding_);
@@ -247,18 +247,18 @@ void InvokeHTTP::onTrigger(const std::shared_ptr<core::ProcessContext> &context,
if (flowFile == nullptr) {
if (!emitFlowFile(method_)) {
- logger_->log_info("InvokeHTTP -- create flow file with %s", method_.c_str());
+ logger_->log_debug("InvokeHTTP -- create flow file with %s", method_);
flowFile = std::static_pointer_cast<FlowFileRecord>(session->create());
} else {
- logger_->log_info("exiting because method is %s", method_.c_str());
+ logger_->log_debug("exiting because method is %s", method_);
return;
}
} else {
context->getProperty(URL.getName(), url, flowFile);
- logger_->log_info("InvokeHTTP -- Received flowfile");
+ logger_->log_debug("InvokeHTTP -- Received flowfile");
}
- logger_->log_info("onTrigger InvokeHTTP with %s to %s", method_, url);
+ logger_->log_debug("onTrigger InvokeHTTP with %s to %s", method_, url);
// create a transaction id
std::string tx_id = generateId();
@@ -285,7 +285,7 @@ void InvokeHTTP::onTrigger(const std::shared_ptr<core::ProcessContext> &context,
std::unique_ptr<utils::ByteInputCallBack> callback = nullptr;
std::unique_ptr<utils::HTTPUploadCallback> callbackObj = nullptr;
if (emitFlowFile(method_)) {
- logger_->log_info("InvokeHTTP -- reading flowfile");
+ logger_->log_trace("InvokeHTTP -- reading flowfile");
std::shared_ptr<ResourceClaim> claim = flowFile->getResourceClaim();
if (claim) {
callback = std::unique_ptr<utils::ByteInputCallBack>(new utils::ByteInputCallBack());
@@ -293,7 +293,7 @@ void InvokeHTTP::onTrigger(const std::shared_ptr<core::ProcessContext> &context,
callbackObj = std::unique_ptr<utils::HTTPUploadCallback>(new utils::HTTPUploadCallback);
callbackObj->ptr = callback.get();
callbackObj->pos = 0;
- logger_->log_info("InvokeHTTP -- Setting callback, size is %d", callback->getBufferSize());
+ logger_->log_trace("InvokeHTTP -- Setting callback, size is %d", callback->getBufferSize());
if (!use_chunked_encoding_) {
client.appendHeader("Content-Length", std::to_string(flowFile->getSize()));
}
@@ -303,15 +303,15 @@ void InvokeHTTP::onTrigger(const std::shared_ptr<core::ProcessContext> &context,
}
} else {
- logger_->log_info("InvokeHTTP -- Not emitting flowfile to HTTP Server");
+ logger_->log_trace("InvokeHTTP -- Not emitting flowfile to HTTP Server");
}
// append all headers
client.build_header_list(attribute_to_send_regex_, flowFile->getAttributes());
- logger_->log_info("InvokeHTTP -- curl performed");
+ logger_->log_trace("InvokeHTTP -- curl performed");
if (client.submit()) {
- logger_->log_info("InvokeHTTP -- curl successful");
+ logger_->log_trace("InvokeHTTP -- curl successful");
bool putToAttribute = !IsNullOrEmpty(put_attribute_name_);
@@ -329,7 +329,7 @@ void InvokeHTTP::onTrigger(const std::shared_ptr<core::ProcessContext> &context,
bool isSuccess = ((int32_t) (http_code / 100)) == 2;
bool output_body_to_content = isSuccess && !putToAttribute;
- logger_->log_info("isSuccess: %d, response code %d", isSuccess, http_code);
+ logger_->log_debug("isSuccess: %d, response code %d", isSuccess, http_code);
std::shared_ptr<FlowFileRecord> response_flow = nullptr;
if (output_body_to_content) {
@@ -350,7 +350,7 @@ void InvokeHTTP::onTrigger(const std::shared_ptr<core::ProcessContext> &context,
// need an import from the data stream.
session->importFrom(stream, response_flow);
} else {
- logger_->log_info("Cannot output body to content");
+ logger_->log_warn("Cannot output body to content");
response_flow = std::static_pointer_cast<FlowFileRecord>(session->create());
}
route(flowFile, response_flow, session, context, isSuccess, http_code);
@@ -371,7 +371,7 @@ void InvokeHTTP::route(std::shared_ptr<FlowFileRecord> &request,
// If the property to output the response flowfile regardless of status code is set then transfer it
bool responseSent = false;
if (always_output_response_ && response != nullptr) {
- logger_->log_info("Outputting success and response");
+ logger_->log_debug("Outputting success and response");
session->transfer(response, Success);
responseSent = true;
}
@@ -384,7 +384,7 @@ void InvokeHTTP::route(std::shared_ptr<FlowFileRecord> &request,
session->transfer(request, Success);
}
if (response != nullptr && !responseSent) {
- logger_->log_info("Outputting success and response");
+ logger_->log_debug("Outputting success and response");
session->transfer(response, Success);
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/http-curl/protocols/RESTSender.cpp
----------------------------------------------------------------------
diff --git a/extensions/http-curl/protocols/RESTSender.cpp b/extensions/http-curl/protocols/RESTSender.cpp
index ebf532a..5f81005 100644
--- a/extensions/http-curl/protocols/RESTSender.cpp
+++ b/extensions/http-curl/protocols/RESTSender.cpp
@@ -43,7 +43,7 @@ void RESTSender::initialize(const std::shared_ptr<core::controller::ControllerSe
configure->get("c2.rest.url", rest_uri_);
configure->get("c2.rest.url.ack", ack_uri_);
}
- logger_->log_info("Submitting to %s", rest_uri_);
+ logger_->log_debug("Submitting to %s", rest_uri_);
}
C2Payload RESTSender::consumePayload(const std::string &url, const C2Payload &payload, Direction direction, bool async) {
std::string operation_request_str = getOperation(payload);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/http-curl/sitetosite/HTTPProtocol.cpp
----------------------------------------------------------------------
diff --git a/extensions/http-curl/sitetosite/HTTPProtocol.cpp b/extensions/http-curl/sitetosite/HTTPProtocol.cpp
index d70ab45..e83c71d 100644
--- a/extensions/http-curl/sitetosite/HTTPProtocol.cpp
+++ b/extensions/http-curl/sitetosite/HTTPProtocol.cpp
@@ -67,7 +67,7 @@ std::shared_ptr<Transaction> HttpSiteToSiteClient::createTransaction(std::string
client->setPostFields("");
client->submit();
if (peer_->getStream() != nullptr)
- logger_->log_info("Closing %s",((io::HttpStream*)peer_->getStream())->getClientRef()->getURL());
+ logger_->log_debug("Closing %s",((io::HttpStream*)peer_->getStream())->getClientRef()->getURL());
if (client->getResponseCode() == 201) {
// parse the headers
auto headers = client->getParsedHeaders();
@@ -135,7 +135,7 @@ int HttpSiteToSiteClient::readResponse(const std::shared_ptr<Transaction> &trans
if (current_code == CONFIRM_TRANSACTION && transaction->getState() == DATA_EXCHANGED) {
auto stream = dynamic_cast<io::HttpStream*>(peer_->getStream());
if (!stream->isFinished()) {
- logger_->log_info("confirm read for %s, but not finished ", transaction->getUUIDStr());
+ logger_->log_debug("confirm read for %s, but not finished ", transaction->getUUIDStr());
if (stream->waitForDataAvailable()) {
code = CONTINUE_TRANSACTION;
}
@@ -146,15 +146,15 @@ int HttpSiteToSiteClient::readResponse(const std::shared_ptr<Transaction> &trans
} else {
auto stream = dynamic_cast<io::HttpStream*>(peer_->getStream());
if (stream->isFinished()) {
- logger_->log_info("Finished %s ", transaction->getUUIDStr());
+ logger_->log_debug("Finished %s ", transaction->getUUIDStr());
code = FINISH_TRANSACTION;
current_code = FINISH_TRANSACTION;
} else {
if (stream->waitForDataAvailable()) {
- logger_->log_info("data is available, so continuing transaction %s ", transaction->getUUIDStr());
+ logger_->log_debug("data is available, so continuing transaction %s ", transaction->getUUIDStr());
code = CONTINUE_TRANSACTION;
} else {
- logger_->log_info("No data available for transaction %s ", transaction->getUUIDStr());
+ logger_->log_debug("No data available for transaction %s ", transaction->getUUIDStr());
code = FINISH_TRANSACTION;
current_code = FINISH_TRANSACTION;
}
@@ -183,7 +183,7 @@ int HttpSiteToSiteClient::writeResponse(const std::shared_ptr<Transaction> &tran
return 1;
} else if (code == CONTINUE_TRANSACTION) {
- logger_->log_info("Continuing HTTP transaction");
+ logger_->log_debug("Continuing HTTP transaction");
return 1;
}
return SiteToSiteClient::writeResponse(transaction, code, message);
@@ -233,7 +233,7 @@ bool HttpSiteToSiteClient::transmitPayload(const std::shared_ptr<core::ProcessCo
void HttpSiteToSiteClient::tearDown() {
if (peer_state_ >= ESTABLISHED) {
- logger_->log_info("Site2Site Protocol tearDown");
+ logger_->log_debug("Site2Site Protocol tearDown");
}
known_transactions_.clear();
peer_->Close();
@@ -256,7 +256,7 @@ void HttpSiteToSiteClient::closeTransaction(const std::string &transactionID) {
}
std::string append_str;
- logger_->log_info("Site2Site close transaction %s", transaction->getUUIDStr().c_str());
+ logger_->log_info("Site to Site closed transaction %s", transaction->getUUIDStr());
int code = UNRECOGNIZED_RESPONSE_CODE;
if (transaction->getState() == TRANSACTION_CONFIRMED) {
@@ -314,7 +314,7 @@ void HttpSiteToSiteClient::deleteTransaction(std::string transactionID) {
}
std::string append_str;
- logger_->log_info("Site2Site delete transaction %s", transaction->getUUIDStr().c_str());
+ logger_->log_debug("Site2Site delete transaction %s", transaction->getUUIDStr());
closeTransaction(transactionID);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/libarchive/BinFiles.cpp
----------------------------------------------------------------------
diff --git a/extensions/libarchive/BinFiles.cpp b/extensions/libarchive/BinFiles.cpp
index 34afcf0..035ed29 100644
--- a/extensions/libarchive/BinFiles.cpp
+++ b/extensions/libarchive/BinFiles.cpp
@@ -77,34 +77,34 @@ void BinFiles::onSchedule(core::ProcessContext *context, core::ProcessSessionFac
int64_t valInt;
if (context->getProperty(MinSize.getName(), value) && !value.empty() && core::Property::StringToInt(value, valInt)) {
this->binManager_.setMinSize(valInt);
- logger_->log_info("BinFiles: MinSize [%d]", valInt);
+ logger_->log_debug("BinFiles: MinSize [%d]", valInt);
}
value = "";
if (context->getProperty(MaxSize.getName(), value) && !value.empty() && core::Property::StringToInt(value, valInt)) {
this->binManager_.setMaxSize(valInt);
- logger_->log_info("BinFiles: MaxSize [%d]", valInt);
+ logger_->log_debug("BinFiles: MaxSize [%d]", valInt);
}
value = "";
if (context->getProperty(MinEntries.getName(), value) && !value.empty() && core::Property::StringToInt(value, valInt)) {
this->binManager_.setMinEntries(valInt);
- logger_->log_info("BinFiles: MinEntries [%d]", valInt);
+ logger_->log_debug("BinFiles: MinEntries [%d]", valInt);
}
value = "";
if (context->getProperty(MaxEntries.getName(), value) && !value.empty() && core::Property::StringToInt(value, valInt)) {
this->binManager_.setMaxEntries(valInt);
- logger_->log_info("BinFiles: MaxEntries [%d]", valInt);
+ logger_->log_debug("BinFiles: MaxEntries [%d]", valInt);
}
value = "";
if (context->getProperty(MaxBinCount.getName(), value) && !value.empty() && core::Property::StringToInt(value, valInt)) {
maxBinCount_ = static_cast<int> (valInt);
- logger_->log_info("BinFiles: MaxBinCount [%d]", valInt);
+ logger_->log_debug("BinFiles: MaxBinCount [%d]", valInt);
}
value = "";
if (context->getProperty(MaxBinAge.getName(), value) && !value.empty()) {
core::TimeUnit unit;
if (core::Property::StringToTime(value, valInt, unit) && core::Property::ConvertTimeUnitToMS(valInt, unit, valInt)) {
this->binManager_.setBinAge(valInt);
- logger_->log_info("BinFiles: MaxBinAge [%d]", valInt);
+ logger_->log_debug("BinFiles: MaxBinAge [%d]", valInt);
}
}
}
@@ -134,7 +134,7 @@ void BinManager::gatherReadyBins() {
readyBin_.push_back(std::move(bin));
queue->pop_front();
binCount_--;
- logger_->log_info("BinManager move bin %s to ready bins for group %s", readyBin_.back()->getUUIDStr(), readyBin_.back()->getGroupId());
+ logger_->log_debug("BinManager move bin %s to ready bins for group %s", readyBin_.back()->getUUIDStr(), readyBin_.back()->getGroupId());
} else {
break;
}
@@ -147,7 +147,7 @@ void BinManager::gatherReadyBins() {
// erase from the map if the queue is empty for the group
groupBinMap_.erase(group);
}
- logger_->log_info("BinManager groupBinMap size %d", groupBinMap_.size());
+ logger_->log_debug("BinManager groupBinMap size %d", groupBinMap_.size());
}
void BinManager::removeOldestBin() {
@@ -170,12 +170,12 @@ void BinManager::removeOldestBin() {
readyBin_.push_back(std::move(remove));
(*oldqueue)->pop_front();
binCount_--;
- logger_->log_info("BinManager move bin %s to ready bins for group %s", readyBin_.back()->getUUIDStr(), readyBin_.back()->getGroupId());
+ logger_->log_debug("BinManager move bin %s to ready bins for group %s", readyBin_.back()->getUUIDStr(), readyBin_.back()->getGroupId());
if ((*oldqueue)->empty()) {
groupBinMap_.erase(group);
}
}
- logger_->log_info("BinManager groupBinMap size %d", groupBinMap_.size());
+ logger_->log_debug("BinManager groupBinMap size %d", groupBinMap_.size());
}
void BinManager::getReadyBin(std::deque<std::unique_ptr<Bin>> &retBins) {
@@ -195,7 +195,7 @@ bool BinManager::offer(const std::string &group, std::shared_ptr<core::FlowFile>
if (!bin->offer(flow))
return false;
readyBin_.push_back(std::move(bin));
- logger_->log_info("BinManager move bin %s to ready bins for group %s", readyBin_.back()->getUUIDStr(), group);
+ logger_->log_debug("BinManager move bin %s to ready bins for group %s", readyBin_.back()->getUUIDStr(), group);
return true;
}
auto search = groupBinMap_.find(group);
@@ -209,7 +209,7 @@ bool BinManager::offer(const std::string &group, std::shared_ptr<core::FlowFile>
if (!bin->offer(flow))
return false;
queue->push_back(std::move(bin));
- logger_->log_info("BinManager add bin %s to group %s", queue->back()->getUUIDStr(), group);
+ logger_->log_debug("BinManager add bin %s to group %s", queue->back()->getUUIDStr(), group);
binCount_++;
}
} else {
@@ -218,7 +218,7 @@ bool BinManager::offer(const std::string &group, std::shared_ptr<core::FlowFile>
return false;
queue->push_back(std::move(bin));
binCount_++;
- logger_->log_info("BinManager add bin %s to group %s", queue->back()->getUUIDStr(), group);
+ logger_->log_debug("BinManager add bin %s to group %s", queue->back()->getUUIDStr(), group);
}
} else {
std::unique_ptr<std::deque<std::unique_ptr<Bin>>> queue = std::unique_ptr<std::deque<std::unique_ptr<Bin>>> (new std::deque<std::unique_ptr<Bin>>());
@@ -226,7 +226,7 @@ bool BinManager::offer(const std::string &group, std::shared_ptr<core::FlowFile>
if (!bin->offer(flow))
return false;
queue->push_back(std::move(bin));
- logger_->log_info("BinManager add bin %s to group %s", queue->back()->getUUIDStr(), group);
+ logger_->log_debug("BinManager add bin %s to group %s", queue->back()->getUUIDStr(), group);
groupBinMap_.insert(std::make_pair(group, std::move(queue)));
binCount_++;
}
@@ -257,7 +257,7 @@ void BinFiles::onTrigger(const std::shared_ptr<core::ProcessContext> &context, c
if (this->binManager_.getBinCount() > maxBinCount_) {
// bin count reach max allowed
context->yield();
- logger_->log_info("BinFiles reach max bin count %d", this->binManager_.getBinCount());
+ logger_->log_debug("BinFiles reach max bin count %d", this->binManager_.getBinCount());
this->binManager_.removeOldestBin();
}
@@ -274,7 +274,7 @@ void BinFiles::onTrigger(const std::shared_ptr<core::ProcessContext> &context, c
readyBins.pop_front();
// add bin's flows to the session
this->addFlowsToSession(context.get(), &mergeSession, bin);
- logger_->log_info("BinFiles start to process bin %s for group %s", bin->getUUIDStr(), bin->getGroupId());
+ logger_->log_debug("BinFiles start to process bin %s for group %s", bin->getUUIDStr(), bin->getGroupId());
if (!this->processBin(context.get(), &mergeSession, bin))
this->transferFlowsToFail(context.get(), &mergeSession, bin);
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/libarchive/BinFiles.h
----------------------------------------------------------------------
diff --git a/extensions/libarchive/BinFiles.h b/extensions/libarchive/BinFiles.h
index db8a6b8..9a463b2 100644
--- a/extensions/libarchive/BinFiles.h
+++ b/extensions/libarchive/BinFiles.h
@@ -59,10 +59,10 @@ class Bin {
id_generator->generate(uuid_);
uuid_unparse_lower(uuid_, uuidStr);
uuid_str_ = uuidStr;
- logger_->log_info("Bin %s for group %s created", uuid_str_, groupId_);
+ logger_->log_debug("Bin %s for group %s created", uuid_str_, groupId_);
}
virtual ~Bin() {
- logger_->log_info("Bin %s for group %s destroyed", uuid_str_, groupId_);
+ logger_->log_debug("Bin %s for group %s destroyed", uuid_str_, groupId_);
}
// check whether the bin is full
bool isFull() {
@@ -107,7 +107,7 @@ class Bin {
queue_.push_back(flow);
queued_data_size_ += flow->getSize();
- logger_->log_info("Bin %s for group %s offer size %d byte %d min_entry %d max_entry %d", uuid_str_, groupId_, queue_.size(), queued_data_size_, minEntries_, maxEntries_);
+ logger_->log_debug("Bin %s for group %s offer size %d byte %d min_entry %d max_entry %d", uuid_str_, groupId_, queue_.size(), queued_data_size_, minEntries_, maxEntries_);
return true;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/libarchive/FocusArchiveEntry.cpp
----------------------------------------------------------------------
diff --git a/extensions/libarchive/FocusArchiveEntry.cpp b/extensions/libarchive/FocusArchiveEntry.cpp
index afcef8b..36b252c 100644
--- a/extensions/libarchive/FocusArchiveEntry.cpp
+++ b/extensions/libarchive/FocusArchiveEntry.cpp
@@ -90,13 +90,13 @@ void FocusArchiveEntry::onTrigger(core::ProcessContext *context, core::ProcessSe
for (auto &entryMetadata : archiveMetadata.entryMetadata) {
if (entryMetadata.entryType == AE_IFREG) {
- logger_->log_info("FocusArchiveEntry importing %s from %s", entryMetadata.entryName.c_str(), entryMetadata.tmpFileName.c_str());
+ logger_->log_info("FocusArchiveEntry importing %s from %s", entryMetadata.entryName, entryMetadata.tmpFileName);
session->import(entryMetadata.tmpFileName, flowFile, false, 0);
char stashKey[37];
uuid_t stashKeyUuid;
id_generator_->generate(stashKeyUuid);
uuid_unparse_lower(stashKeyUuid, stashKey);
- logger_->log_debug("FocusArchiveEntry generated stash key %s for entry %s", stashKey, entryMetadata.entryName.c_str());
+ logger_->log_debug("FocusArchiveEntry generated stash key %s for entry %s", stashKey, entryMetadata.entryName);
entryMetadata.stashKey.assign(stashKey);
if (entryMetadata.entryName == targetEntry) {
@@ -112,7 +112,7 @@ void FocusArchiveEntry::onTrigger(core::ProcessContext *context, core::ProcessSe
if (targetEntryStashKey != "") {
session->restore(targetEntryStashKey, flowFile);
} else {
- logger_->log_warn("FocusArchiveEntry failed to locate target entry: %s", targetEntry.c_str());
+ logger_->log_warn("FocusArchiveEntry failed to locate target entry: %s", targetEntry);
}
// Set new/updated lens stack to attribute
@@ -276,7 +276,7 @@ int64_t FocusArchiveEntry::ReadCallback::process(std::shared_ptr<io::BaseStream>
auto tmpFileName = file_man_->unique_file(true);
metadata.tmpFileName = tmpFileName;
metadata.entryType = entryType;
- logger_->log_info("FocusArchiveEntry extracting %s to: %s", entryName, tmpFileName.c_str());
+ logger_->log_info("FocusArchiveEntry extracting %s to: %s", entryName, tmpFileName);
auto fd = fopen(tmpFileName.c_str(), "w");
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/libarchive/MergeContent.cpp
----------------------------------------------------------------------
diff --git a/extensions/libarchive/MergeContent.cpp b/extensions/libarchive/MergeContent.cpp
index 91f542b..8bd3f79 100644
--- a/extensions/libarchive/MergeContent.cpp
+++ b/extensions/libarchive/MergeContent.cpp
@@ -128,8 +128,8 @@ void MergeContent::onSchedule(core::ProcessContext *context, core::ProcessSessio
if (mergeStratgey_ == MERGE_STRATEGY_DEFRAGMENT) {
binManager_.setFileCount(FRAGMENT_COUNT_ATTRIBUTE);
}
- logger_->log_info("Merge Content: Strategy [%s] Format [%s] Correlation Attribute [%s] Delimiter [%s]", mergeStratgey_, mergeFormat_, correlationAttributeName_, delimiterStratgey_);
- logger_->log_info("Merge Content: Footer [%s] Header [%s] Demarcator [%s] KeepPath [%d]", footer_, header_, demarcator_, keepPath_);
+ logger_->log_debug("Merge Content: Strategy [%s] Format [%s] Correlation Attribute [%s] Delimiter [%s]", mergeStratgey_, mergeFormat_, correlationAttributeName_, delimiterStratgey_);
+ logger_->log_debug("Merge Content: Footer [%s] Header [%s] Demarcator [%s] KeepPath [%d]", footer_, header_, demarcator_, keepPath_);
if (delimiterStratgey_ == DELIMITER_STRATEGY_FILENAME) {
if (!header_.empty()) {
this->headerContent_ = readContent(header_);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/libarchive/MergeContent.h
----------------------------------------------------------------------
diff --git a/extensions/libarchive/MergeContent.h b/extensions/libarchive/MergeContent.h
index d73eb71..cb6b469 100644
--- a/extensions/libarchive/MergeContent.h
+++ b/extensions/libarchive/MergeContent.h
@@ -220,7 +220,7 @@ public:
if (flow->getAttribute(BinFiles::TAR_PERMISSIONS_ATTRIBUTE, perm)) {
try {
permInt = std::stoi(perm);
- logger_->log_info("Merge Tar File %s permission %s", fileName, perm);
+ logger_->log_debug("Merge Tar File %s permission %s", fileName, perm);
archive_entry_set_perm(entry, (mode_t) permInt);
} catch (...) {
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/libarchive/UnfocusArchiveEntry.cpp
----------------------------------------------------------------------
diff --git a/extensions/libarchive/UnfocusArchiveEntry.cpp b/extensions/libarchive/UnfocusArchiveEntry.cpp
index 840ad89..e904946 100644
--- a/extensions/libarchive/UnfocusArchiveEntry.cpp
+++ b/extensions/libarchive/UnfocusArchiveEntry.cpp
@@ -212,7 +212,7 @@ int64_t UnfocusArchiveEntry::WriteCallback::process(std::shared_ptr<io::BaseStre
for (const auto &entryMetadata : _archiveMetadata->entryMetadata) {
entry = archive_entry_new();
- logger_->log_info("UnfocusArchiveEntry writing entry %s", entryMetadata.entryName.c_str());
+ logger_->log_info("UnfocusArchiveEntry writing entry %s", entryMetadata.entryName);
if (entryMetadata.entryType == AE_IFREG && entryMetadata.entrySize > 0) {
size_t stat_ok = stat(entryMetadata.tmpFileName.c_str(), &st);
@@ -230,7 +230,7 @@ int64_t UnfocusArchiveEntry::WriteCallback::process(std::shared_ptr<io::BaseStre
archive_entry_set_gid(entry, entryMetadata.entryGID);
archive_entry_set_mtime(entry, entryMetadata.entryMTime, entryMetadata.entryMTimeNsec);
- logger_->log_info("Writing %s with type %d, perms %d, size %d, uid %d, gid %d, mtime %d,%d", entryMetadata.entryName.c_str(), entryMetadata.entryType, entryMetadata.entryPerm,
+ logger_->log_info("Writing %s with type %d, perms %d, size %d, uid %d, gid %d, mtime %d,%d", entryMetadata.entryName, entryMetadata.entryType, entryMetadata.entryPerm,
entryMetadata.entrySize, entryMetadata.entryUID, entryMetadata.entryGID, entryMetadata.entryMTime, entryMetadata.entryMTimeNsec);
archive_write_header(outputArchive, entry);
@@ -239,7 +239,7 @@ int64_t UnfocusArchiveEntry::WriteCallback::process(std::shared_ptr<io::BaseStre
if (entryMetadata.entryType == AE_IFREG && entryMetadata.entrySize > 0) {
logger_->log_info("UnfocusArchiveEntry writing %d bytes of "
"data from tmp file %s to archive entry %s",
- st.st_size, entryMetadata.tmpFileName.c_str(), entryMetadata.entryName.c_str());
+ st.st_size, entryMetadata.tmpFileName, entryMetadata.entryName);
std::ifstream ifs(entryMetadata.tmpFileName, std::ifstream::in | std::ios::binary);
while (ifs.good()) {
@@ -249,7 +249,7 @@ int64_t UnfocusArchiveEntry::WriteCallback::process(std::shared_ptr<io::BaseStre
if (written < 0) {
logger_->log_error("UnfocusArchiveEntry failed to write data to "
"archive entry %s due to error: %s",
- entryMetadata.entryName.c_str(), archive_error_string(outputArchive));
+ entryMetadata.entryName, archive_error_string(outputArchive));
} else {
nlen += written;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/librdkafka/PublishKafka.cpp
----------------------------------------------------------------------
diff --git a/extensions/librdkafka/PublishKafka.cpp b/extensions/librdkafka/PublishKafka.cpp
index dc99a80..5f39e03 100644
--- a/extensions/librdkafka/PublishKafka.cpp
+++ b/extensions/librdkafka/PublishKafka.cpp
@@ -95,7 +95,7 @@ void PublishKafka::onSchedule(core::ProcessContext *context, core::ProcessSessio
if (context->getProperty(SeedBrokers.getName(), value) && !value.empty()) {
result = rd_kafka_conf_set(conf_, "bootstrap.servers", value.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: bootstrap.servers [%s]", value);
+ logger_->log_debug("PublishKafka: bootstrap.servers [%s]", value);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
@@ -103,21 +103,21 @@ void PublishKafka::onSchedule(core::ProcessContext *context, core::ProcessSessio
if (context->getProperty(MaxMessageSize.getName(), value) && !value.empty() && core::Property::StringToInt(value, valInt)) {
valueConf = std::to_string(valInt);
result = rd_kafka_conf_set(conf_, "message.max.bytes", valueConf.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: message.max.bytes [%s]", valueConf);
+ logger_->log_debug("PublishKafka: message.max.bytes [%s]", valueConf);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
value = "";
if (context->getProperty(ClientName.getName(), value) && !value.empty()) {
rd_kafka_conf_set(conf_, "client.id", value.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: client.id [%s]", value);
+ logger_->log_debug("PublishKafka: client.id [%s]", value);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
value = "";
if (context->getProperty(QueueBufferMaxMessage.getName(), value) && !value.empty()) {
rd_kafka_conf_set(conf_, "queue.buffering.max.messages", value.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: queue.buffering.max.messages [%s]", value);
+ logger_->log_debug("PublishKafka: queue.buffering.max.messages [%s]", value);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
@@ -126,7 +126,7 @@ void PublishKafka::onSchedule(core::ProcessContext *context, core::ProcessSessio
valInt = valInt/1024;
valueConf = std::to_string(valInt);
rd_kafka_conf_set(conf_, "queue.buffering.max.kbytes", valueConf.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: queue.buffering.max.kbytes [%s]", valueConf);
+ logger_->log_debug("PublishKafka: queue.buffering.max.kbytes [%s]", valueConf);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
@@ -134,7 +134,7 @@ void PublishKafka::onSchedule(core::ProcessContext *context, core::ProcessSessio
max_seg_size_ = ULLONG_MAX;
if (context->getProperty(MaxFlowSegSize.getName(), value) && !value.empty() && core::Property::StringToInt(value, valInt)) {
max_seg_size_ = valInt;
- logger_->log_info("PublishKafka: max flow segment size [%d]", max_seg_size_);
+ logger_->log_debug("PublishKafka: max flow segment size [%d]", max_seg_size_);
}
value = "";
if (context->getProperty(QueueBufferMaxTime.getName(), value) && !value.empty()) {
@@ -142,7 +142,7 @@ void PublishKafka::onSchedule(core::ProcessContext *context, core::ProcessSessio
if (core::Property::StringToTime(value, valInt, unit) && core::Property::ConvertTimeUnitToMS(valInt, unit, valInt)) {
valueConf = std::to_string(valInt);
rd_kafka_conf_set(conf_, "queue.buffering.max.ms", valueConf.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: queue.buffering.max.ms [%s]", valueConf);
+ logger_->log_debug("PublishKafka: queue.buffering.max.ms [%s]", valueConf);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
@@ -150,21 +150,21 @@ void PublishKafka::onSchedule(core::ProcessContext *context, core::ProcessSessio
value = "";
if (context->getProperty(BatchSize.getName(), value) && !value.empty()) {
rd_kafka_conf_set(conf_, "batch.num.messages", value.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: batch.num.messages [%s]", value);
+ logger_->log_debug("PublishKafka: batch.num.messages [%s]", value);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
value = "";
if (context->getProperty(CompressCodec.getName(), value) && !value.empty()) {
rd_kafka_conf_set(conf_, "compression.codec", value.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: compression.codec [%s]", value);
+ logger_->log_debug("PublishKafka: compression.codec [%s]", value);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
value = "";
if (context->getProperty(DeliveryGuarantee.getName(), value) && !value.empty()) {
rd_kafka_topic_conf_set(topic_conf_, "request.required.acks", value.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: request.required.acks [%s]", value);
+ logger_->log_debug("PublishKafka: request.required.acks [%s]", value);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
@@ -174,7 +174,7 @@ void PublishKafka::onSchedule(core::ProcessContext *context, core::ProcessSessio
if (core::Property::StringToTime(value, valInt, unit) && core::Property::ConvertTimeUnitToMS(valInt, unit, valInt)) {
valueConf = std::to_string(valInt);
rd_kafka_topic_conf_set(topic_conf_, "request.timeout.ms", valueConf.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: request.timeout.ms [%s]", valueConf);
+ logger_->log_debug("PublishKafka: request.timeout.ms [%s]", valueConf);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
@@ -183,35 +183,35 @@ void PublishKafka::onSchedule(core::ProcessContext *context, core::ProcessSessio
if (context->getProperty(SecurityProtocol.getName(), value) && !value.empty()) {
if (value == SECURITY_PROTOCOL_SSL) {
rd_kafka_conf_set(conf_, "security.protocol", value.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: security.protocol [%s]", value);
+ logger_->log_debug("PublishKafka: security.protocol [%s]", value);
if (result != RD_KAFKA_CONF_OK) {
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
} else {
value = "";
if (context->getProperty(SecurityCA.getName(), value) && !value.empty()) {
rd_kafka_conf_set(conf_, "ssl.ca.location", value.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: ssl.ca.location [%s]", value);
+ logger_->log_debug("PublishKafka: ssl.ca.location [%s]", value);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
value = "";
if (context->getProperty(SecurityCert.getName(), value) && !value.empty()) {
rd_kafka_conf_set(conf_, "ssl.certificate.location", value.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: ssl.certificate.location [%s]", value);
+ logger_->log_debug("PublishKafka: ssl.certificate.location [%s]", value);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
value = "";
if (context->getProperty(SecurityPrivateKey.getName(), value) && !value.empty()) {
rd_kafka_conf_set(conf_, "ssl.key.location", value.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: ssl.key.location [%s]", value);
+ logger_->log_debug("PublishKafka: ssl.key.location [%s]", value);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
value = "";
if (context->getProperty(SecurityPrivateKeyPassWord.getName(), value) && !value.empty()) {
rd_kafka_conf_set(conf_, "ssl.key.password", value.c_str(), errstr, sizeof(errstr));
- logger_->log_info("PublishKafka: ssl.key.password [%s]", value);
+ logger_->log_debug("PublishKafka: ssl.key.password [%s]", value);
if (result != RD_KAFKA_CONF_OK)
logger_->log_error("PublishKafka: configure error result [%s]", errstr);
}
@@ -221,9 +221,9 @@ void PublishKafka::onSchedule(core::ProcessContext *context, core::ProcessSessio
value = "";
if (context->getProperty(Topic.getName(), value) && !value.empty()) {
topic_ = value;
- logger_->log_info("PublishKafka: topic [%s]", topic_);
+ logger_->log_debug("PublishKafka: topic [%s]", topic_);
} else {
- logger_->log_info("PublishKafka: topic not configured");
+ logger_->log_warn("PublishKafka: topic not configured");
return;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/mqtt/AbstractMQTTProcessor.cpp
----------------------------------------------------------------------
diff --git a/extensions/mqtt/AbstractMQTTProcessor.cpp b/extensions/mqtt/AbstractMQTTProcessor.cpp
index 9ce052d..409b69f 100644
--- a/extensions/mqtt/AbstractMQTTProcessor.cpp
+++ b/extensions/mqtt/AbstractMQTTProcessor.cpp
@@ -70,39 +70,39 @@ void AbstractMQTTProcessor::onSchedule(core::ProcessContext *context, core::Proc
value = "";
if (context->getProperty(BrokerURL.getName(), value) && !value.empty()) {
uri_ = value;
- logger_->log_info("AbstractMQTTProcessor: BrokerURL [%s]", uri_);
+ logger_->log_debug("AbstractMQTTProcessor: BrokerURL [%s]", uri_);
}
value = "";
if (context->getProperty(ClientID.getName(), value) && !value.empty()) {
clientID_ = value;
- logger_->log_info("AbstractMQTTProcessor: ClientID [%s]", clientID_);
+ logger_->log_debug("AbstractMQTTProcessor: ClientID [%s]", clientID_);
}
value = "";
if (context->getProperty(Topic.getName(), value) && !value.empty()) {
topic_ = value;
- logger_->log_info("AbstractMQTTProcessor: Topic [%s]", topic_);
+ logger_->log_debug("AbstractMQTTProcessor: Topic [%s]", topic_);
}
value = "";
if (context->getProperty(UserName.getName(), value) && !value.empty()) {
userName_ = value;
- logger_->log_info("AbstractMQTTProcessor: UserName [%s]", userName_);
+ logger_->log_debug("AbstractMQTTProcessor: UserName [%s]", userName_);
}
value = "";
if (context->getProperty(PassWord.getName(), value) && !value.empty()) {
passWord_ = value;
- logger_->log_info("AbstractMQTTProcessor: PassWord [%s]", passWord_);
+ logger_->log_debug("AbstractMQTTProcessor: PassWord [%s]", passWord_);
}
value = "";
if (context->getProperty(CleanSession.getName(), value) && !value.empty() &&
org::apache::nifi::minifi::utils::StringUtils::StringToBool(value, cleanSession_)) {
- logger_->log_info("AbstractMQTTProcessor: CleanSession [%d]", cleanSession_);
+ logger_->log_debug("AbstractMQTTProcessor: CleanSession [%d]", cleanSession_);
}
value = "";
if (context->getProperty(KeepLiveInterval.getName(), value) && !value.empty()) {
core::TimeUnit unit;
if (core::Property::StringToTime(value, valInt, unit) && core::Property::ConvertTimeUnitToMS(valInt, unit, valInt)) {
keepAliveInterval_ = valInt/1000;
- logger_->log_info("AbstractMQTTProcessor: KeepLiveInterval [%ll]", keepAliveInterval_);
+ logger_->log_debug("AbstractMQTTProcessor: KeepLiveInterval [%ll]", keepAliveInterval_);
}
}
value = "";
@@ -110,14 +110,14 @@ void AbstractMQTTProcessor::onSchedule(core::ProcessContext *context, core::Proc
core::TimeUnit unit;
if (core::Property::StringToTime(value, valInt, unit) && core::Property::ConvertTimeUnitToMS(valInt, unit, valInt)) {
connectionTimeOut_ = valInt/1000;
- logger_->log_info("AbstractMQTTProcessor: ConnectionTimeOut [%ll]", connectionTimeOut_);
+ logger_->log_debug("AbstractMQTTProcessor: ConnectionTimeOut [%ll]", connectionTimeOut_);
}
}
value = "";
if (context->getProperty(QOS.getName(), value) && !value.empty() && (value == MQTT_QOS_0 || value == MQTT_QOS_1 || MQTT_QOS_2) &&
core::Property::StringToInt(value, valInt)) {
qos_ = valInt;
- logger_->log_info("AbstractMQTTProcessor: QOS [%ll]", qos_);
+ logger_->log_debug("AbstractMQTTProcessor: QOS [%ll]", qos_);
}
if (!client_) {
MQTTClient_create(&client_, uri_.c_str(), clientID_.c_str(), MQTTCLIENT_PERSISTENCE_NONE, NULL);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/mqtt/ConsumeMQTT.cpp
----------------------------------------------------------------------
diff --git a/extensions/mqtt/ConsumeMQTT.cpp b/extensions/mqtt/ConsumeMQTT.cpp
index 12de6bf..21cb79d 100644
--- a/extensions/mqtt/ConsumeMQTT.cpp
+++ b/extensions/mqtt/ConsumeMQTT.cpp
@@ -75,7 +75,7 @@ void ConsumeMQTT::onSchedule(core::ProcessContext *context, core::ProcessSession
value = "";
if (context->getProperty(MaxQueueSize.getName(), value) && !value.empty() && core::Property::StringToInt(value, valInt)) {
maxQueueSize_ = valInt;
- logger_->log_info("ConsumeMQTT: max queue size [%ll]", maxQueueSize_);
+ logger_->log_debug("ConsumeMQTT: max queue size [%ll]", maxQueueSize_);
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/mqtt/PublishMQTT.cpp
----------------------------------------------------------------------
diff --git a/extensions/mqtt/PublishMQTT.cpp b/extensions/mqtt/PublishMQTT.cpp
index 44ce298..411cc2d 100644
--- a/extensions/mqtt/PublishMQTT.cpp
+++ b/extensions/mqtt/PublishMQTT.cpp
@@ -67,11 +67,11 @@ void PublishMQTT::onSchedule(core::ProcessContext *context, core::ProcessSession
value = "";
if (context->getProperty(MaxFlowSegSize.getName(), value) && !value.empty() && core::Property::StringToInt(value, valInt)) {
max_seg_size_ = valInt;
- logger_->log_info("PublishMQTT: max flow segment size [%ll]", max_seg_size_);
+ logger_->log_debug("PublishMQTT: max flow segment size [%ll]", max_seg_size_);
}
value = "";
if (context->getProperty(Retain.getName(), value) && !value.empty() && org::apache::nifi::minifi::utils::StringUtils::StringToBool(value, retain_)) {
- logger_->log_info("PublishMQTT: Retain [%d]", retain_);
+ logger_->log_debug("PublishMQTT: Retain [%d]", retain_);
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/rocksdb-repos/DatabaseContentRepository.cpp
----------------------------------------------------------------------
diff --git a/extensions/rocksdb-repos/DatabaseContentRepository.cpp b/extensions/rocksdb-repos/DatabaseContentRepository.cpp
index 50f007f..40b2124 100644
--- a/extensions/rocksdb-repos/DatabaseContentRepository.cpp
+++ b/extensions/rocksdb-repos/DatabaseContentRepository.cpp
@@ -45,10 +45,10 @@ bool DatabaseContentRepository::initialize(const std::shared_ptr<minifi::Configu
options.max_successive_merges = 0;
rocksdb::Status status = rocksdb::DB::Open(options, directory_.c_str(), &db_);
if (status.ok()) {
- logger_->log_debug("NiFi Content DB Repository database open %s success", directory_.c_str());
+ logger_->log_debug("NiFi Content DB Repository database open %s success", directory_);
is_valid_ = true;
} else {
- logger_->log_error("NiFi Content DB Repository database open %s fail", directory_.c_str());
+ logger_->log_error("NiFi Content DB Repository database open %s fail", directory_);
is_valid_ = false;
}
return is_valid_;
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/rocksdb-repos/FlowFileRepository.cpp
----------------------------------------------------------------------
diff --git a/extensions/rocksdb-repos/FlowFileRepository.cpp b/extensions/rocksdb-repos/FlowFileRepository.cpp
index d5b403c..0d055ea 100644
--- a/extensions/rocksdb-repos/FlowFileRepository.cpp
+++ b/extensions/rocksdb-repos/FlowFileRepository.cpp
@@ -47,12 +47,12 @@ void FlowFileRepository::flush() {
if (eventRead->DeSerialize(reinterpret_cast<const uint8_t *>(value.data()), value.size())) {
purgeList.push_back(eventRead);
}
- logger_->log_info("Issuing batch delete, including %s, Content path %s", eventRead->getUUIDStr(), eventRead->getContentFullPath());
+ logger_->log_debug("Issuing batch delete, including %s, Content path %s", eventRead->getUUIDStr(), eventRead->getContentFullPath());
batch.Delete(key);
}
}
if (db_->Write(rocksdb::WriteOptions(), &batch).ok()) {
- logger_->log_info("Decrementing %u from a repo size of %u", decrement_total, repo_size_.load());
+ logger_->log_trace("Decrementing %u from a repo size of %u", decrement_total, repo_size_.load());
if (decrement_total > repo_size_.load()) {
repo_size_ = 0;
} else {
@@ -98,7 +98,7 @@ void FlowFileRepository::loadComponent(const std::shared_ptr<core::ContentReposi
std::string key = it->key().ToString();
repo_size_ += it->value().size();
if (eventRead->DeSerialize(reinterpret_cast<const uint8_t *>(it->value().data()), it->value().size())) {
- logger_->log_info("Found connection for %s, path %s ", eventRead->getConnectionUuid(), eventRead->getContentFullPath());
+ logger_->log_debug("Found connection for %s, path %s ", eventRead->getConnectionUuid(), eventRead->getContentFullPath());
auto search = connectionMap.find(eventRead->getConnectionUuid());
if (search != connectionMap.end()) {
// we find the connection for the persistent flowfile, create the flowfile and enqueue that
@@ -106,7 +106,7 @@ void FlowFileRepository::loadComponent(const std::shared_ptr<core::ContentReposi
eventRead->setStoredToRepository(true);
search->second->put(eventRead);
} else {
- logger_->log_info("Could not find connectinon for %s, path %s ", eventRead->getConnectionUuid(), eventRead->getContentFullPath());
+ logger_->log_warn("Could not find connectinon for %s, path %s ", eventRead->getConnectionUuid(), eventRead->getContentFullPath());
if (eventRead->getContentFullPath().length() > 0) {
if (nullptr != eventRead->getResourceClaim()) {
content_repo_->remove(eventRead->getResourceClaim());
@@ -121,7 +121,7 @@ void FlowFileRepository::loadComponent(const std::shared_ptr<core::ContentReposi
delete it;
for (auto eventId : purgeList) {
- logger_->log_info("Repository Repo %s Purge %s", name_.c_str(), eventId.first.c_str());
+ logger_->log_debug("Repository Repo %s Purge %s", name_, eventId.first);
if (Delete(eventId.first)) {
repo_size_ -= eventId.second;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/rocksdb-repos/FlowFileRepository.h
----------------------------------------------------------------------
diff --git a/extensions/rocksdb-repos/FlowFileRepository.h b/extensions/rocksdb-repos/FlowFileRepository.h
index a826bad..1bdd440 100644
--- a/extensions/rocksdb-repos/FlowFileRepository.h
+++ b/extensions/rocksdb-repos/FlowFileRepository.h
@@ -75,26 +75,26 @@ class FlowFileRepository : public core::Repository, public std::enable_shared_fr
if (configure->get(Configure::nifi_flowfile_repository_directory_default, value)) {
directory_ = value;
}
- logger_->log_info("NiFi FlowFile Repository Directory %s", directory_.c_str());
+ logger_->log_debug("NiFi FlowFile Repository Directory %s", directory_);
if (configure->get(Configure::nifi_flowfile_repository_max_storage_size, value)) {
Property::StringToInt(value, max_partition_bytes_);
}
- logger_->log_info("NiFi FlowFile Max Partition Bytes %d", max_partition_bytes_);
+ logger_->log_debug("NiFi FlowFile Max Partition Bytes %d", max_partition_bytes_);
if (configure->get(Configure::nifi_flowfile_repository_max_storage_time, value)) {
TimeUnit unit;
if (Property::StringToTime(value, max_partition_millis_, unit) && Property::ConvertTimeUnitToMS(max_partition_millis_, unit, max_partition_millis_)) {
}
}
- logger_->log_info("NiFi FlowFile Max Storage Time: [%d] ms", max_partition_millis_);
+ logger_->log_debug("NiFi FlowFile Max Storage Time: [%d] ms", max_partition_millis_);
rocksdb::Options options;
options.create_if_missing = true;
options.use_direct_io_for_flush_and_compaction = true;
options.use_direct_reads = true;
- rocksdb::Status status = rocksdb::DB::Open(options, directory_.c_str(), &db_);
+ rocksdb::Status status = rocksdb::DB::Open(options, directory_, &db_);
if (status.ok()) {
- logger_->log_info("NiFi FlowFile Repository database open %s success", directory_.c_str());
+ logger_->log_debug("NiFi FlowFile Repository database open %s success", directory_);
} else {
- logger_->log_error("NiFi FlowFile Repository database open %s fail", directory_.c_str());
+ logger_->log_error("NiFi FlowFile Repository database open %s fail", directory_);
return false;
}
return true;
@@ -148,7 +148,7 @@ class FlowFileRepository : public core::Repository, public std::enable_shared_fr
}
running_ = true;
thread_ = std::thread(&FlowFileRepository::run, shared_from_this());
- logger_->log_info("%s Repository Monitor Thread Start", getName());
+ logger_->log_debug("%s Repository Monitor Thread Start", getName());
}
private:
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/rocksdb-repos/ProvenanceRepository.cpp
----------------------------------------------------------------------
diff --git a/extensions/rocksdb-repos/ProvenanceRepository.cpp b/extensions/rocksdb-repos/ProvenanceRepository.cpp
index 6ff4056..bbd58fd 100644
--- a/extensions/rocksdb-repos/ProvenanceRepository.cpp
+++ b/extensions/rocksdb-repos/ProvenanceRepository.cpp
@@ -39,11 +39,11 @@ void ProvenanceRepository::flush() {
db_->Get(options, key, &value);
decrement_total += value.size();
batch.Delete(key);
- logger_->log_info("Removing %s", key);
+ logger_->log_debug("Removing %s", key);
}
}
if (db_->Write(rocksdb::WriteOptions(), &batch).ok()) {
- logger_->log_info("Decrementing %u from a repo size of %u", decrement_total, repo_size_.load());
+ logger_->log_debug("Decrementing %u from a repo size of %u", decrement_total, repo_size_.load());
if (decrement_total > repo_size_.load()) {
repo_size_ = 0;
} else {
@@ -71,7 +71,7 @@ void ProvenanceRepository::run() {
if ((curTime - eventTime) > (uint64_t)max_partition_millis_)
Delete(key);
} else {
- logger_->log_debug("NiFi Provenance retrieve event %s fail", key.c_str());
+ logger_->log_debug("NiFi Provenance retrieve event %s fail", key);
Delete(key);
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/rocksdb-repos/ProvenanceRepository.h
----------------------------------------------------------------------
diff --git a/extensions/rocksdb-repos/ProvenanceRepository.h b/extensions/rocksdb-repos/ProvenanceRepository.h
index 38d63e0..d325c24 100644
--- a/extensions/rocksdb-repos/ProvenanceRepository.h
+++ b/extensions/rocksdb-repos/ProvenanceRepository.h
@@ -71,7 +71,7 @@ class ProvenanceRepository : public core::Repository, public std::enable_shared_
return;
running_ = true;
thread_ = std::thread(&ProvenanceRepository::run, shared_from_this());
- logger_->log_info("%s Repository Monitor Thread Start", name_);
+ logger_->log_debug("%s Repository Monitor Thread Start", name_);
}
// initialize
@@ -80,26 +80,26 @@ class ProvenanceRepository : public core::Repository, public std::enable_shared_
if (config->get(Configure::nifi_provenance_repository_directory_default, value)) {
directory_ = value;
}
- logger_->log_info("NiFi Provenance Repository Directory %s", directory_.c_str());
+ logger_->log_debug("NiFi Provenance Repository Directory %s", directory_);
if (config->get(Configure::nifi_provenance_repository_max_storage_size, value)) {
core::Property::StringToInt(value, max_partition_bytes_);
}
- logger_->log_info("NiFi Provenance Max Partition Bytes %d", max_partition_bytes_);
+ logger_->log_debug("NiFi Provenance Max Partition Bytes %d", max_partition_bytes_);
if (config->get(Configure::nifi_provenance_repository_max_storage_time, value)) {
core::TimeUnit unit;
if (core::Property::StringToTime(value, max_partition_millis_, unit) && core::Property::ConvertTimeUnitToMS(max_partition_millis_, unit, max_partition_millis_)) {
}
}
- logger_->log_info("NiFi Provenance Max Storage Time: [%d] ms", max_partition_millis_);
+ logger_->log_debug("NiFi Provenance Max Storage Time: [%d] ms", max_partition_millis_);
rocksdb::Options options;
options.create_if_missing = true;
options.use_direct_io_for_flush_and_compaction = true;
options.use_direct_reads = true;
- rocksdb::Status status = rocksdb::DB::Open(options, directory_.c_str(), &db_);
+ rocksdb::Status status = rocksdb::DB::Open(options, directory_, &db_);
if (status.ok()) {
- logger_->log_info("NiFi Provenance Repository database open %s success", directory_.c_str());
+ logger_->log_debug("NiFi Provenance Repository database open %s success", directory_);
} else {
- logger_->log_error("NiFi Provenance Repository database open %s fail", directory_.c_str());
+ logger_->log_error("NiFi Provenance Repository database open %s fail", directory_);
return false;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/extensions/usb-camera/GetUSBCamera.cpp
----------------------------------------------------------------------
diff --git a/extensions/usb-camera/GetUSBCamera.cpp b/extensions/usb-camera/GetUSBCamera.cpp
index 761431f..09699bd 100644
--- a/extensions/usb-camera/GetUSBCamera.cpp
+++ b/extensions/usb-camera/GetUSBCamera.cpp
@@ -77,7 +77,7 @@ void GetUSBCamera::onFrame(uvc_frame_t *frame, void *ptr) {
try {
uvc_error_t ret;
- cb_data->logger->log_info("Got frame");
+ cb_data->logger->log_debug("Got frame");
ret = uvc_any2rgb(frame, cb_data->frame_buffer);
@@ -170,7 +170,7 @@ void GetUSBCamera::onSchedule(core::ProcessContext *context, core::ProcessSessio
return;
}
- logger_->log_info("UVC initialized");
+ logger_->log_debug("UVC initialized");
// Locate device
res = uvc_find_device(ctx_, &dev_, usb_vendor_id, usb_product_id, usb_serial_no);
@@ -215,7 +215,7 @@ void GetUSBCamera::onSchedule(core::ProcessContext *context, core::ProcessSessio
logger_->log_info("Skipping MJPEG frame formats");
default:
- logger_->log_info("Found unknown format");
+ logger_->log_warn("Found unknown format");
}
}
@@ -277,30 +277,30 @@ void GetUSBCamera::cleanupUvc() {
std::lock_guard<std::recursive_mutex> lock(*dev_access_mtx_);
if (frame_buffer_ != nullptr) {
- logger_->log_info("Deallocating frame buffer");
+ logger_->log_debug("Deallocating frame buffer");
uvc_free_frame(frame_buffer_);
}
if (devh_ != nullptr) {
- logger_->log_info("Stopping UVC streaming");
+ logger_->log_debug("Stopping UVC streaming");
uvc_stop_streaming(devh_);
- logger_->log_info("Closing UVC device handle");
+ logger_->log_debug("Closing UVC device handle");
uvc_close(devh_);
}
if (dev_ != nullptr) {
- logger_->log_info("Closing UVC device descriptor");
+ logger_->log_debug("Closing UVC device descriptor");
uvc_unref_device(dev_);
}
if (ctx_ != nullptr) {
- logger_->log_info("Closing UVC context");
+ logger_->log_debug("Closing UVC context");
uvc_exit(ctx_);
}
if (camera_thread_ != nullptr) {
camera_thread_->join();
- logger_->log_info("UVC thread ended");
+ logger_->log_debug("UVC thread ended");
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/include/FlowControlProtocol.h
----------------------------------------------------------------------
diff --git a/libminifi/include/FlowControlProtocol.h b/libminifi/include/FlowControlProtocol.h
index 56b1f62..72da2d3 100644
--- a/libminifi/include/FlowControlProtocol.h
+++ b/libminifi/include/FlowControlProtocol.h
@@ -169,7 +169,7 @@ class FlowControlProtocol {
if (configure->get(Configure::nifi_server_name, value)) {
_serverName = value;
- logger_->log_info("NiFi Server Name %s", _serverName.c_str());
+ 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: [%ll]", _serverPort);
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/include/core/controller/StandardControllerServiceProvider.h
----------------------------------------------------------------------
diff --git a/libminifi/include/core/controller/StandardControllerServiceProvider.h b/libminifi/include/core/controller/StandardControllerServiceProvider.h
index a6ca684..4d70757 100644
--- a/libminifi/include/core/controller/StandardControllerServiceProvider.h
+++ b/libminifi/include/core/controller/StandardControllerServiceProvider.h
@@ -117,7 +117,7 @@ class StandardControllerServiceProvider : public ControllerServiceProvider, publ
logger_->log_info("Enabling %s", service->getName());
agent_->enableControllerService(service);
} else {
- logger_->log_info("Could not enable %s", service->getName());
+ logger_->log_warn("Could not enable %s", service->getName());
}
}
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/include/core/repository/VolatileRepository.h
----------------------------------------------------------------------
diff --git a/libminifi/include/core/repository/VolatileRepository.h b/libminifi/include/core/repository/VolatileRepository.h
index 33fcf83..be23a0b 100644
--- a/libminifi/include/core/repository/VolatileRepository.h
+++ b/libminifi/include/core/repository/VolatileRepository.h
@@ -378,7 +378,7 @@ void VolatileRepository<T>::start() {
return;
running_ = true;
thread_ = std::thread(&VolatileRepository<T>::run, std::enable_shared_from_this<VolatileRepository<T>>::shared_from_this());
- logger_->log_info("%s Repository Monitor Thread Start", name_);
+ logger_->log_debug("%s Repository Monitor Thread Start", name_);
}
#if defined(__clang__)
#pragma clang diagnostic pop
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/include/processors/ListenSyslog.h
----------------------------------------------------------------------
diff --git a/libminifi/include/processors/ListenSyslog.h b/libminifi/include/processors/ListenSyslog.h
index 25acac9..f9dc678 100644
--- a/libminifi/include/processors/ListenSyslog.h
+++ b/libminifi/include/processors/ListenSyslog.h
@@ -91,7 +91,7 @@ class ListenSyslog : public core::Processor {
}
_clientSockets.clear();
if (_serverSocket > 0) {
- logger_->log_info("ListenSysLog Server socket %d close", _serverSocket);
+ logger_->log_debug("ListenSysLog Server socket %d close", _serverSocket);
close(_serverSocket);
_serverSocket = 0;
}
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/233d1d44/libminifi/src/Connection.cpp
----------------------------------------------------------------------
diff --git a/libminifi/src/Connection.cpp b/libminifi/src/Connection.cpp
index 5b9187f..9513f68 100644
--- a/libminifi/src/Connection.cpp
+++ b/libminifi/src/Connection.cpp
@@ -58,7 +58,7 @@ Connection::Connection(const std::shared_ptr<core::Repository> &flow_repository,
expired_duration_ = 0;
queued_data_size_ = 0;
- logger_->log_info("Connection %s created", name_.c_str());
+ logger_->log_debug("Connection %s created", name_);
}
bool Connection::isEmpty() {
@@ -122,7 +122,7 @@ std::shared_ptr<core::FlowFile> Connection::poll(std::set<std::shared_ptr<core::
if (getTimeMillis() > (item->getEntryDate() + expired_duration_)) {
// Flow record expired
expiredFlowRecords.insert(item);
- logger_->log_debug("Delete flow file UUID %s from connection %s, because it expired", item->getUUIDStr().c_str(), name_.c_str());
+ logger_->log_debug("Delete flow file UUID %s from connection %s, because it expired", item->getUUIDStr(), name_);
if (flow_repository_->Delete(item->getUUIDStr())) {
item->setStoredToRepository(false);
}
@@ -136,7 +136,7 @@ std::shared_ptr<core::FlowFile> Connection::poll(std::set<std::shared_ptr<core::
}
std::shared_ptr<Connectable> connectable = std::static_pointer_cast<Connectable>(shared_from_this());
item->setOriginalConnection(connectable);
- logger_->log_debug("Dequeue flow file UUID %s from connection %s", item->getUUIDStr().c_str(), name_.c_str());
+ logger_->log_debug("Dequeue flow file UUID %s from connection %s", item->getUUIDStr(), name_);
return item;
}
} else {
@@ -149,7 +149,7 @@ std::shared_ptr<core::FlowFile> Connection::poll(std::set<std::shared_ptr<core::
}
std::shared_ptr<Connectable> connectable = std::static_pointer_cast<Connectable>(shared_from_this());
item->setOriginalConnection(connectable);
- logger_->log_debug("Dequeue flow file UUID %s from connection %s", item->getUUIDStr().c_str(), name_.c_str());
+ logger_->log_debug("Dequeue flow file UUID %s from connection %s", item->getUUIDStr(), name_);
return item;
}
}
@@ -163,13 +163,13 @@ void Connection::drain() {
while (!queue_.empty()) {
std::shared_ptr<core::FlowFile> item = queue_.front();
queue_.pop();
- logger_->log_debug("Delete flow file UUID %s from connection %s, because it expired", item->getUUIDStr().c_str(), name_.c_str());
+ logger_->log_debug("Delete flow file UUID %s from connection %s, because it expired", item->getUUIDStr(), name_);
if (flow_repository_->Delete(item->getUUIDStr())) {
item->setStoredToRepository(false);
}
}
queued_data_size_ = 0;
- logger_->log_debug("Drain connection %s", name_.c_str());
+ logger_->log_debug("Drain connection %s", name_);
}
} /* namespace minifi */