You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by benqiu2016 <gi...@git.apache.org> on 2017/04/04 16:01:42 UTC
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
GitHub user benqiu2016 opened a pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74
Minifi 227
Thank you for submitting a contribution to Apache NiFi - MiNiFi C++.
In order to streamline the review of the contribution we ask you
to ensure the following steps have been taken:
### For all changes:
- [ ] Is there a JIRA ticket associated with this PR? Is it referenced
in the commit message?
- [ ] Does your PR title start with MINIFI-XXXX where XXXX is the JIRA number you are trying to resolve? Pay particular attention to the hyphen "-" character.
- [ ] Has your PR been rebased against the latest commit within the target branch (typically master)?
- [ ] Is your initial contribution a single, squashed commit?
### For code changes:
- [ ] If adding new dependencies to the code, are these dependencies licensed in a way that is compatible for inclusion under [ASF 2.0](http://www.apache.org/legal/resolved.html#category-a)?
- [ ] If applicable, have you updated the LICENSE file?
- [ ] If applicable, have you updated the NOTICE file?
### For documentation related changes:
- [ ] Have you ensured that format looks appropriate for the output in which it is rendered?
### Note:
Please ensure that once the PR is submitted, you check travis-ci for build issues and submit an update to your PR as soon as possible.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/benqiu2016/nifi-minifi-cpp MINIFI-227
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/nifi-minifi-cpp/pull/74.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #74
----
commit 1e1823ab583e0d19dec7a8e36d3aba1fbe5f806e
Author: Bin Qiu <be...@gmail.com>
Date: 2017-03-31T23:04:56Z
MINIFI-227: Provenance report
commit 8cc6906ff060b68585abef925cf13d0d0f376857
Author: Bin Qiu <be...@gmail.com>
Date: 2017-04-01T00:53:06Z
MINIFI-227: Provenance report
commit d639efa280a330fdd6d4764175efe8745cd810ce
Author: Bin Qiu <be...@gmail.com>
Date: 2017-04-04T15:56:01Z
MINIFI-227: Provenance report
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp issue #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on the issue:
https://github.com/apache/nifi-minifi-cpp/pull/74
@apiri i addressed your above review comments. Adding the new test case to support full site2site is not as easy as we thought, we need to support hand shake/CRC/two phase commit to make sure that end to end works.. Right now we only support S2S client in CPP, we can add S2S server in CPP. If we can add that, it means that we can run both client and server S2S in CPP test suite. We can address that via a different JIRA.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109955536
--- Diff: libminifi/src/Site2SiteClientProtocol.cpp ---
@@ -682,6 +682,7 @@ bool Site2SiteClientProtocol::receive(std::string transactionID,
bool Site2SiteClientProtocol::send(
std::string transactionID, DataPacket *packet, std::shared_ptr<FlowFileRecord> flowFile,
+ uint8_t *payload, int length,
--- End diff --
OK. will change that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r111843708
--- Diff: libminifi/include/core/Processor.h ---
@@ -239,12 +242,21 @@ class Processor : public Connectable, public ConfigurableComponent,
// Trigger the Processor even if the incoming connection is empty
std::atomic<bool> _triggerWhenEmpty;
- private:
+ //! obtainSite2SiteProtocol for use
+ std::shared_ptr<Site2SiteClientProtocol> obtainSite2SiteProtocol(std::string host, uint16_t sport, uuid_t portId);
--- End diff --
Open MINIFI-269 for support site2site server protocol
Open MINIFI-270 for the refactor of above
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109995160
--- Diff: libminifi/src/provenance/ProvenanceTaskReport.cpp ---
@@ -0,0 +1,221 @@
+/**
+ * @file ProvenanceTaskReport.cpp
+ * ProvenanceTaskReport class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <sys/time.h>
+#include <time.h>
+#include <sstream>
+#include <string.h>
+#include <iostream>
+
+#include "provenance/ProvenanceTaskReport.h"
+#include "../include/io/StreamFactory.h"
+#include "io/ClientSocket.h"
+#include "utils/TimeUtil.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "provenance/Provenance.h"
+#include "FlowController.h"
+
+#include "json/json.h"
+#include "json/writer.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance{
+
+const std::string ProvenanceTaskReport::ProcessorName("ProvenanceTaskReport");
+core::Property ProvenanceTaskReport::hostName("Host Name", "Remote Host Name.", "localhost");
+core::Property ProvenanceTaskReport::port("Port", "Remote Port", "9999");
+core::Property ProvenanceTaskReport::batchSize("Batch Size", "Specifies how many records to send in a single batch, at most.", "100");
+core::Relationship ProvenanceTaskReport::relation;
+
+void ProvenanceTaskReport::initialize()
+{
+ //! Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(hostName);
+ properties.insert(port);
+ properties.insert(batchSize);
+ setSupportedProperties(properties);
+ //! Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(relation);
+ setSupportedRelationships(relationships);
+}
+
--- End diff --
I don't see how potential change obviates the need to minimize duplicate code.
If I may be able to rephrase your response, am I correct in understanding that you mean: "I am copying and pasting code into a class because it may eventually change. Therefore, I don't wish to encapsulate this functionality elsewhere. As a result, change may make that copied code be different at some point?"
I want to make sure I understand you before I can respond.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109733328
--- Diff: libminifi/src/Site2SiteClientProtocol.cpp ---
@@ -1240,6 +1257,88 @@ void Site2SiteClientProtocol::transferFlowFiles(
return;
}
+void Site2SiteClientProtocol::transferBytes(core::ProcessContext *context, core::ProcessSession *session, uint8_t *payload, int length,
--- End diff --
vector<uint8_t> avoids requiring two arguments and the need to allocate and free memory elsewhere.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109954861
--- Diff: thirdparty/jsoncpp/devtools/batchbuild.py ---
@@ -0,0 +1,278 @@
+from __future__ import print_function
--- End diff --
OK.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109954528
--- Diff: libminifi/src/provenance/ProvenanceTaskReport.cpp ---
@@ -0,0 +1,221 @@
+/**
+ * @file ProvenanceTaskReport.cpp
+ * ProvenanceTaskReport class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <sys/time.h>
+#include <time.h>
+#include <sstream>
+#include <string.h>
+#include <iostream>
+
+#include "provenance/ProvenanceTaskReport.h"
+#include "../include/io/StreamFactory.h"
+#include "io/ClientSocket.h"
+#include "utils/TimeUtil.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "provenance/Provenance.h"
+#include "FlowController.h"
+
+#include "json/json.h"
+#include "json/writer.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance{
+
+const std::string ProvenanceTaskReport::ProcessorName("ProvenanceTaskReport");
+core::Property ProvenanceTaskReport::hostName("Host Name", "Remote Host Name.", "localhost");
+core::Property ProvenanceTaskReport::port("Port", "Remote Port", "9999");
+core::Property ProvenanceTaskReport::batchSize("Batch Size", "Specifies how many records to send in a single batch, at most.", "100");
+core::Relationship ProvenanceTaskReport::relation;
+
+void ProvenanceTaskReport::initialize()
+{
+ //! Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(hostName);
+ properties.insert(port);
+ properties.insert(batchSize);
+ setSupportedProperties(properties);
+ //! Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(relation);
+ setSupportedRelationships(relationships);
+}
+
--- End diff --
right now, we propose the Provenance Report property the same as RemoteProcess Group as hostName and Port, it may change. Also the way s2s protocol negotiation may change. For example. Remote Process Group may support round robin between different NiFi server instead of hardcode the hostName and port in the processor property. The decouple of these two provide more flexible and it make sense because they behave differently.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109757748
--- Diff: libminifi/src/provenance/ProvenanceTaskReport.cpp ---
@@ -0,0 +1,221 @@
+/**
+ * @file ProvenanceTaskReport.cpp
+ * ProvenanceTaskReport class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <sys/time.h>
+#include <time.h>
+#include <sstream>
+#include <string.h>
+#include <iostream>
+
+#include "provenance/ProvenanceTaskReport.h"
+#include "../include/io/StreamFactory.h"
+#include "io/ClientSocket.h"
+#include "utils/TimeUtil.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "provenance/Provenance.h"
+#include "FlowController.h"
+
+#include "json/json.h"
+#include "json/writer.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance{
+
+const std::string ProvenanceTaskReport::ProcessorName("ProvenanceTaskReport");
+core::Property ProvenanceTaskReport::hostName("Host Name", "Remote Host Name.", "localhost");
+core::Property ProvenanceTaskReport::port("Port", "Remote Port", "9999");
+core::Property ProvenanceTaskReport::batchSize("Batch Size", "Specifies how many records to send in a single batch, at most.", "100");
+core::Relationship ProvenanceTaskReport::relation;
+
+void ProvenanceTaskReport::initialize()
+{
+ //! Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(hostName);
+ properties.insert(port);
+ properties.insert(batchSize);
+ setSupportedProperties(properties);
+ //! Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(relation);
+ setSupportedRelationships(relationships);
+}
+
--- End diff --
we may want to add specific S2S protocol logic for shipping provenance report data. It make the code more flexible and decoupled.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109741638
--- Diff: libminifi/src/provenance/ProvenanceRepository.cpp ---
@@ -36,6 +36,7 @@ void ProvenanceRepository::run() {
uint64_t curTime = getTimeMillis();
uint64_t size = repoSize();
if (size >= purgeThreshold) {
+ std::lock_guard<std::mutex> lock(mutex_);
--- End diff --
What makes this necessary? is there a way to avoid locking? LevelDB is thread safe due to process level locking.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109758846
--- Diff: libminifi/src/provenance/ProvenanceRepository.cpp ---
@@ -36,6 +36,7 @@ void ProvenanceRepository::run() {
uint64_t curTime = getTimeMillis();
uint64_t size = repoSize();
if (size >= purgeThreshold) {
+ std::lock_guard<std::mutex> lock(mutex_);
--- End diff --
yes, levelDB is thread save, but when we purge the record in provenanceRepository task, i do not want to ProvenanceReportTask to pick the record and ship to NiFi.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109792275
--- Diff: libminifi/src/Site2SiteClientProtocol.cpp ---
@@ -682,6 +682,7 @@ bool Site2SiteClientProtocol::receive(std::string transactionID,
bool Site2SiteClientProtocol::send(
std::string transactionID, DataPacket *packet, std::shared_ptr<FlowFileRecord> flowFile,
+ uint8_t *payload, int length,
--- End diff --
As someone reading this that does not make sense. Why not make DataPacket hold the "packet"?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/nifi-minifi-cpp/pull/74
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by apiri <gi...@git.apache.org>.
Github user apiri commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r111136647
--- Diff: README.md ---
@@ -286,6 +287,26 @@ Additionally, users can utilize the MiNiFi Toolkit Converter (version 0.0.1 - sc
if you do not want to enable client certificate base authorization
nifi.security.need.ClientAuth=false
+### Provenance Report
+
+ Add ProvenanceTaskReport processor to config.yml
+ Port UUID of the ProvenanceTaskReport is the remote NiFi input port UUID
+
+ - name: ProvenanceTaskReport
+ id: 471deef6-2a6e-4a7d-912a-81cc17e3a209
+ class: org.apache.nifi.processors.standard.ProvenanceTaskReport
+ max concurrent tasks: 1
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 1 sec
+ penalization period: 30 sec
+ yield period: 10 sec
+ run duration nanos: 0
+ auto-terminated relationships list:
+ Properties:
+ Port: 10001
+ Host Name: localhost
+ Port UUID: 471deef6-2a6e-4a7d-912a-81cc17e3a204
--- End diff --
This is typically accomplished via port uuid
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109761443
--- Diff: libminifi/src/Site2SiteClientProtocol.cpp ---
@@ -682,6 +682,7 @@ bool Site2SiteClientProtocol::receive(std::string transactionID,
bool Site2SiteClientProtocol::send(
std::string transactionID, DataPacket *packet, std::shared_ptr<FlowFileRecord> flowFile,
+ uint8_t *payload, int length,
--- End diff --
send can take flowfile or raw bytes, DataPacket is more like a hold for the meta info like transaction, etc.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp issue #74: Minifi 227
Posted by apiri <gi...@git.apache.org>.
Github user apiri commented on the issue:
https://github.com/apache/nifi-minifi-cpp/pull/74
reviewing changes
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp issue #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on the issue:
https://github.com/apache/nifi-minifi-cpp/pull/74
@benqiu2016 Did you make that JIRA and make it a blocker? It should be the next most important task.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r110435863
--- Diff: libminifi/src/provenance/ProvenanceTaskReport.cpp ---
@@ -0,0 +1,191 @@
+/**
+ * @file ProvenanceTaskReport.cpp
+ * ProvenanceTaskReport class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <string>
+#include <memory>
+#include <sstream>
+#include <iostream>
+
+#include "provenance/ProvenanceTaskReport.h"
+#include "../include/io/StreamFactory.h"
+#include "io/ClientSocket.h"
+#include "utils/TimeUtil.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "provenance/Provenance.h"
+#include "FlowController.h"
+
+#include "json/json.h"
+#include "json/writer.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance {
+
+core::Property ProvenanceTaskReport::hostName("Host Name", "Remote Host Name.",
+ "localhost");
+core::Property ProvenanceTaskReport::port("Port", "Remote Port", "9999");
+core::Property ProvenanceTaskReport::batchSize("Batch Size",
+ "Specifies how many records to send in a single batch, at most.", "100");
+core::Property ProvenanceTaskReport::portUUID("Port UUID",
+ "Specifies remote NiFi Port UUID.", "");
+core::Relationship ProvenanceTaskReport::relation;
+const char *ProvenanceTaskReport::ProvenanceAppStr = "MiNiFi Flow";
+
+void ProvenanceTaskReport::initialize() {
+ //! Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(hostName);
+ properties.insert(port);
+ properties.insert(batchSize);
+ properties.insert(portUUID);
+ setSupportedProperties(properties);
+ //! Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(relation);
+ setSupportedRelationships(relationships);
+}
+
+void ProvenanceTaskReport::onTrigger(core::ProcessContext *context,
+ core::ProcessSession *session) {
+ std::string value;
+ int64_t lvalue;
+ std::string host = "";
+ uint16_t sport = 0;
+
+ if (context->getProperty(hostName.getName(), value)) {
+ host = value;
+ }
+ if (context->getProperty(port.getName(), value)
+ && core::Property::StringToInt(value, lvalue)) {
+ sport = (uint16_t) lvalue;
+ }
+ if (context->getProperty(portUUID.getName(), value)) {
+ uuid_parse(value.c_str(), protocol_uuid_);
+ }
+
+ std::shared_ptr<Site2SiteClientProtocol> protocol_ =
+ this->obtainSite2SiteProtocol(host, sport, protocol_uuid_);
+
+ if (!protocol_) {
+ context->yield();
+ return;
+ }
+
+ if (!protocol_->bootstrap()) {
+ // bootstrap the client protocol if needeed
+ context->yield();
+ std::shared_ptr<Processor> processor = std::static_pointer_cast < Processor
+ > (context->getProcessorNode().getProcessor());
+ logger_->log_error("Site2Site bootstrap failed yield period %d peer ",
+ processor->getYieldPeriodMsec());
+ returnSite2SiteProtocol(protocol_);
+ return;
+ }
+
+ int64_t batch = 100;
+
+ if (context->getProperty(batchSize.getName(), value)
+ && core::Property::StringToInt(value, lvalue)) {
+ batch = lvalue;
+ }
+
+ std::vector < std::shared_ptr < ProvenanceEventRecord >> records;
+ std::shared_ptr<ProvenanceRepository> repo = std::static_pointer_cast
+ < ProvenanceRepository > (context->getProvenanceRepository());
+
+ repo->getProvenanceRecord(records, batch);
+
+ if (records.size() <= 0) {
+ returnSite2SiteProtocol(protocol_);
+ return;
+ }
+
+ Json::Value array;
+ for (auto record : records) {
+ Json::Value recordJson;
+ Json::Value updatedAttributesJson;
+ Json::Value parentUuidJson;
+ Json::Value childUuidJson;
+ recordJson["eventId"] = record->getEventId().c_str();
+ recordJson["eventType"] =
+ ProvenanceEventRecord::ProvenanceEventTypeStr[record->getEventType()];
+ recordJson["timestampMillis"] = record->getEventTime();
+ recordJson["durationMillis"] = record->getEventDuration();
+ recordJson["lineageStart"] = record->getlineageStartDate();
+ recordJson["details"] = record->getDetails().c_str();
+ recordJson["componentId"] = record->getComponentId().c_str();
+ recordJson["componentType"] = record->getComponentType().c_str();
+ recordJson["entityId"] = record->getFlowFileUuid().c_str();
+ recordJson["entityType"] = "org.apache.nifi.flowfile.FlowFile";
+ recordJson["entitySize"] = record->getFileSize();
+ recordJson["entityOffset"] = record->getFileOffset();
+
+ for (auto attr : record->getAttributes()) {
+ updatedAttributesJson[attr.first] = attr.second;
+ }
+ recordJson["updatedAttributes"] = updatedAttributesJson;
+
+ for (auto parentUUID : record->getParentUuids()) {
+ parentUuidJson.append(parentUUID.c_str());
+ }
+ recordJson["parentIds"] = parentUuidJson;
+
+ for (auto childUUID : record->getChildrenUuids()) {
+ childUuidJson.append(childUUID.c_str());
+ }
+ recordJson["childIds"] = childUuidJson;
+ recordJson["transitUri"] = record->getTransitUri().c_str();
+ recordJson["remoteIdentifier"] =
+ record->getSourceSystemFlowFileIdentifier().c_str();
+ recordJson["alternateIdentifier"] =
+ record->getAlternateIdentifierUri().c_str();
+ recordJson["application"] = ProvenanceAppStr;
+ array.append(recordJson);
+ }
+
+ Json::StyledWriter writer;
+ std::string jsonStr = writer.write(array);
+
+ try {
+ std::map < std::string, std::string > attributes;
+ protocol_->transferString(context, session, jsonStr, attributes);
+ } catch (...) {
+ // if transfer bytes failed, return instead of purge the provenance records
+ returnSite2SiteProtocol(protocol_);
--- End diff --
The master based RPG only returns a protocol upon success. Should we be using the protocol objects if the underlying transport potentially fails here?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r110006513
--- Diff: libminifi/src/Site2SiteClientProtocol.cpp ---
@@ -1240,6 +1257,88 @@ void Site2SiteClientProtocol::transferFlowFiles(
return;
}
+void Site2SiteClientProtocol::transferBytes(core::ProcessContext *context, core::ProcessSession *session, uint8_t *payload, int length,
--- End diff --
OK. will change to pass std::string.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp issue #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on the issue:
https://github.com/apache/nifi-minifi-cpp/pull/74
@phrocker it looks like the travis for MAC is OK.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109759205
--- Diff: thirdparty/jsoncpp/devtools/batchbuild.py ---
@@ -0,0 +1,278 @@
+from __future__ import print_function
--- End diff --
the tests source code are there because it was from the drop for jsoncpp. but i disable build and run the same in the minifi build for the jasoncpp cmake.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp issue #74: Minifi 227
Posted by apiri <gi...@git.apache.org>.
Github user apiri commented on the issue:
https://github.com/apache/nifi-minifi-cpp/pull/74
changes looked good. was able to verify functionality. licensing looks good. will get this merged
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by apiri <gi...@git.apache.org>.
Github user apiri commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r111129786
--- Diff: README.md ---
@@ -286,6 +287,26 @@ Additionally, users can utilize the MiNiFi Toolkit Converter (version 0.0.1 - sc
if you do not want to enable client certificate base authorization
nifi.security.need.ClientAuth=false
+### Provenance Report
+
+ Add ProvenanceTaskReport processor to config.yml
+ Port UUID of the ProvenanceTaskReport is the remote NiFi input port UUID
+
+ - name: ProvenanceTaskReport
+ id: 471deef6-2a6e-4a7d-912a-81cc17e3a209
+ class: org.apache.nifi.processors.standard.ProvenanceTaskReport
--- End diff --
Should namespace this to org.apache.nifi.reporting
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109791553
--- Diff: thirdparty/jsoncpp/devtools/batchbuild.py ---
@@ -0,0 +1,278 @@
+from __future__ import print_function
--- End diff --
Remove the code if it is unnecessary.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109760197
--- Diff: libminifi/src/Site2SiteClientProtocol.cpp ---
@@ -1240,6 +1257,88 @@ void Site2SiteClientProtocol::transferFlowFiles(
return;
}
+void Site2SiteClientProtocol::transferBytes(core::ProcessContext *context, core::ProcessSession *session, uint8_t *payload, int length,
--- End diff --
but we need to copy the payload to vector<uint8_t>, it is overhead
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109792191
--- Diff: libminifi/src/provenance/ProvenanceTaskReport.cpp ---
@@ -0,0 +1,221 @@
+/**
+ * @file ProvenanceTaskReport.cpp
+ * ProvenanceTaskReport class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <sys/time.h>
+#include <time.h>
+#include <sstream>
+#include <string.h>
+#include <iostream>
+
+#include "provenance/ProvenanceTaskReport.h"
+#include "../include/io/StreamFactory.h"
+#include "io/ClientSocket.h"
+#include "utils/TimeUtil.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "provenance/Provenance.h"
+#include "FlowController.h"
+
+#include "json/json.h"
+#include "json/writer.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance{
+
+const std::string ProvenanceTaskReport::ProcessorName("ProvenanceTaskReport");
+core::Property ProvenanceTaskReport::hostName("Host Name", "Remote Host Name.", "localhost");
+core::Property ProvenanceTaskReport::port("Port", "Remote Port", "9999");
+core::Property ProvenanceTaskReport::batchSize("Batch Size", "Specifies how many records to send in a single batch, at most.", "100");
+core::Relationship ProvenanceTaskReport::relation;
+
+void ProvenanceTaskReport::initialize()
+{
+ //! Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(hostName);
+ properties.insert(port);
+ properties.insert(batchSize);
+ setSupportedProperties(properties);
+ //! Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(relation);
+ setSupportedRelationships(relationships);
+}
+
+std::unique_ptr<Site2SiteClientProtocol> ProvenanceTaskReport::getNextProtocol()
+{
+ std::lock_guard<std::mutex> protocol_lock_(protocol_mutex_);
+ if (available_protocols_.empty())
+ return nullptr;
+ std::unique_ptr<Site2SiteClientProtocol> return_pointer = std::move(available_protocols_.top());
+ available_protocols_.pop();
+ return std::move(return_pointer);
+}
+
+void ProvenanceTaskReport::returnProtocol(
+ std::unique_ptr<Site2SiteClientProtocol> return_protocol)
+{
+ std::lock_guard<std::mutex> protocol_lock_(protocol_mutex_);
+ available_protocols_.push(std::move(return_protocol));
+}
+
+void ProvenanceTaskReport::onTrigger(core::ProcessContext *context, core::ProcessSession *session)
+{
+ std::string value;
+ int64_t lvalue;
+
+ std::unique_ptr<Site2SiteClientProtocol> protocol_ = getNextProtocol();
+
+ if (protocol_ == nullptr)
+ {
+ protocol_ = std::unique_ptr<Site2SiteClientProtocol>(
+ new Site2SiteClientProtocol(0));
+ protocol_->setPortId(protocol_uuid_);
+
+ std::string host = "";
+ uint16_t sport = 0;
+
+ if (context->getProperty(hostName.getName(), value)) {
+ host = value;
+ }
+ if (context->getProperty(port.getName(), value)
+ && core::Property::StringToInt(value, lvalue)) {
+ sport = (uint16_t) lvalue;
+ }
+ std::unique_ptr<org::apache::nifi::minifi::io::DataStream> str =
+ std::unique_ptr<org::apache::nifi::minifi::io::DataStream>(
+ org::apache::nifi::minifi::io::StreamFactory::getInstance()
+ ->createSocket(host, sport));
+
+ std::unique_ptr<Site2SitePeer> peer_ = std::unique_ptr<Site2SitePeer>(
+ new Site2SitePeer(std::move(str), host, sport));
+
+ protocol_->setPeer(std::move(peer_));
+ }
+
+ if (!protocol_->bootstrap())
+ {
+ // bootstrap the client protocol if needeed
+ context->yield();
+ std::shared_ptr<Processor> processor = std::static_pointer_cast<Processor>(
+ context->getProcessorNode().getProcessor());
+ logger_->log_error("Site2Site bootstrap failed yield period %d peer ",
+ processor->getYieldPeriodMsec());
+ return;
+ }
+
+ int batch = 100;
+
+ if (context->getProperty(batchSize.getName(), value) && core::Property::StringToInt(value, lvalue))
+ {
+ batch = (int) lvalue;
+ }
+
+ std::vector<std::shared_ptr<ProvenanceEventRecord>> records;
+ std::shared_ptr<ProvenanceRepository> repo = std::static_pointer_cast<ProvenanceRepository> (context->getProvenanceRepository());
+
+ repo->getProvenanceRecord(records, batch);
+
+ if (records.size() <= 0)
+ {
+ returnProtocol(std::move(protocol_));
+ return;
+ }
+
+ Json::Value array;
+ for (auto record : records)
+ {
+ Json::Value recordJson;
+ Json::Value updatedAttributesJson;
+ Json::Value parentUuidJson;
+ Json::Value childUuidJson;
+ recordJson["eventId"] = record->getEventId().c_str();
+ recordJson["eventType"] = ProvenanceEventRecord::ProvenanceEventTypeStr[record->getEventType()];
+ recordJson["timestampMillis"] = record->getEventTime();
+ recordJson["durationMillis"] = record->getEventDuration();
+ recordJson["lineageStart"] = record->getlineageStartDate();
+ recordJson["details"] = record->getDetails().c_str();
+ recordJson["componentId"] = record->getComponentId().c_str();
+ recordJson["componentType"] = record->getComponentType().c_str();
+ recordJson["entityId"] = record->getFlowFileUuid().c_str();
+ recordJson["entityType"] = "org.apache.nifi.flowfile.FlowFile";
+ recordJson["entitySize"] = record->getFileSize();
+ recordJson["entityOffset"] = record->getFileOffset();
+
+ for (auto attr : record->getAttributes())
+ {
+ updatedAttributesJson[attr.first] = attr.second;
+ }
+ recordJson["updatedAttributes"] = updatedAttributesJson;
+
+ for (auto parentUUID : record->getParentUuids())
+ {
+ parentUuidJson.append(parentUUID.c_str());
+ }
+ recordJson["parentIds"] = parentUuidJson;
+
+ for (auto childUUID : record->getChildrenUuids())
+ {
+ childUuidJson.append(childUUID.c_str());
+ }
+ recordJson["childIds"] = childUuidJson;
+ recordJson["transitUri"] = record->getTransitUri().c_str();
+ recordJson["remoteIdentifier"] = record->getSourceSystemFlowFileIdentifier().c_str();
+ recordJson["alternateIdentifier"] = record->getAlternateIdentifierUri().c_str();
+ recordJson["application"] = "MiNiFi Flow";
+ array.append(recordJson);
+ }
+
+ Json::StyledWriter writer;
+ std::string jsonStr = writer.write(array);
+ uint8_t *payload = (uint8_t *) jsonStr.c_str();
+ int length = jsonStr.length();
+
+ try
+ {
+ std::map<std::string, std::string> attributes;
+ protocol_->transferBytes(context, session, payload, length, attributes);
+ }
+ catch (...)
+ {
+ // if transfer bytes failed, return instead of purge the provenance records
+ returnProtocol(std::move(protocol_));
+ return;
+ }
+
+ // we transfer the record, purge the record from DB
+ repo->purgeProvenanceRecord(records);
+
+ returnProtocol(std::move(protocol_));
+
+ return;
--- End diff --
I'm sorry I don't follow. an empty return on a void isn't necessary. It's unnecessary code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109958322
--- Diff: libminifi/src/provenance/ProvenanceRepository.cpp ---
@@ -36,6 +36,7 @@ void ProvenanceRepository::run() {
uint64_t curTime = getTimeMillis();
uint64_t size = repoSize();
if (size >= purgeThreshold) {
+ std::lock_guard<std::mutex> lock(mutex_);
--- End diff --
OK
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109742909
--- Diff: libminifi/include/provenance/ProvenanceTaskReport.h ---
@@ -0,0 +1,88 @@
+/**
+ * @file ProvenanceTaskReport.h
+ * ProvenanceTaskReport class declaration
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#ifndef __PROVENANCE_TASK_REPORT_H__
+#define __PROVENANCE_TASK_REPORT_H__
+
+#include <mutex>
+#include <memory>
+#include <stack>
+#include "FlowFileRecord.h"
+#include "core/Processor.h"
+#include "core/ProcessSession.h"
+#include "Site2SiteClientProtocol.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance {
+
+//! ProvenanceTaskReport Class
+class ProvenanceTaskReport: public core::Processor {
--- End diff --
Would a more applicable name be ProvenanceTaskReporter?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109726099
--- Diff: thirdparty/jsoncpp/devtools/batchbuild.py ---
@@ -0,0 +1,278 @@
+from __future__ import print_function
--- End diff --
Are the tests or devtools or tests needed for jsoncpp?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r110004694
--- Diff: libminifi/src/Site2SiteClientProtocol.cpp ---
@@ -1240,6 +1257,88 @@ void Site2SiteClientProtocol::transferFlowFiles(
return;
}
+void Site2SiteClientProtocol::transferBytes(core::ProcessContext *context, core::ProcessSession *session, uint8_t *payload, int length,
--- End diff --
how can i do a std::move from a string to a std::vector<uint8_t> without copy?
http://stackoverflow.com/questions/10445042/stdmove-between-stdstring-and-stdvectorunsigned-char
pass the raw byte looks OK to me and make the code much simple in this case instead of convert from one type to another.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r110004494
--- Diff: libminifi/src/provenance/ProvenanceTaskReport.cpp ---
@@ -0,0 +1,221 @@
+/**
+ * @file ProvenanceTaskReport.cpp
+ * ProvenanceTaskReport class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <sys/time.h>
+#include <time.h>
+#include <sstream>
+#include <string.h>
+#include <iostream>
+
+#include "provenance/ProvenanceTaskReport.h"
+#include "../include/io/StreamFactory.h"
+#include "io/ClientSocket.h"
+#include "utils/TimeUtil.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "provenance/Provenance.h"
+#include "FlowController.h"
+
+#include "json/json.h"
+#include "json/writer.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance{
+
+const std::string ProvenanceTaskReport::ProcessorName("ProvenanceTaskReport");
+core::Property ProvenanceTaskReport::hostName("Host Name", "Remote Host Name.", "localhost");
+core::Property ProvenanceTaskReport::port("Port", "Remote Port", "9999");
+core::Property ProvenanceTaskReport::batchSize("Batch Size", "Specifies how many records to send in a single batch, at most.", "100");
+core::Relationship ProvenanceTaskReport::relation;
+
+void ProvenanceTaskReport::initialize()
+{
+ //! Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(hostName);
+ properties.insert(port);
+ properties.insert(batchSize);
+ setSupportedProperties(properties);
+ //! Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(relation);
+ setSupportedRelationships(relationships);
+}
+
--- End diff --
what i mean is that they are different processor and it make sense to make the code path differently. Certainly we can extract that returnProtocol and getNextProtocol to a parent class that both ProvenanceTaskReport and RemoteProcessGroup can inherit from if it is what you mean by duplicate code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109956092
--- Diff: libminifi/src/Site2SiteClientProtocol.cpp ---
@@ -1240,6 +1257,88 @@ void Site2SiteClientProtocol::transferFlowFiles(
return;
}
+void Site2SiteClientProtocol::transferBytes(core::ProcessContext *context, core::ProcessSession *session, uint8_t *payload, int length,
--- End diff --
Can we move a byte array to vector<uint8_t>?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by apiri <gi...@git.apache.org>.
Github user apiri commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r111136804
--- Diff: README.md ---
@@ -286,6 +287,26 @@ Additionally, users can utilize the MiNiFi Toolkit Converter (version 0.0.1 - sc
if you do not want to enable client certificate base authorization
nifi.security.need.ClientAuth=false
+### Provenance Report
--- End diff --
This is handled as a discrete component type as opposed to a processor.
https://github.com/apache/nifi-minifi/blob/master/minifi-bootstrap/src/test/resources/MINIFI-216/config.yml#L97
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by apiri <gi...@git.apache.org>.
Github user apiri commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r111129669
--- Diff: README.md ---
@@ -53,6 +53,7 @@ Perspectives of the role of MiNiFi should be from the perspective of the agent a
* ListenSyslog
* ExecuteProcess
* AppendHostInfo
+ * ProvenanceTaskReport
--- End diff --
We should make this SiteToSiteProvenanceReportingTask for consistency between instance types.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp issue #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on the issue:
https://github.com/apache/nifi-minifi-cpp/pull/74
Hey @benqiu2016, the master build (https://travis-ci.org/apache/nifi-minifi-cpp/builds/219032189 ) works for both Linux and Mac. I'm uncertain what you meant by pasting the error from your Travis output, but I'll be happy to take a stab. It looks like you pasted the Linux output, which fails because of build properties in Linux. My guess this is is in your CMake, perhaps, but I can't be certain without pulling it down. I'll take a look at that tomorrow as I'm currently running a Linux build and it succeeds from master.
The Mac build failed due to linter errors, which was a recent addition. This is easy to solve by typing make linter and resolving the output. Let me know if I can help with that.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109758278
--- Diff: libminifi/src/provenance/ProvenanceTaskReport.cpp ---
@@ -0,0 +1,221 @@
+/**
+ * @file ProvenanceTaskReport.cpp
+ * ProvenanceTaskReport class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <sys/time.h>
+#include <time.h>
+#include <sstream>
+#include <string.h>
+#include <iostream>
+
+#include "provenance/ProvenanceTaskReport.h"
+#include "../include/io/StreamFactory.h"
+#include "io/ClientSocket.h"
+#include "utils/TimeUtil.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "provenance/Provenance.h"
+#include "FlowController.h"
+
+#include "json/json.h"
+#include "json/writer.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance{
+
+const std::string ProvenanceTaskReport::ProcessorName("ProvenanceTaskReport");
+core::Property ProvenanceTaskReport::hostName("Host Name", "Remote Host Name.", "localhost");
+core::Property ProvenanceTaskReport::port("Port", "Remote Port", "9999");
+core::Property ProvenanceTaskReport::batchSize("Batch Size", "Specifies how many records to send in a single batch, at most.", "100");
+core::Relationship ProvenanceTaskReport::relation;
+
+void ProvenanceTaskReport::initialize()
+{
+ //! Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(hostName);
+ properties.insert(port);
+ properties.insert(batchSize);
+ setSupportedProperties(properties);
+ //! Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(relation);
+ setSupportedRelationships(relationships);
+}
+
+std::unique_ptr<Site2SiteClientProtocol> ProvenanceTaskReport::getNextProtocol()
+{
+ std::lock_guard<std::mutex> protocol_lock_(protocol_mutex_);
+ if (available_protocols_.empty())
+ return nullptr;
+ std::unique_ptr<Site2SiteClientProtocol> return_pointer = std::move(available_protocols_.top());
+ available_protocols_.pop();
+ return std::move(return_pointer);
+}
+
+void ProvenanceTaskReport::returnProtocol(
+ std::unique_ptr<Site2SiteClientProtocol> return_protocol)
+{
+ std::lock_guard<std::mutex> protocol_lock_(protocol_mutex_);
+ available_protocols_.push(std::move(return_protocol));
+}
+
+void ProvenanceTaskReport::onTrigger(core::ProcessContext *context, core::ProcessSession *session)
+{
+ std::string value;
+ int64_t lvalue;
+
+ std::unique_ptr<Site2SiteClientProtocol> protocol_ = getNextProtocol();
+
+ if (protocol_ == nullptr)
+ {
+ protocol_ = std::unique_ptr<Site2SiteClientProtocol>(
+ new Site2SiteClientProtocol(0));
+ protocol_->setPortId(protocol_uuid_);
+
+ std::string host = "";
+ uint16_t sport = 0;
+
+ if (context->getProperty(hostName.getName(), value)) {
+ host = value;
+ }
+ if (context->getProperty(port.getName(), value)
+ && core::Property::StringToInt(value, lvalue)) {
+ sport = (uint16_t) lvalue;
+ }
+ std::unique_ptr<org::apache::nifi::minifi::io::DataStream> str =
+ std::unique_ptr<org::apache::nifi::minifi::io::DataStream>(
+ org::apache::nifi::minifi::io::StreamFactory::getInstance()
+ ->createSocket(host, sport));
+
+ std::unique_ptr<Site2SitePeer> peer_ = std::unique_ptr<Site2SitePeer>(
+ new Site2SitePeer(std::move(str), host, sport));
+
+ protocol_->setPeer(std::move(peer_));
+ }
+
+ if (!protocol_->bootstrap())
+ {
+ // bootstrap the client protocol if needeed
+ context->yield();
+ std::shared_ptr<Processor> processor = std::static_pointer_cast<Processor>(
+ context->getProcessorNode().getProcessor());
+ logger_->log_error("Site2Site bootstrap failed yield period %d peer ",
+ processor->getYieldPeriodMsec());
+ return;
+ }
+
+ int batch = 100;
+
+ if (context->getProperty(batchSize.getName(), value) && core::Property::StringToInt(value, lvalue))
+ {
+ batch = (int) lvalue;
+ }
+
+ std::vector<std::shared_ptr<ProvenanceEventRecord>> records;
+ std::shared_ptr<ProvenanceRepository> repo = std::static_pointer_cast<ProvenanceRepository> (context->getProvenanceRepository());
+
+ repo->getProvenanceRecord(records, batch);
+
+ if (records.size() <= 0)
+ {
+ returnProtocol(std::move(protocol_));
+ return;
+ }
+
+ Json::Value array;
+ for (auto record : records)
+ {
+ Json::Value recordJson;
+ Json::Value updatedAttributesJson;
+ Json::Value parentUuidJson;
+ Json::Value childUuidJson;
+ recordJson["eventId"] = record->getEventId().c_str();
+ recordJson["eventType"] = ProvenanceEventRecord::ProvenanceEventTypeStr[record->getEventType()];
+ recordJson["timestampMillis"] = record->getEventTime();
+ recordJson["durationMillis"] = record->getEventDuration();
+ recordJson["lineageStart"] = record->getlineageStartDate();
+ recordJson["details"] = record->getDetails().c_str();
+ recordJson["componentId"] = record->getComponentId().c_str();
+ recordJson["componentType"] = record->getComponentType().c_str();
+ recordJson["entityId"] = record->getFlowFileUuid().c_str();
+ recordJson["entityType"] = "org.apache.nifi.flowfile.FlowFile";
+ recordJson["entitySize"] = record->getFileSize();
+ recordJson["entityOffset"] = record->getFileOffset();
+
+ for (auto attr : record->getAttributes())
+ {
+ updatedAttributesJson[attr.first] = attr.second;
+ }
+ recordJson["updatedAttributes"] = updatedAttributesJson;
+
+ for (auto parentUUID : record->getParentUuids())
+ {
+ parentUuidJson.append(parentUUID.c_str());
+ }
+ recordJson["parentIds"] = parentUuidJson;
+
+ for (auto childUUID : record->getChildrenUuids())
+ {
+ childUuidJson.append(childUUID.c_str());
+ }
+ recordJson["childIds"] = childUuidJson;
+ recordJson["transitUri"] = record->getTransitUri().c_str();
+ recordJson["remoteIdentifier"] = record->getSourceSystemFlowFileIdentifier().c_str();
+ recordJson["alternateIdentifier"] = record->getAlternateIdentifierUri().c_str();
+ recordJson["application"] = "MiNiFi Flow";
+ array.append(recordJson);
+ }
+
+ Json::StyledWriter writer;
+ std::string jsonStr = writer.write(array);
+ uint8_t *payload = (uint8_t *) jsonStr.c_str();
+ int length = jsonStr.length();
+
+ try
+ {
+ std::map<std::string, std::string> attributes;
+ protocol_->transferBytes(context, session, payload, length, attributes);
+ }
+ catch (...)
+ {
+ // if transfer bytes failed, return instead of purge the provenance records
+ returnProtocol(std::move(protocol_));
+ return;
+ }
+
+ // we transfer the record, purge the record from DB
+ repo->purgeProvenanceRecord(records);
+
+ returnProtocol(std::move(protocol_));
+
+ return;
--- End diff --
i want to return the same so that it can be reused in next trigger.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109725610
--- Diff: libminifi/src/provenance/ProvenanceTaskReport.cpp ---
@@ -0,0 +1,221 @@
+/**
+ * @file ProvenanceTaskReport.cpp
+ * ProvenanceTaskReport class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <sys/time.h>
+#include <time.h>
+#include <sstream>
+#include <string.h>
+#include <iostream>
+
+#include "provenance/ProvenanceTaskReport.h"
+#include "../include/io/StreamFactory.h"
+#include "io/ClientSocket.h"
+#include "utils/TimeUtil.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "provenance/Provenance.h"
+#include "FlowController.h"
+
+#include "json/json.h"
+#include "json/writer.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance{
+
+const std::string ProvenanceTaskReport::ProcessorName("ProvenanceTaskReport");
+core::Property ProvenanceTaskReport::hostName("Host Name", "Remote Host Name.", "localhost");
+core::Property ProvenanceTaskReport::port("Port", "Remote Port", "9999");
+core::Property ProvenanceTaskReport::batchSize("Batch Size", "Specifies how many records to send in a single batch, at most.", "100");
+core::Relationship ProvenanceTaskReport::relation;
+
+void ProvenanceTaskReport::initialize()
+{
+ //! Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(hostName);
+ properties.insert(port);
+ properties.insert(batchSize);
+ setSupportedProperties(properties);
+ //! Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(relation);
+ setSupportedRelationships(relationships);
+}
+
--- End diff --
This is identical to code in RemoteProcessGroup, so this should likely be abstracted elsewhere.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109745089
--- Diff: libminifi/src/provenance/ProvenanceTaskReport.cpp ---
@@ -0,0 +1,221 @@
+/**
+ * @file ProvenanceTaskReport.cpp
+ * ProvenanceTaskReport class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <sys/time.h>
+#include <time.h>
+#include <sstream>
+#include <string.h>
+#include <iostream>
+
+#include "provenance/ProvenanceTaskReport.h"
+#include "../include/io/StreamFactory.h"
+#include "io/ClientSocket.h"
+#include "utils/TimeUtil.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "provenance/Provenance.h"
+#include "FlowController.h"
+
+#include "json/json.h"
+#include "json/writer.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance{
+
+const std::string ProvenanceTaskReport::ProcessorName("ProvenanceTaskReport");
+core::Property ProvenanceTaskReport::hostName("Host Name", "Remote Host Name.", "localhost");
+core::Property ProvenanceTaskReport::port("Port", "Remote Port", "9999");
+core::Property ProvenanceTaskReport::batchSize("Batch Size", "Specifies how many records to send in a single batch, at most.", "100");
+core::Relationship ProvenanceTaskReport::relation;
+
+void ProvenanceTaskReport::initialize()
+{
+ //! Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(hostName);
+ properties.insert(port);
+ properties.insert(batchSize);
+ setSupportedProperties(properties);
+ //! Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(relation);
+ setSupportedRelationships(relationships);
+}
+
+std::unique_ptr<Site2SiteClientProtocol> ProvenanceTaskReport::getNextProtocol()
+{
+ std::lock_guard<std::mutex> protocol_lock_(protocol_mutex_);
+ if (available_protocols_.empty())
+ return nullptr;
+ std::unique_ptr<Site2SiteClientProtocol> return_pointer = std::move(available_protocols_.top());
+ available_protocols_.pop();
+ return std::move(return_pointer);
+}
+
+void ProvenanceTaskReport::returnProtocol(
+ std::unique_ptr<Site2SiteClientProtocol> return_protocol)
+{
+ std::lock_guard<std::mutex> protocol_lock_(protocol_mutex_);
+ available_protocols_.push(std::move(return_protocol));
+}
+
+void ProvenanceTaskReport::onTrigger(core::ProcessContext *context, core::ProcessSession *session)
+{
+ std::string value;
+ int64_t lvalue;
+
+ std::unique_ptr<Site2SiteClientProtocol> protocol_ = getNextProtocol();
+
+ if (protocol_ == nullptr)
+ {
+ protocol_ = std::unique_ptr<Site2SiteClientProtocol>(
+ new Site2SiteClientProtocol(0));
+ protocol_->setPortId(protocol_uuid_);
+
+ std::string host = "";
+ uint16_t sport = 0;
+
+ if (context->getProperty(hostName.getName(), value)) {
+ host = value;
+ }
+ if (context->getProperty(port.getName(), value)
+ && core::Property::StringToInt(value, lvalue)) {
+ sport = (uint16_t) lvalue;
+ }
+ std::unique_ptr<org::apache::nifi::minifi::io::DataStream> str =
+ std::unique_ptr<org::apache::nifi::minifi::io::DataStream>(
+ org::apache::nifi::minifi::io::StreamFactory::getInstance()
+ ->createSocket(host, sport));
+
+ std::unique_ptr<Site2SitePeer> peer_ = std::unique_ptr<Site2SitePeer>(
+ new Site2SitePeer(std::move(str), host, sport));
+
+ protocol_->setPeer(std::move(peer_));
+ }
+
+ if (!protocol_->bootstrap())
+ {
+ // bootstrap the client protocol if needeed
+ context->yield();
+ std::shared_ptr<Processor> processor = std::static_pointer_cast<Processor>(
+ context->getProcessorNode().getProcessor());
+ logger_->log_error("Site2Site bootstrap failed yield period %d peer ",
+ processor->getYieldPeriodMsec());
+ return;
+ }
+
+ int batch = 100;
+
+ if (context->getProperty(batchSize.getName(), value) && core::Property::StringToInt(value, lvalue))
+ {
+ batch = (int) lvalue;
+ }
+
+ std::vector<std::shared_ptr<ProvenanceEventRecord>> records;
+ std::shared_ptr<ProvenanceRepository> repo = std::static_pointer_cast<ProvenanceRepository> (context->getProvenanceRepository());
+
+ repo->getProvenanceRecord(records, batch);
+
+ if (records.size() <= 0)
+ {
+ returnProtocol(std::move(protocol_));
+ return;
+ }
+
+ Json::Value array;
+ for (auto record : records)
+ {
+ Json::Value recordJson;
+ Json::Value updatedAttributesJson;
+ Json::Value parentUuidJson;
+ Json::Value childUuidJson;
+ recordJson["eventId"] = record->getEventId().c_str();
+ recordJson["eventType"] = ProvenanceEventRecord::ProvenanceEventTypeStr[record->getEventType()];
+ recordJson["timestampMillis"] = record->getEventTime();
+ recordJson["durationMillis"] = record->getEventDuration();
+ recordJson["lineageStart"] = record->getlineageStartDate();
+ recordJson["details"] = record->getDetails().c_str();
+ recordJson["componentId"] = record->getComponentId().c_str();
+ recordJson["componentType"] = record->getComponentType().c_str();
+ recordJson["entityId"] = record->getFlowFileUuid().c_str();
+ recordJson["entityType"] = "org.apache.nifi.flowfile.FlowFile";
+ recordJson["entitySize"] = record->getFileSize();
+ recordJson["entityOffset"] = record->getFileOffset();
+
+ for (auto attr : record->getAttributes())
+ {
+ updatedAttributesJson[attr.first] = attr.second;
+ }
+ recordJson["updatedAttributes"] = updatedAttributesJson;
+
+ for (auto parentUUID : record->getParentUuids())
+ {
+ parentUuidJson.append(parentUUID.c_str());
+ }
+ recordJson["parentIds"] = parentUuidJson;
+
+ for (auto childUUID : record->getChildrenUuids())
+ {
+ childUuidJson.append(childUUID.c_str());
+ }
+ recordJson["childIds"] = childUuidJson;
+ recordJson["transitUri"] = record->getTransitUri().c_str();
+ recordJson["remoteIdentifier"] = record->getSourceSystemFlowFileIdentifier().c_str();
+ recordJson["alternateIdentifier"] = record->getAlternateIdentifierUri().c_str();
+ recordJson["application"] = "MiNiFi Flow";
+ array.append(recordJson);
+ }
+
+ Json::StyledWriter writer;
+ std::string jsonStr = writer.write(array);
+ uint8_t *payload = (uint8_t *) jsonStr.c_str();
+ int length = jsonStr.length();
+
+ try
+ {
+ std::map<std::string, std::string> attributes;
+ protocol_->transferBytes(context, session, payload, length, attributes);
+ }
+ catch (...)
+ {
+ // if transfer bytes failed, return instead of purge the provenance records
+ returnProtocol(std::move(protocol_));
+ return;
+ }
+
+ // we transfer the record, purge the record from DB
+ repo->purgeProvenanceRecord(records);
+
+ returnProtocol(std::move(protocol_));
+
+ return;
--- End diff --
this isn't necessary.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109792063
--- Diff: libminifi/src/provenance/ProvenanceRepository.cpp ---
@@ -36,6 +36,7 @@ void ProvenanceRepository::run() {
uint64_t curTime = getTimeMillis();
uint64_t size = repoSize();
if (size >= purgeThreshold) {
+ std::lock_guard<std::mutex> lock(mutex_);
--- End diff --
If LevelDB is thread safe and is being picked up that means at that moment in the thread which picks it it is not eligible to be purged and is eligible to be sent. That means you can atomically remove said item and thread A will never get it to send OR lock all access. If you need locking here above LevelDB we should change the access pattern.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp issue #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on the issue:
https://github.com/apache/nifi-minifi-cpp/pull/74
@phrocker
97%] Building C object thirdparty/civetweb-1.9.1/src/CMakeFiles/c-executable.dir/main.c.o
Linking C executable civetweb
/usr/include/x86_64-linux-gnu/bits/stdio2.h: In function \u2018send_file_data\u2019:
/usr/include/x86_64-linux-gnu/bits/stdio2.h:290:2: error: call to \u2018__fread_chk_warn\u2019 declared with attribute warning: fread called with bigger size * nmemb than length of destination buffer [-Werror]
return __fread_chk (__ptr, __bos0 (__ptr), __size, __n, __stream);
^
lto1: all warnings being treated as errors
lto-wrapper: /usr/bin/gcc returned 1 exit status
/usr/bin/ld: lto-wrapper failed
collect2: error: ld returned 1 exit status
make[2]: *** [thirdparty/civetweb-1.9.1/src/civetweb] Error 1
make[1]: *** [thirdparty/civetweb-1.9.1/src/CMakeFiles/c-executable.dir/all] Error 2
make: *** [all] Error 2
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by apiri <gi...@git.apache.org>.
Github user apiri commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r111130615
--- Diff: libminifi/include/provenance/ProvenanceTaskReport.h ---
@@ -0,0 +1,88 @@
+/**
+ * @file ProvenanceTaskReport.h
+ * ProvenanceTaskReport class declaration
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#ifndef __PROVENANCE_TASK_REPORT_H__
+#define __PROVENANCE_TASK_REPORT_H__
+
+#include <mutex>
+#include <memory>
+#include <stack>
+#include "FlowFileRecord.h"
+#include "core/Processor.h"
+#include "core/ProcessSession.h"
+#include "Site2SiteClientProtocol.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance {
+
+//! ProvenanceTaskReport Class
+class ProvenanceTaskReport: public core::Processor {
--- End diff --
would also be up for a rename, preferably SiteToSiteProvenanceReportingTask
would also like to see this namespaced to org.apache.nifi.reporting
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by apiri <gi...@git.apache.org>.
Github user apiri commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r111133055
--- Diff: libminifi/include/provenance/ProvenanceTaskReport.h ---
@@ -0,0 +1,91 @@
+/**
+ * @file ProvenanceTaskReport.h
+ * ProvenanceTaskReport class declaration
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#ifndef __PROVENANCE_TASK_REPORT_H__
+#define __PROVENANCE_TASK_REPORT_H__
+
+#include <mutex>
+#include <memory>
+#include <stack>
+#include "FlowFileRecord.h"
+#include "core/Processor.h"
+#include "core/ProcessSession.h"
+#include "Site2SiteClientProtocol.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance {
+
+//! ProvenanceTaskReport Class
+class ProvenanceTaskReport: public core::Processor {
+public:
+ //! Constructor
+ /*!
+ * Create a new processor
+ */
+ ProvenanceTaskReport(std::string name, uuid_t uuid = NULL) :
+ core::Processor(name, uuid) {
+ logger_ = logging::Logger::getLogger();
+ if (uuid)
+ uuid_copy(protocol_uuid_,uuid);
+ this->setTriggerWhenEmpty(true);
+ }
+ //! Destructor
+ virtual ~ProvenanceTaskReport() {
+
+ }
+ //! Processor Name
+ static constexpr char const* ProcessorName = "ProvenanceTaskReport";
+ //! Supported Properties
+ static core::Property hostName;
+ static core::Property port;
+ static core::Property batchSize;
+ static core::Property portUUID;
+ //! Supported Relationships
+ static core::Relationship relation;
+ static const char *ProvenanceAppStr;
+public:
+ //! Get provenance jason report
+ void getJasonReport(core::ProcessContext *context,
--- End diff --
can we update this to JsonReport?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r110003365
--- Diff: libminifi/src/Site2SiteClientProtocol.cpp ---
@@ -1240,6 +1257,88 @@ void Site2SiteClientProtocol::transferFlowFiles(
return;
}
+void Site2SiteClientProtocol::transferBytes(core::ProcessContext *context, core::ProcessSession *session, uint8_t *payload, int length,
--- End diff --
I can see the confusion. I was referring to a desire to minimize pointer usage and more strongly type functions within a public access.
This we are trying to avoid:
*Having arbitrarily large sized arrays. payload comes in with a ptr and length comes in as a value well above what the NiFi instance could support.
*What happens to NiFI if length is accidentally negative? You convert a -1 to uint64_t meaning that we would attempt to send a value well outside of our process control.
* You have a conditional in send in transferBytes, both of which are public. what would happen if someone uses the protocol to call send instead of transferBytes with a payload and non null flowFile? The payload would be ignored. This is not ideal since send/transferbytes are don't differentiate each other very well.
What inherent need is there to have a generic function that has few safeguards? Further, this function was intended to be used for sending strings. Why not use std::string instead of uint8t*,int ? your calls write the size with length without any checks, whereas the usage of std::vector OR std::string make this unnecessary and allow us to be more defensive.
Why aren't you using writeUTF ? Do you have the link to the respective java code for the serialization?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109792168
--- Diff: libminifi/src/Site2SiteClientProtocol.cpp ---
@@ -1240,6 +1257,88 @@ void Site2SiteClientProtocol::transferFlowFiles(
return;
}
+void Site2SiteClientProtocol::transferBytes(core::ProcessContext *context, core::ProcessSession *session, uint8_t *payload, int length,
--- End diff --
No @benqiu2016 that is why std::move(...) exists.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109954793
--- Diff: libminifi/src/provenance/ProvenanceTaskReport.cpp ---
@@ -0,0 +1,221 @@
+/**
+ * @file ProvenanceTaskReport.cpp
+ * ProvenanceTaskReport class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <sys/time.h>
+#include <time.h>
+#include <sstream>
+#include <string.h>
+#include <iostream>
+
+#include "provenance/ProvenanceTaskReport.h"
+#include "../include/io/StreamFactory.h"
+#include "io/ClientSocket.h"
+#include "utils/TimeUtil.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "provenance/Provenance.h"
+#include "FlowController.h"
+
+#include "json/json.h"
+#include "json/writer.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance{
+
+const std::string ProvenanceTaskReport::ProcessorName("ProvenanceTaskReport");
+core::Property ProvenanceTaskReport::hostName("Host Name", "Remote Host Name.", "localhost");
+core::Property ProvenanceTaskReport::port("Port", "Remote Port", "9999");
+core::Property ProvenanceTaskReport::batchSize("Batch Size", "Specifies how many records to send in a single batch, at most.", "100");
+core::Relationship ProvenanceTaskReport::relation;
+
+void ProvenanceTaskReport::initialize()
+{
+ //! Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(hostName);
+ properties.insert(port);
+ properties.insert(batchSize);
+ setSupportedProperties(properties);
+ //! Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(relation);
+ setSupportedRelationships(relationships);
+}
+
+std::unique_ptr<Site2SiteClientProtocol> ProvenanceTaskReport::getNextProtocol()
+{
+ std::lock_guard<std::mutex> protocol_lock_(protocol_mutex_);
+ if (available_protocols_.empty())
+ return nullptr;
+ std::unique_ptr<Site2SiteClientProtocol> return_pointer = std::move(available_protocols_.top());
+ available_protocols_.pop();
+ return std::move(return_pointer);
+}
+
+void ProvenanceTaskReport::returnProtocol(
+ std::unique_ptr<Site2SiteClientProtocol> return_protocol)
+{
+ std::lock_guard<std::mutex> protocol_lock_(protocol_mutex_);
+ available_protocols_.push(std::move(return_protocol));
+}
+
+void ProvenanceTaskReport::onTrigger(core::ProcessContext *context, core::ProcessSession *session)
+{
+ std::string value;
+ int64_t lvalue;
+
+ std::unique_ptr<Site2SiteClientProtocol> protocol_ = getNextProtocol();
+
+ if (protocol_ == nullptr)
+ {
+ protocol_ = std::unique_ptr<Site2SiteClientProtocol>(
+ new Site2SiteClientProtocol(0));
+ protocol_->setPortId(protocol_uuid_);
+
+ std::string host = "";
+ uint16_t sport = 0;
+
+ if (context->getProperty(hostName.getName(), value)) {
+ host = value;
+ }
+ if (context->getProperty(port.getName(), value)
+ && core::Property::StringToInt(value, lvalue)) {
+ sport = (uint16_t) lvalue;
+ }
+ std::unique_ptr<org::apache::nifi::minifi::io::DataStream> str =
+ std::unique_ptr<org::apache::nifi::minifi::io::DataStream>(
+ org::apache::nifi::minifi::io::StreamFactory::getInstance()
+ ->createSocket(host, sport));
+
+ std::unique_ptr<Site2SitePeer> peer_ = std::unique_ptr<Site2SitePeer>(
+ new Site2SitePeer(std::move(str), host, sport));
+
+ protocol_->setPeer(std::move(peer_));
+ }
+
+ if (!protocol_->bootstrap())
+ {
+ // bootstrap the client protocol if needeed
+ context->yield();
+ std::shared_ptr<Processor> processor = std::static_pointer_cast<Processor>(
+ context->getProcessorNode().getProcessor());
+ logger_->log_error("Site2Site bootstrap failed yield period %d peer ",
+ processor->getYieldPeriodMsec());
+ return;
+ }
+
+ int batch = 100;
+
+ if (context->getProperty(batchSize.getName(), value) && core::Property::StringToInt(value, lvalue))
+ {
+ batch = (int) lvalue;
+ }
+
+ std::vector<std::shared_ptr<ProvenanceEventRecord>> records;
+ std::shared_ptr<ProvenanceRepository> repo = std::static_pointer_cast<ProvenanceRepository> (context->getProvenanceRepository());
+
+ repo->getProvenanceRecord(records, batch);
+
+ if (records.size() <= 0)
+ {
+ returnProtocol(std::move(protocol_));
+ return;
+ }
+
+ Json::Value array;
+ for (auto record : records)
+ {
+ Json::Value recordJson;
+ Json::Value updatedAttributesJson;
+ Json::Value parentUuidJson;
+ Json::Value childUuidJson;
+ recordJson["eventId"] = record->getEventId().c_str();
+ recordJson["eventType"] = ProvenanceEventRecord::ProvenanceEventTypeStr[record->getEventType()];
+ recordJson["timestampMillis"] = record->getEventTime();
+ recordJson["durationMillis"] = record->getEventDuration();
+ recordJson["lineageStart"] = record->getlineageStartDate();
+ recordJson["details"] = record->getDetails().c_str();
+ recordJson["componentId"] = record->getComponentId().c_str();
+ recordJson["componentType"] = record->getComponentType().c_str();
+ recordJson["entityId"] = record->getFlowFileUuid().c_str();
+ recordJson["entityType"] = "org.apache.nifi.flowfile.FlowFile";
+ recordJson["entitySize"] = record->getFileSize();
+ recordJson["entityOffset"] = record->getFileOffset();
+
+ for (auto attr : record->getAttributes())
+ {
+ updatedAttributesJson[attr.first] = attr.second;
+ }
+ recordJson["updatedAttributes"] = updatedAttributesJson;
+
+ for (auto parentUUID : record->getParentUuids())
+ {
+ parentUuidJson.append(parentUUID.c_str());
+ }
+ recordJson["parentIds"] = parentUuidJson;
+
+ for (auto childUUID : record->getChildrenUuids())
+ {
+ childUuidJson.append(childUUID.c_str());
+ }
+ recordJson["childIds"] = childUuidJson;
+ recordJson["transitUri"] = record->getTransitUri().c_str();
+ recordJson["remoteIdentifier"] = record->getSourceSystemFlowFileIdentifier().c_str();
+ recordJson["alternateIdentifier"] = record->getAlternateIdentifierUri().c_str();
+ recordJson["application"] = "MiNiFi Flow";
+ array.append(recordJson);
+ }
+
+ Json::StyledWriter writer;
+ std::string jsonStr = writer.write(array);
+ uint8_t *payload = (uint8_t *) jsonStr.c_str();
+ int length = jsonStr.length();
+
+ try
+ {
+ std::map<std::string, std::string> attributes;
+ protocol_->transferBytes(context, session, payload, length, attributes);
+ }
+ catch (...)
+ {
+ // if transfer bytes failed, return instead of purge the provenance records
+ returnProtocol(std::move(protocol_));
+ return;
+ }
+
+ // we transfer the record, purge the record from DB
+ repo->purgeProvenanceRecord(records);
+
+ returnProtocol(std::move(protocol_));
+
+ return;
--- End diff --
sorry, i will remove the return. i thought you mean returnProtocol.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109742049
--- Diff: libminifi/src/provenance/ProvenanceTaskReport.cpp ---
@@ -0,0 +1,221 @@
+/**
+ * @file ProvenanceTaskReport.cpp
+ * ProvenanceTaskReport class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <sys/time.h>
+#include <time.h>
+#include <sstream>
+#include <string.h>
+#include <iostream>
+
+#include "provenance/ProvenanceTaskReport.h"
+#include "../include/io/StreamFactory.h"
+#include "io/ClientSocket.h"
+#include "utils/TimeUtil.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "provenance/Provenance.h"
+#include "FlowController.h"
+
+#include "json/json.h"
+#include "json/writer.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance{
+
+const std::string ProvenanceTaskReport::ProcessorName("ProvenanceTaskReport");
+core::Property ProvenanceTaskReport::hostName("Host Name", "Remote Host Name.", "localhost");
+core::Property ProvenanceTaskReport::port("Port", "Remote Port", "9999");
+core::Property ProvenanceTaskReport::batchSize("Batch Size", "Specifies how many records to send in a single batch, at most.", "100");
+core::Relationship ProvenanceTaskReport::relation;
+
+void ProvenanceTaskReport::initialize()
+{
+ //! Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(hostName);
+ properties.insert(port);
+ properties.insert(batchSize);
+ setSupportedProperties(properties);
+ //! Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(relation);
+ setSupportedRelationships(relationships);
+}
+
+std::unique_ptr<Site2SiteClientProtocol> ProvenanceTaskReport::getNextProtocol()
+{
+ std::lock_guard<std::mutex> protocol_lock_(protocol_mutex_);
+ if (available_protocols_.empty())
+ return nullptr;
+ std::unique_ptr<Site2SiteClientProtocol> return_pointer = std::move(available_protocols_.top());
+ available_protocols_.pop();
+ return std::move(return_pointer);
+}
+
+void ProvenanceTaskReport::returnProtocol(
+ std::unique_ptr<Site2SiteClientProtocol> return_protocol)
+{
+ std::lock_guard<std::mutex> protocol_lock_(protocol_mutex_);
+ available_protocols_.push(std::move(return_protocol));
+}
+
+void ProvenanceTaskReport::onTrigger(core::ProcessContext *context, core::ProcessSession *session)
+{
+ std::string value;
+ int64_t lvalue;
+
+ std::unique_ptr<Site2SiteClientProtocol> protocol_ = getNextProtocol();
+
+ if (protocol_ == nullptr)
+ {
+ protocol_ = std::unique_ptr<Site2SiteClientProtocol>(
+ new Site2SiteClientProtocol(0));
+ protocol_->setPortId(protocol_uuid_);
+
+ std::string host = "";
+ uint16_t sport = 0;
+
+ if (context->getProperty(hostName.getName(), value)) {
+ host = value;
+ }
+ if (context->getProperty(port.getName(), value)
+ && core::Property::StringToInt(value, lvalue)) {
+ sport = (uint16_t) lvalue;
+ }
+ std::unique_ptr<org::apache::nifi::minifi::io::DataStream> str =
+ std::unique_ptr<org::apache::nifi::minifi::io::DataStream>(
+ org::apache::nifi::minifi::io::StreamFactory::getInstance()
+ ->createSocket(host, sport));
+
+ std::unique_ptr<Site2SitePeer> peer_ = std::unique_ptr<Site2SitePeer>(
+ new Site2SitePeer(std::move(str), host, sport));
+
+ protocol_->setPeer(std::move(peer_));
+ }
+
+ if (!protocol_->bootstrap())
+ {
+ // bootstrap the client protocol if needeed
+ context->yield();
+ std::shared_ptr<Processor> processor = std::static_pointer_cast<Processor>(
+ context->getProcessorNode().getProcessor());
+ logger_->log_error("Site2Site bootstrap failed yield period %d peer ",
+ processor->getYieldPeriodMsec());
+ return;
+ }
+
+ int batch = 100;
+
+ if (context->getProperty(batchSize.getName(), value) && core::Property::StringToInt(value, lvalue))
+ {
+ batch = (int) lvalue;
--- End diff --
why is (int) necessary? Why not use the same type as lvalue?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp issue #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on the issue:
https://github.com/apache/nifi-minifi-cpp/pull/74
Hey @benqiu2016 is this ready to continue to be reviewed? I see the build did not succeed in travis.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109791492
--- Diff: libminifi/src/provenance/ProvenanceTaskReport.cpp ---
@@ -0,0 +1,221 @@
+/**
+ * @file ProvenanceTaskReport.cpp
+ * ProvenanceTaskReport class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <sys/time.h>
+#include <time.h>
+#include <sstream>
+#include <string.h>
+#include <iostream>
+
+#include "provenance/ProvenanceTaskReport.h"
+#include "../include/io/StreamFactory.h"
+#include "io/ClientSocket.h"
+#include "utils/TimeUtil.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "provenance/Provenance.h"
+#include "FlowController.h"
+
+#include "json/json.h"
+#include "json/writer.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance{
+
+const std::string ProvenanceTaskReport::ProcessorName("ProvenanceTaskReport");
+core::Property ProvenanceTaskReport::hostName("Host Name", "Remote Host Name.", "localhost");
+core::Property ProvenanceTaskReport::port("Port", "Remote Port", "9999");
+core::Property ProvenanceTaskReport::batchSize("Batch Size", "Specifies how many records to send in a single batch, at most.", "100");
+core::Relationship ProvenanceTaskReport::relation;
+
+void ProvenanceTaskReport::initialize()
+{
+ //! Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(hostName);
+ properties.insert(port);
+ properties.insert(batchSize);
+ setSupportedProperties(properties);
+ //! Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(relation);
+ setSupportedRelationships(relationships);
+}
+
--- End diff --
I don't follow. How does duplicating code decouple and add flexibility?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109740706
--- Diff: libminifi/src/Site2SiteClientProtocol.cpp ---
@@ -682,6 +682,7 @@ bool Site2SiteClientProtocol::receive(std::string transactionID,
bool Site2SiteClientProtocol::send(
std::string transactionID, DataPacket *packet, std::shared_ptr<FlowFileRecord> flowFile,
+ uint8_t *payload, int length,
--- End diff --
Can payload be embedded into DataPacket, thus simplifying this function?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r109742314
--- Diff: libminifi/src/provenance/ProvenanceTaskReport.cpp ---
@@ -0,0 +1,221 @@
+/**
+ * @file ProvenanceTaskReport.cpp
+ * ProvenanceTaskReport class implementation
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+#include <vector>
+#include <queue>
+#include <map>
+#include <set>
+#include <sys/time.h>
+#include <time.h>
+#include <sstream>
+#include <string.h>
+#include <iostream>
+
+#include "provenance/ProvenanceTaskReport.h"
+#include "../include/io/StreamFactory.h"
+#include "io/ClientSocket.h"
+#include "utils/TimeUtil.h"
+#include "core/ProcessContext.h"
+#include "core/ProcessSession.h"
+#include "provenance/Provenance.h"
+#include "FlowController.h"
+
+#include "json/json.h"
+#include "json/writer.h"
+
+namespace org {
+namespace apache {
+namespace nifi {
+namespace minifi {
+namespace provenance{
+
+const std::string ProvenanceTaskReport::ProcessorName("ProvenanceTaskReport");
+core::Property ProvenanceTaskReport::hostName("Host Name", "Remote Host Name.", "localhost");
+core::Property ProvenanceTaskReport::port("Port", "Remote Port", "9999");
+core::Property ProvenanceTaskReport::batchSize("Batch Size", "Specifies how many records to send in a single batch, at most.", "100");
+core::Relationship ProvenanceTaskReport::relation;
+
+void ProvenanceTaskReport::initialize()
+{
+ //! Set the supported properties
+ std::set<core::Property> properties;
+ properties.insert(hostName);
+ properties.insert(port);
+ properties.insert(batchSize);
+ setSupportedProperties(properties);
+ //! Set the supported relationships
+ std::set<core::Relationship> relationships;
+ relationships.insert(relation);
+ setSupportedRelationships(relationships);
+}
+
+std::unique_ptr<Site2SiteClientProtocol> ProvenanceTaskReport::getNextProtocol()
+{
+ std::lock_guard<std::mutex> protocol_lock_(protocol_mutex_);
+ if (available_protocols_.empty())
+ return nullptr;
+ std::unique_ptr<Site2SiteClientProtocol> return_pointer = std::move(available_protocols_.top());
+ available_protocols_.pop();
+ return std::move(return_pointer);
+}
+
+void ProvenanceTaskReport::returnProtocol(
+ std::unique_ptr<Site2SiteClientProtocol> return_protocol)
+{
+ std::lock_guard<std::mutex> protocol_lock_(protocol_mutex_);
+ available_protocols_.push(std::move(return_protocol));
+}
+
+void ProvenanceTaskReport::onTrigger(core::ProcessContext *context, core::ProcessSession *session)
+{
+ std::string value;
+ int64_t lvalue;
+
+ std::unique_ptr<Site2SiteClientProtocol> protocol_ = getNextProtocol();
+
+ if (protocol_ == nullptr)
+ {
+ protocol_ = std::unique_ptr<Site2SiteClientProtocol>(
+ new Site2SiteClientProtocol(0));
+ protocol_->setPortId(protocol_uuid_);
+
+ std::string host = "";
+ uint16_t sport = 0;
+
+ if (context->getProperty(hostName.getName(), value)) {
+ host = value;
+ }
+ if (context->getProperty(port.getName(), value)
+ && core::Property::StringToInt(value, lvalue)) {
+ sport = (uint16_t) lvalue;
+ }
+ std::unique_ptr<org::apache::nifi::minifi::io::DataStream> str =
+ std::unique_ptr<org::apache::nifi::minifi::io::DataStream>(
+ org::apache::nifi::minifi::io::StreamFactory::getInstance()
+ ->createSocket(host, sport));
+
+ std::unique_ptr<Site2SitePeer> peer_ = std::unique_ptr<Site2SitePeer>(
+ new Site2SitePeer(std::move(str), host, sport));
+
+ protocol_->setPeer(std::move(peer_));
+ }
+
+ if (!protocol_->bootstrap())
+ {
+ // bootstrap the client protocol if needeed
+ context->yield();
+ std::shared_ptr<Processor> processor = std::static_pointer_cast<Processor>(
+ context->getProcessorNode().getProcessor());
+ logger_->log_error("Site2Site bootstrap failed yield period %d peer ",
+ processor->getYieldPeriodMsec());
+ return;
+ }
+
+ int batch = 100;
+
+ if (context->getProperty(batchSize.getName(), value) && core::Property::StringToInt(value, lvalue))
+ {
+ batch = (int) lvalue;
+ }
+
+ std::vector<std::shared_ptr<ProvenanceEventRecord>> records;
+ std::shared_ptr<ProvenanceRepository> repo = std::static_pointer_cast<ProvenanceRepository> (context->getProvenanceRepository());
+
+ repo->getProvenanceRecord(records, batch);
+
+ if (records.size() <= 0)
+ {
+ returnProtocol(std::move(protocol_));
+ return;
+ }
+
+ Json::Value array;
+ for (auto record : records)
+ {
+ Json::Value recordJson;
+ Json::Value updatedAttributesJson;
+ Json::Value parentUuidJson;
+ Json::Value childUuidJson;
+ recordJson["eventId"] = record->getEventId().c_str();
+ recordJson["eventType"] = ProvenanceEventRecord::ProvenanceEventTypeStr[record->getEventType()];
+ recordJson["timestampMillis"] = record->getEventTime();
+ recordJson["durationMillis"] = record->getEventDuration();
+ recordJson["lineageStart"] = record->getlineageStartDate();
+ recordJson["details"] = record->getDetails().c_str();
+ recordJson["componentId"] = record->getComponentId().c_str();
+ recordJson["componentType"] = record->getComponentType().c_str();
+ recordJson["entityId"] = record->getFlowFileUuid().c_str();
+ recordJson["entityType"] = "org.apache.nifi.flowfile.FlowFile";
+ recordJson["entitySize"] = record->getFileSize();
+ recordJson["entityOffset"] = record->getFileOffset();
+
+ for (auto attr : record->getAttributes())
+ {
+ updatedAttributesJson[attr.first] = attr.second;
+ }
+ recordJson["updatedAttributes"] = updatedAttributesJson;
+
+ for (auto parentUUID : record->getParentUuids())
+ {
+ parentUuidJson.append(parentUUID.c_str());
+ }
+ recordJson["parentIds"] = parentUuidJson;
+
+ for (auto childUUID : record->getChildrenUuids())
+ {
+ childUuidJson.append(childUUID.c_str());
+ }
+ recordJson["childIds"] = childUuidJson;
+ recordJson["transitUri"] = record->getTransitUri().c_str();
+ recordJson["remoteIdentifier"] = record->getSourceSystemFlowFileIdentifier().c_str();
+ recordJson["alternateIdentifier"] = record->getAlternateIdentifierUri().c_str();
+ recordJson["application"] = "MiNiFi Flow";
--- End diff --
Maybe application name should be externalized?
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp pull request #74: Minifi 227
Posted by phrocker <gi...@git.apache.org>.
Github user phrocker commented on a diff in the pull request:
https://github.com/apache/nifi-minifi-cpp/pull/74#discussion_r111775350
--- Diff: libminifi/include/core/Processor.h ---
@@ -239,12 +242,21 @@ class Processor : public Connectable, public ConfigurableComponent,
// Trigger the Processor even if the incoming connection is empty
std::atomic<bool> _triggerWhenEmpty;
- private:
+ //! obtainSite2SiteProtocol for use
+ std::shared_ptr<Site2SiteClientProtocol> obtainSite2SiteProtocol(std::string host, uint16_t sport, uuid_t portId);
--- End diff --
@benqiu2016 Can you make a ticket to move this out of Processor? It should be a separate component. If you make the ticket I'll make this particular change.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp issue #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on the issue:
https://github.com/apache/nifi-minifi-cpp/pull/74
@apiri Thanks.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] nifi-minifi-cpp issue #74: Minifi 227
Posted by benqiu2016 <gi...@git.apache.org>.
Github user benqiu2016 commented on the issue:
https://github.com/apache/nifi-minifi-cpp/pull/74
my MAC build linter is OKW12612:build binqiu$ make linter
/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//Connection.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/ConfigurableComponent.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/ConfigurationFactory.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/Connectable.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/Core.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/FlowConfiguration.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/FlowFile.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/logging/BaseLogger.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/logging/LogAppenders.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/logging/Logger.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/ProcessContext.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/ProcessGroup.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/P
rocessor.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/ProcessorConfig.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/ProcessorNode.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/ProcessSession.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/ProcessSessionFactory.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/Property.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/Relationship.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/repository/FlowFileRepository.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/Repository.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/RepositoryFactory.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/Scheduling.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//core/yaml/YamlConfiguration.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//EventDrivenSchedulingAgent.h,/Users/binqiu/report/nifi-
minifi-cpp/libminifi/include//Exception.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//FlowController.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//FlowControlProtocol.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//FlowFileRecord.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//io/BaseStream.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//io/ClientSocket.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//io/CRCStream.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//io/DataStream.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//io/EndianCheck.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//io/Serializable.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//io/Sockets.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//io/StreamFactory.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//io/tls/TLSSocket.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//io/validation.h,/Use
rs/binqiu/report/nifi-minifi-cpp/libminifi/include//processors/AppendHostInfo.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//processors/ExecuteProcess.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//processors/GenerateFlowFile.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//processors/GetFile.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//processors/ListenHTTP.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//processors/ListenSyslog.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//processors/LogAttribute.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//processors/PutFile.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//processors/TailFile.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//properties/Configure.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//provenance/Provenance.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//provenance/ProvenanceRepository.h,/Users/binqiu/report/n
ifi-minifi-cpp/libminifi/include//provenance/ProvenanceTaskReport.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//RemoteProcessorGroupPort.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//ResourceClaim.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//SchedulingAgent.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//Site2SiteClientProtocol.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//Site2SitePeer.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//ThreadedSchedulingAgent.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//TimerDrivenSchedulingAgent.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//utils/FailurePolicy.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//utils/StringUtils.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//utils/ThreadPool.h,/Users/binqiu/report/nifi-minifi-cpp/libminifi/include//utils/TimeUtil.h,
/Users/binqiu/report/nifi-minifi-cpp/libminifi/src//Configure.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//Connection.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/ConfigurableComponent.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/ConfigurationFactory.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/Connectable.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/Core.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/FlowConfiguration.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/FlowFile.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/logging/BaseLogger.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/logging/LogAppenders.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/logging/Logger.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/ProcessGroup.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/Processor.cpp /Users/binqiu/report/nifi
-minifi-cpp/libminifi/src//core/ProcessorNode.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/ProcessSession.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/ProcessSessionFactory.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/Property.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/repository/FlowFileRepository.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/Repository.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/RepositoryFactory.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/yaml/YamlConfiguration.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//EventDrivenSchedulingAgent.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//FlowController.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//FlowControlProtocol.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//FlowFileRecord.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/BaseStream.cpp /Users/binqiu/report/
nifi-minifi-cpp/libminifi/src//io/ClientSocket.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/CRCStream.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/DataStream.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/EndianCheck.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/Serializable.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/StreamFactory.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/tls/TLSSocket.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/AppendHostInfo.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/ExecuteProcess.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/GenerateFlowFile.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/GetFile.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/ListenHTTP.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/ListenSyslog.cpp /Users/binqiu/report/nifi-minifi-cpp/
libminifi/src//processors/LogAttribute.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/PutFile.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/TailFile.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//provenance/Provenance.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//provenance/ProvenanceRepository.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//provenance/ProvenanceTaskReport.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//RemoteProcessorGroupPort.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//ResourceClaim.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//SchedulingAgent.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//Site2SiteClientProtocol.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//Site2SitePeer.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//ThreadedSchedulingAgent.cpp /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//TimerDrivenSchedulingAgent.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//Configure.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//Connection.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/ConfigurableComponent.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/ConfigurationFactory.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/Connectable.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/Core.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/FlowConfiguration.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/FlowFile.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/logging/BaseLogger.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/logging/LogAppenders.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/logging/Logger.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/ProcessGroup.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/Processor.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/ProcessorNode.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/ProcessSession.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/ProcessSessionFactory.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/Property.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/repository/FlowFileRepository.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/Repository.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/RepositoryFactory.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//core/yaml/YamlConfiguration.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//EventDrivenSchedulingAgent.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//FlowController.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//FlowControlProtocol.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//FlowFileRecord.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/BaseStream.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/ClientSocket.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/CRCStream.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/DataStream.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/EndianCheck.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/Serializable.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/StreamFactory.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//io/tls/TLSSocket.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/AppendHostInfo.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/ExecuteProcess.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/GenerateFlowFile.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/GetFile.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/ListenHTTP.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/ListenSyslog.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/LogAttribute.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/PutFile.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//processors/TailFile.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//provenance/Provenance.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//provenance/ProvenanceRepository.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//provenance/ProvenanceTaskReport.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//RemoteProcessorGroupPort.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//ResourceClaim.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//SchedulingAgent.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//Site2SiteClientProtocol.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//Site2SitePeer.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//ThreadedSchedulingAgent.cpp
Done processing /Users/binqiu/report/nifi-minifi-cpp/libminifi/src//TimerDrivenSchedulingAgent.cpp
Total errors found: 0
Built target linter
HW12612:build binqiu$
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---