This is an automated email from the ASF dual-hosted git repository.
exceptionfactory pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 6a007f9159 NIFI-12270 Added UPLOAD Provenance Event Type to nifi-api
6a007f9159 is described below
commit 6a007f9159f4b2cc71a01cbff650136a2fa62bbf
Author: lehelb <[email protected]>
AuthorDate: Fri Nov 3 09:55:23 2023 -0500
NIFI-12270 Added UPLOAD Provenance Event Type to nifi-api
This closes #8094
Signed-off-by: David Handermann <[email protected]>
---
.../org/apache/nifi/provenance/FileResource.java | 29 ++++++++++++++
.../nifi/provenance/ProvenanceEventType.java | 5 +++
.../apache/nifi/provenance/ProvenanceReporter.java | 45 ++++++++++++++++++++-
.../provenance/StandardProvenanceEventRecord.java | 8 ++--
nifi-docs/src/main/asciidoc/developer-guide.adoc | 1 +
nifi-docs/src/main/asciidoc/user-guide.adoc | 1 +
.../apache/nifi/util/MockProvenanceReporter.java | 37 ++++++++++++++---
.../repository/StandardProcessSession.java | 3 +-
.../repository/StandardProvenanceReporter.java | 46 ++++++++++++++++++----
.../webapp/js/nf/provenance/nf-provenance-table.js | 2 +-
10 files changed, 158 insertions(+), 19 deletions(-)
diff --git
a/nifi-api/src/main/java/org/apache/nifi/provenance/FileResource.java
b/nifi-api/src/main/java/org/apache/nifi/provenance/FileResource.java
new file mode 100644
index 0000000000..8ec84b58b0
--- /dev/null
+++ b/nifi-api/src/main/java/org/apache/nifi/provenance/FileResource.java
@@ -0,0 +1,29 @@
+/*
+ * 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.
+ */
+package org.apache.nifi.provenance;
+
+/**
+ * Holds information of a file resource for UPLOAD
+ * provenance events.
+ */
+public record FileResource(String location, long size) {
+
+ @Override
+ public String toString() {
+ return "FileResource[location=%s, size=%d]".formatted(location, size);
+ }
+}
diff --git
a/nifi-api/src/main/java/org/apache/nifi/provenance/ProvenanceEventType.java
b/nifi-api/src/main/java/org/apache/nifi/provenance/ProvenanceEventType.java
index 19b13248e7..b5a8affc94 100644
--- a/nifi-api/src/main/java/org/apache/nifi/provenance/ProvenanceEventType.java
+++ b/nifi-api/src/main/java/org/apache/nifi/provenance/ProvenanceEventType.java
@@ -46,6 +46,11 @@ public enum ProvenanceEventType {
*/
SEND,
+ /**
+ * Indicates a provenance event for sending data from a filesystem to an
external process
+ */
+ UPLOAD,
+
/**
* Indicates a provenance event for sending remote invocation request to
an external process.
* This event type is used to represent other operations than transferring
data (RECEIVE, FETCH or SEND),
diff --git
a/nifi-api/src/main/java/org/apache/nifi/provenance/ProvenanceReporter.java
b/nifi-api/src/main/java/org/apache/nifi/provenance/ProvenanceReporter.java
index eeffc7fe93..2f64864a00 100644
--- a/nifi-api/src/main/java/org/apache/nifi/provenance/ProvenanceReporter.java
+++ b/nifi-api/src/main/java/org/apache/nifi/provenance/ProvenanceReporter.java
@@ -16,11 +16,12 @@
*/
package org.apache.nifi.provenance;
-import java.util.Collection;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.Relationship;
+import java.util.Collection;
+
/**
* ProvenanceReporter generates and records Provenance-related events. A
* ProvenanceReporter is always tied to a {@link ProcessSession}. Any events
@@ -315,6 +316,47 @@ public interface ProvenanceReporter {
*/
void send(FlowFile flowFile, String transitUri, String details, long
transmissionMillis, boolean force);
+ /**
+ * Emits a Provenance Event of type {@link ProvenanceEventType#UPLOAD
UPLOAD}
+ * that indicates that an external resource was sent to an external
+ * destination. The external resource may be a remote system or may be a
+ * local destination, such as the local file system but is external to
NiFi.
+ *
+ * @param flowFile the FlowFile that was sent
+ * @param fileResource the FileResource that was uploaded
+ * @param transitUri A URI that provides information about the System and
+ * Protocol information over which the transfer occurred. The intent of
this
+ * field is such that both the sender and the receiver can publish the
+ * events to an external Enterprise-wide system that is then able to
+ * correlate the SEND and RECEIVE events.
+ */
+ void upload(FlowFile flowFile, FileResource fileResource, String
transitUri);
+
+ /**
+ * Emits a Provenance Event of type {@link ProvenanceEventType#UPLOAD
UPLOAD}
+ * that indicates that an external resource was sent to an external
+ * destination. The external resource may be a remote system or may be a
+ * local destination, such as the local file system but is external to
NiFi.
+ *
+ * @param flowFile the FlowFile that was sent
+ * @param fileResource the FileResource that was uploaded
+ * @param transitUri A URI that provides information about the System and
+ * Protocol information over which the transfer occurred. The intent of
this
+ * field is such that both the sender and the receiver can publish the
+ * events to an external Enterprise-wide system that is then able to
+ * correlate the SEND and RECEIVE events.
+ * @param details additional details related to the SEND event, such as a
+ * remote system's Distinguished Name
+ * @param transmissionMillis the number of milliseconds spent sending the
+ * data to the remote system
+ * @param force if <code>true</code>, this event will be added to the
+ * Provenance Repository immediately and will still be persisted if the
+ * {@link org.apache.nifi.processor.ProcessSession ProcessSession} to
which this
+ * ProvenanceReporter is associated is rolled back. Otherwise, the Event
+ * will be recorded only on a successful session commit.
+ */
+ void upload(FlowFile flowFile, FileResource fileResource, String
transitUri, String details, long transmissionMillis, boolean force);
+
/**
* Emits a Provenance Event of type {@link
ProvenanceEventType#REMOTE_INVOCATION}
* that indicates a remote invocation is requested to an external endpoint
using
@@ -675,7 +717,6 @@ public interface ProvenanceReporter {
* @param details any relevant details about the CREATE event
*/
void create(FlowFile flowFile, String details);
-
/**
* @return the number of FlowFiles for which there was a RECEIVE event
*/
diff --git
a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StandardProvenanceEventRecord.java
b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StandardProvenanceEventRecord.java
index 84e7419a29..cc3306fca8 100644
---
a/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StandardProvenanceEventRecord.java
+++
b/nifi-commons/nifi-data-provenance-utils/src/main/java/org/apache/nifi/provenance/StandardProvenanceEventRecord.java
@@ -16,15 +16,16 @@
*/
package org.apache.nifi.provenance;
+import org.apache.nifi.flowfile.FlowFile;
+import org.apache.nifi.flowfile.attributes.CoreAttributes;
+import org.apache.nifi.processor.Relationship;
+
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import org.apache.nifi.flowfile.FlowFile;
-import org.apache.nifi.flowfile.attributes.CoreAttributes;
-import org.apache.nifi.processor.Relationship;
/**
* Holder for provenance relevant information
@@ -821,6 +822,7 @@ public class StandardProvenanceEventRecord implements
ProvenanceEventRecord {
case RECEIVE:
case FETCH:
case SEND:
+ case UPLOAD:
assertSet(transitUri, "Transit URI");
break;
case ROUTE:
diff --git a/nifi-docs/src/main/asciidoc/developer-guide.adoc
b/nifi-docs/src/main/asciidoc/developer-guide.adoc
index 1765391ee6..db9816bc2b 100644
--- a/nifi-docs/src/main/asciidoc/developer-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/developer-guide.adoc
@@ -969,6 +969,7 @@ The different event types for provenance reporting are:
|REPLAY |Indicates a provenance event for replaying a
FlowFile. The UUID of the event indicates the UUID of the original FlowFile
that is being replayed. The event contains one Parent UUID that is also the
UUID of the FlowFile that is being replayed and one Child UUID that is the UUID
of the a newly created FlowFile that will be re-queued for processing
|ROUTE |Indicates that a FlowFile was routed to a specified
relationship and provides information about why the FlowFile was routed to this
relationship
|SEND |Indicates a provenance event for sending data to an
external process
+|UPLOAD |Indicates a provenance event for uploading external
data to an external system
|UNKNOWN |Indicates that the type of provenance event is
unknown because the user who is attempting to access the event is not
authorized to know the type
|======================
diff --git a/nifi-docs/src/main/asciidoc/user-guide.adoc
b/nifi-docs/src/main/asciidoc/user-guide.adoc
index bc5cc447d7..b8c3129a85 100644
--- a/nifi-docs/src/main/asciidoc/user-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/user-guide.adoc
@@ -2897,6 +2897,7 @@ The provenance event types are:
|REPLAY |Indicates a provenance event for replaying a FlowFile
|ROUTE |Indicates that a FlowFile was routed to a specified
relationship and provides information about why the FlowFile was routed to this
relationship
|SEND |Indicates a provenance event for sending data to an
external process
+|UPLOAD |Indicates a provenance event for uploading external
data to an external system
|UNKNOWN |Indicates that the type of provenance event is
unknown because the user who is attempting to access the event is not
authorized to know the type
|======================
diff --git
a/nifi-mock/src/main/java/org/apache/nifi/util/MockProvenanceReporter.java
b/nifi-mock/src/main/java/org/apache/nifi/util/MockProvenanceReporter.java
index f5af5e3d10..38c1f0e8dc 100644
--- a/nifi-mock/src/main/java/org/apache/nifi/util/MockProvenanceReporter.java
+++ b/nifi-mock/src/main/java/org/apache/nifi/util/MockProvenanceReporter.java
@@ -16,10 +16,6 @@
*/
package org.apache.nifi.util;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.LinkedHashSet;
-import java.util.Set;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.FlowFileHandlingException;
@@ -28,9 +24,15 @@ import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.ProvenanceEventType;
import org.apache.nifi.provenance.ProvenanceReporter;
import org.apache.nifi.provenance.StandardProvenanceEventRecord;
+import org.apache.nifi.provenance.FileResource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.LinkedHashSet;
+import java.util.Set;
+
public class MockProvenanceReporter implements ProvenanceReporter {
private static final Logger logger =
LoggerFactory.getLogger(MockProvenanceReporter.class);
private final MockProcessSession session;
@@ -225,6 +227,32 @@ public class MockProvenanceReporter implements
ProvenanceReporter {
}
}
+ @Override
+ public void upload(final FlowFile flowFile, final FileResource
fileResource, final String transitUri) {
+ upload(flowFile, fileResource, transitUri, null, -1L, true);
+
+ }
+
+ @Override
+ public void upload(FlowFile flowFile, FileResource fileResource, String
transitUri, String details, long transmissionMillis, boolean force) {
+ try {
+ final String fileResourceDetails = fileResource.toString();
+ final String enrichedDetails = details == null ?
fileResourceDetails : details + " " + fileResourceDetails;
+ final ProvenanceEventRecord record = build(flowFile,
ProvenanceEventType.UPLOAD)
+ .setTransitUri(transitUri)
+ .setEventDuration(transmissionMillis)
+ .setDetails(enrichedDetails)
+ .build();
+ if (force) {
+
sharedSessionState.addProvenanceEvents(Collections.singleton(record));
+ } else {
+ events.add(record);
+ }
+ } catch (final Exception e) {
+ logger.error("Failed to generate Provenance Event", e);
+ }
+ }
+
@Override
public void send(final FlowFile flowFile, final String transitUri, final
boolean force) {
send(flowFile, transitUri, -1L, true);
@@ -542,5 +570,4 @@ public class MockProvenanceReporter implements
ProvenanceReporter {
builder.setComponentType(processorType);
return builder;
}
-
}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
index 480bdf7fc7..64ef1c5aff 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
@@ -1003,7 +1003,8 @@ public class StandardProcessSession implements
ProcessSession, ProvenanceEventEn
// the representation of the FlowFile as it is
committed, as this is the only way in which it really
// exists in our system -- all other
representations are volatile representations that have not been
// exposed.
- return enrich(rawEvent, flowFileRecordMap,
checkpoint.records, rawEvent.getEventType() != ProvenanceEventType.SEND,
commitNanos);
+ final boolean isUpdateAttributes =
rawEvent.getEventType() != ProvenanceEventType.SEND && rawEvent.getEventType()
!= ProvenanceEventType.UPLOAD;
+ return enrich(rawEvent, flowFileRecordMap,
checkpoint.records, isUpdateAttributes, commitNanos);
} else if (autoTermIterator != null &&
autoTermIterator.hasNext()) {
return enrich(autoTermIterator.next(),
flowFileRecordMap, checkpoint.records, true, commitNanos);
}
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProvenanceReporter.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProvenanceReporter.java
index 38b7ffe5f3..790f173b49 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProvenanceReporter.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProvenanceReporter.java
@@ -24,6 +24,7 @@ import org.apache.nifi.provenance.ProvenanceEventBuilder;
import org.apache.nifi.provenance.ProvenanceEventRecord;
import org.apache.nifi.provenance.ProvenanceEventRepository;
import org.apache.nifi.provenance.ProvenanceEventType;
+import org.apache.nifi.provenance.FileResource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -50,7 +51,7 @@ public class StandardProvenanceReporter implements
InternalProvenanceReporter {
private long bytesFetched = 0L;
public StandardProvenanceReporter(final Predicate<FlowFile>
flowfileKnownCheck, final String processorId, final String processorType,
- final ProvenanceEventRepository repository, final
ProvenanceEventEnricher enricher) {
+ final ProvenanceEventRepository
repository, final ProvenanceEventEnricher enricher) {
this.flowfileKnownCheck = flowfileKnownCheck;
this.processorId = processorId;
this.processorType = processorType;
@@ -204,10 +205,10 @@ public class StandardProvenanceReporter implements
InternalProvenanceReporter {
try {
final ProvenanceEventRecord record = build(flowFile,
ProvenanceEventType.FETCH)
- .setTransitUri(transitUri)
- .setEventDuration(transmissionMillis)
- .setDetails(details)
- .build();
+ .setTransitUri(transitUri)
+ .setEventDuration(transmissionMillis)
+ .setDetails(details)
+ .build();
events.add(record);
@@ -226,6 +227,11 @@ public class StandardProvenanceReporter implements
InternalProvenanceReporter {
send(flowFile, transitUri, transmissionMillis, true);
}
+ @Override
+ public void send(final FlowFile flowFile, final String transitUri, final
boolean force) {
+ send(flowFile, transitUri, -1L, force);
+ }
+
@Override
public void send(final FlowFile flowFile, final String transitUri) {
send(flowFile, transitUri, null, -1L, true);
@@ -276,8 +282,34 @@ public class StandardProvenanceReporter implements
InternalProvenanceReporter {
}
@Override
- public void send(final FlowFile flowFile, final String transitUri, final
boolean force) {
- send(flowFile, transitUri, -1L, force);
+ public void upload(final FlowFile flowFile, final FileResource
fileResource, final String transitUri) {
+ upload(flowFile, fileResource, transitUri, null, -1L, true);
+ }
+
+ @Override
+ public void upload(final FlowFile flowFile, final FileResource
fileResource, final String transitUri, final String details, final long
transmissionMillis, final boolean force) {
+ try {
+ final String fileResourceDetails = fileResource.toString();
+ final String enrichedDetails = details == null ?
fileResourceDetails : details + " " + fileResourceDetails;
+ final ProvenanceEventRecord record = build(flowFile,
ProvenanceEventType.UPLOAD)
+ .setTransitUri(transitUri)
+ .setEventDuration(transmissionMillis)
+ .setDetails(enrichedDetails)
+ .build();
+ // If the transmissionMillis field has been populated, use zero as
the value of commitNanos (the call to System.nanoTime() is expensive but the
value will be ignored).
+ final long commitNanos = transmissionMillis < 0 ?
System.nanoTime() : 0L;
+ final ProvenanceEventRecord enriched = eventEnricher == null ?
record : eventEnricher.enrich(record, flowFile, commitNanos);
+
+ if (force) {
+ repository.registerEvent(enriched);
+ } else {
+ events.add(enriched);
+ }
+
+ bytesSent += fileResource.size();
+ } catch (final Exception e) {
+ logger.error("Failed to generate Provenance Event", e);
+ }
}
@Override
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-ui/src/main/webapp/js/nf/provenance/nf-provenance-table.js
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-ui/src/main/webapp/js/nf/provenance/nf-provenance-table.js
index 09d134c547..1b526fda82 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-ui/src/main/webapp/js/nf/provenance/nf-provenance-table.js
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-ui/src/main/webapp/js/nf/provenance/nf-provenance-table.js
@@ -1321,7 +1321,7 @@
}
// conditionally show SEND details
- if (event.eventType === 'SEND') {
+ if (event.eventType === 'SEND' || event.eventType ===
'UPLOAD') {
formatEventDetail('Transit Uri', event.transitUri);
}
