You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "exceptionfactory (via GitHub)" <gi...@apache.org> on 2023/04/06 21:07:21 UTC
[GitHub] [nifi] exceptionfactory commented on a diff in pull request #7003: NIFI-11241: Initial implementation of Python-based Processor API with…
exceptionfactory commented on code in PR #7003:
URL: https://github.com/apache/nifi/pull/7003#discussion_r1158737181
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcess.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.py4j;
+
+import org.apache.nifi.py4j.client.JavaObjectBindings;
+import org.apache.nifi.py4j.client.NiFiPythonGateway;
+import org.apache.nifi.py4j.client.StandardPythonClient;
+import org.apache.nifi.py4j.server.NiFiGatewayServer;
+import org.apache.nifi.python.ControllerServiceTypeLookup;
+import org.apache.nifi.python.PythonController;
+import org.apache.nifi.python.PythonProcessConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import py4j.CallbackClient;
+import py4j.GatewayServer;
+
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+// TODO / Figure Out for MVP:
+// MUST DO:
+// - Documentation
+// - Admin Guide
+// - JavaDocs
+// - Developer Guide
+// - Explain how communication between Java & Python work.
+// - Java is preferred, Python is slower and more expensive b/c of network
+// - Different Extension Points (FlowFileTransform, RecordTransform)
+// - What the API Looks like, Links to JavaDocs for ProcessContext, etc.
+// - Example Code
+// - Exposing properties
+// - Relationships
+// - Controller Services
+// - Need to update docs to show the interfaces that are exposed, explain how to get these...
+// - Design Doc
+// - Setup proper logging on the Python side: https://docs.python.org/2/howto/logging-cookbook.html#using-file-rotation
+// - For FlowFileTransform, allow the result to contain either a byte array or a String. If a String, just convert in the parent class.
+// - Figure out how to deal with Python Packaging
+// - Need to figure out how to deal with additionalDetails.html, docs directory in python project typically?
+// - Understand how to deal with versioning
+// - Look at performance improvements for Py4J - socket comms appear to be INCREDIBLY slow.
+// - Create test that calls Python 1M times. Just returns 'hello'. See how long it takes
+// - Create test that calls Python 1M times. Returns <java object>.toString() and see how long it takes.
+// - Will help to understand if it's the call from Java to Python that's slow, Python to Java, or both.
+// - Performance concern for TransformRecord
+// - Currently, triggering the transform() method is pretty fast. But then the Result object comes back and we have to call into the Python side to call the getters
+// over and over. Need to look into instead serializing the entire response as JSON and sending that back.
+// - Also, since this is heavy JSON processing, might want to consider ORJSON or something like that instead of inbuilt JSON parser/generator
+// - Test pip install nifi-my-proc, does nifi pick it up?
+// - When ran DetectObjectInImage with multiple threads, Python died. Need to figure out why.
+// - If Python Process dies, need to create a new process and need to then create all of the Processors that were in that Process and initialize them.
+// - Milestone 2 or 3, not Milestone 1.
+// - Remove test-pypi usage from ExtensionManager.py
+// - Additional Interfaces beyond just FlowFileTransform
+// - FlowFileSource
+// - Restructure Maven projects
+// - Should this all go under Framework?
+//
+//
+// CONSIDER:
+// - Clustering: Ensure component on all nodes?
+// - Consider "pip freeze" type of thing to ensure that python dependencies are same across nodes when joining cluster.
+// - Update python code to use python_style_method_names instead of javaStyleMethodNames
+// - Also add 'failure' and 'original' relationships to FlowFileTransform
+//
+//
+// Can punt for now:
+// - We have an issue with objects created from Processor calling into Java. Is fine when we provide objects to Python but when
+// it makes a callback, those objects are bound and never unbound!!!
+// *** This appears to be fine as long as on the Python side we set manage_memory=True ***
+// But it does cut perf in half. May be room for improvement somehow? By implementing a 'bulk delete' custom command?
+// - Appears to still be issues with timeout in nifi when set to 10 secs... initially will probably just leave default of 0sec.
+
+public class PythonProcess {
+ private static final Logger logger = LoggerFactory.getLogger(PythonProcess.class);
+ private static final String PYTHON_CONTROLLER_FILENAME = "Controller.py";
+
+ private final PythonProcessConfig processConfig;
+ private final ControllerServiceTypeLookup controllerServiceTypeLookup;
+ private final File virtualEnvHome;
+ private GatewayServer server;
+ private PythonController controller;
+ private Process process;
+ private NiFiPythonGateway gateway;
+ private final Map<String, Boolean> processorPrefersIsolation = new ConcurrentHashMap<>();
+
+
+ public PythonProcess(final PythonProcessConfig processConfig, final ControllerServiceTypeLookup controllerServiceTypeLookup, final File virtualEnvHome) {
+ this.processConfig = processConfig;
+ this.controllerServiceTypeLookup = controllerServiceTypeLookup;
+ this.virtualEnvHome = virtualEnvHome;
+ }
+
+ public PythonController getController() {
+ return controller;
+ }
+
+ public void start() throws IOException {
+ // TODO: Look into using configured TLS Certs to make this secure by default.
+ final ServerSocketFactory serverSocketFactory = ServerSocketFactory.getDefault();
+ final SocketFactory socketFactory = SocketFactory.getDefault();
+
+ final int timeoutMillis = (int) processConfig.getCommsTimeout().toMillis();
+ final String authToken = null;
Review Comment:
Although TLS might not be required, using a random authentication token would avoid potential manipulation from other processes running on localhost. Granted this is somewhat unlikely, but with standard NiFi Processors like PutTCP and other scripted processors, it would be good to have this as a basic security measure.
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcess.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.py4j;
+
+import org.apache.nifi.py4j.client.JavaObjectBindings;
+import org.apache.nifi.py4j.client.NiFiPythonGateway;
+import org.apache.nifi.py4j.client.StandardPythonClient;
+import org.apache.nifi.py4j.server.NiFiGatewayServer;
+import org.apache.nifi.python.ControllerServiceTypeLookup;
+import org.apache.nifi.python.PythonController;
+import org.apache.nifi.python.PythonProcessConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import py4j.CallbackClient;
+import py4j.GatewayServer;
+
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+// TODO / Figure Out for MVP:
+// MUST DO:
+// - Documentation
+// - Admin Guide
+// - JavaDocs
+// - Developer Guide
+// - Explain how communication between Java & Python work.
+// - Java is preferred, Python is slower and more expensive b/c of network
+// - Different Extension Points (FlowFileTransform, RecordTransform)
+// - What the API Looks like, Links to JavaDocs for ProcessContext, etc.
+// - Example Code
+// - Exposing properties
+// - Relationships
+// - Controller Services
+// - Need to update docs to show the interfaces that are exposed, explain how to get these...
+// - Design Doc
+// - Setup proper logging on the Python side: https://docs.python.org/2/howto/logging-cookbook.html#using-file-rotation
+// - For FlowFileTransform, allow the result to contain either a byte array or a String. If a String, just convert in the parent class.
+// - Figure out how to deal with Python Packaging
+// - Need to figure out how to deal with additionalDetails.html, docs directory in python project typically?
+// - Understand how to deal with versioning
+// - Look at performance improvements for Py4J - socket comms appear to be INCREDIBLY slow.
+// - Create test that calls Python 1M times. Just returns 'hello'. See how long it takes
+// - Create test that calls Python 1M times. Returns <java object>.toString() and see how long it takes.
+// - Will help to understand if it's the call from Java to Python that's slow, Python to Java, or both.
+// - Performance concern for TransformRecord
+// - Currently, triggering the transform() method is pretty fast. But then the Result object comes back and we have to call into the Python side to call the getters
+// over and over. Need to look into instead serializing the entire response as JSON and sending that back.
+// - Also, since this is heavy JSON processing, might want to consider ORJSON or something like that instead of inbuilt JSON parser/generator
+// - Test pip install nifi-my-proc, does nifi pick it up?
+// - When ran DetectObjectInImage with multiple threads, Python died. Need to figure out why.
+// - If Python Process dies, need to create a new process and need to then create all of the Processors that were in that Process and initialize them.
+// - Milestone 2 or 3, not Milestone 1.
+// - Remove test-pypi usage from ExtensionManager.py
+// - Additional Interfaces beyond just FlowFileTransform
+// - FlowFileSource
+// - Restructure Maven projects
+// - Should this all go under Framework?
+//
+//
+// CONSIDER:
+// - Clustering: Ensure component on all nodes?
+// - Consider "pip freeze" type of thing to ensure that python dependencies are same across nodes when joining cluster.
+// - Update python code to use python_style_method_names instead of javaStyleMethodNames
+// - Also add 'failure' and 'original' relationships to FlowFileTransform
+//
+//
+// Can punt for now:
+// - We have an issue with objects created from Processor calling into Java. Is fine when we provide objects to Python but when
+// it makes a callback, those objects are bound and never unbound!!!
+// *** This appears to be fine as long as on the Python side we set manage_memory=True ***
+// But it does cut perf in half. May be room for improvement somehow? By implementing a 'bulk delete' custom command?
+// - Appears to still be issues with timeout in nifi when set to 10 secs... initially will probably just leave default of 0sec.
+
+public class PythonProcess {
+ private static final Logger logger = LoggerFactory.getLogger(PythonProcess.class);
+ private static final String PYTHON_CONTROLLER_FILENAME = "Controller.py";
+
+ private final PythonProcessConfig processConfig;
+ private final ControllerServiceTypeLookup controllerServiceTypeLookup;
+ private final File virtualEnvHome;
+ private GatewayServer server;
+ private PythonController controller;
+ private Process process;
+ private NiFiPythonGateway gateway;
+ private final Map<String, Boolean> processorPrefersIsolation = new ConcurrentHashMap<>();
+
+
+ public PythonProcess(final PythonProcessConfig processConfig, final ControllerServiceTypeLookup controllerServiceTypeLookup, final File virtualEnvHome) {
+ this.processConfig = processConfig;
+ this.controllerServiceTypeLookup = controllerServiceTypeLookup;
+ this.virtualEnvHome = virtualEnvHome;
+ }
+
+ public PythonController getController() {
+ return controller;
+ }
+
+ public void start() throws IOException {
+ // TODO: Look into using configured TLS Certs to make this secure by default.
+ final ServerSocketFactory serverSocketFactory = ServerSocketFactory.getDefault();
+ final SocketFactory socketFactory = SocketFactory.getDefault();
Review Comment:
All of this communication should be limited to `localhost`, correct? If so, TLS may not be necessary. Is there some other reason to support TLS? Depending on the rationale, this may need to be addressed for the initial version.
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcess.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.py4j;
+
+import org.apache.nifi.py4j.client.JavaObjectBindings;
+import org.apache.nifi.py4j.client.NiFiPythonGateway;
+import org.apache.nifi.py4j.client.StandardPythonClient;
+import org.apache.nifi.py4j.server.NiFiGatewayServer;
+import org.apache.nifi.python.ControllerServiceTypeLookup;
+import org.apache.nifi.python.PythonController;
+import org.apache.nifi.python.PythonProcessConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import py4j.CallbackClient;
+import py4j.GatewayServer;
+
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+// TODO / Figure Out for MVP:
+// MUST DO:
+// - Documentation
+// - Admin Guide
+// - JavaDocs
+// - Developer Guide
+// - Explain how communication between Java & Python work.
+// - Java is preferred, Python is slower and more expensive b/c of network
+// - Different Extension Points (FlowFileTransform, RecordTransform)
+// - What the API Looks like, Links to JavaDocs for ProcessContext, etc.
+// - Example Code
+// - Exposing properties
+// - Relationships
+// - Controller Services
+// - Need to update docs to show the interfaces that are exposed, explain how to get these...
+// - Design Doc
+// - Setup proper logging on the Python side: https://docs.python.org/2/howto/logging-cookbook.html#using-file-rotation
+// - For FlowFileTransform, allow the result to contain either a byte array or a String. If a String, just convert in the parent class.
+// - Figure out how to deal with Python Packaging
+// - Need to figure out how to deal with additionalDetails.html, docs directory in python project typically?
+// - Understand how to deal with versioning
+// - Look at performance improvements for Py4J - socket comms appear to be INCREDIBLY slow.
+// - Create test that calls Python 1M times. Just returns 'hello'. See how long it takes
+// - Create test that calls Python 1M times. Returns <java object>.toString() and see how long it takes.
+// - Will help to understand if it's the call from Java to Python that's slow, Python to Java, or both.
+// - Performance concern for TransformRecord
+// - Currently, triggering the transform() method is pretty fast. But then the Result object comes back and we have to call into the Python side to call the getters
+// over and over. Need to look into instead serializing the entire response as JSON and sending that back.
+// - Also, since this is heavy JSON processing, might want to consider ORJSON or something like that instead of inbuilt JSON parser/generator
+// - Test pip install nifi-my-proc, does nifi pick it up?
+// - When ran DetectObjectInImage with multiple threads, Python died. Need to figure out why.
+// - If Python Process dies, need to create a new process and need to then create all of the Processors that were in that Process and initialize them.
+// - Milestone 2 or 3, not Milestone 1.
+// - Remove test-pypi usage from ExtensionManager.py
+// - Additional Interfaces beyond just FlowFileTransform
+// - FlowFileSource
+// - Restructure Maven projects
+// - Should this all go under Framework?
+//
+//
+// CONSIDER:
+// - Clustering: Ensure component on all nodes?
+// - Consider "pip freeze" type of thing to ensure that python dependencies are same across nodes when joining cluster.
+// - Update python code to use python_style_method_names instead of javaStyleMethodNames
+// - Also add 'failure' and 'original' relationships to FlowFileTransform
+//
+//
+// Can punt for now:
+// - We have an issue with objects created from Processor calling into Java. Is fine when we provide objects to Python but when
+// it makes a callback, those objects are bound and never unbound!!!
+// *** This appears to be fine as long as on the Python side we set manage_memory=True ***
+// But it does cut perf in half. May be room for improvement somehow? By implementing a 'bulk delete' custom command?
+// - Appears to still be issues with timeout in nifi when set to 10 secs... initially will probably just leave default of 0sec.
+
+public class PythonProcess {
+ private static final Logger logger = LoggerFactory.getLogger(PythonProcess.class);
+ private static final String PYTHON_CONTROLLER_FILENAME = "Controller.py";
+
+ private final PythonProcessConfig processConfig;
+ private final ControllerServiceTypeLookup controllerServiceTypeLookup;
+ private final File virtualEnvHome;
+ private GatewayServer server;
+ private PythonController controller;
+ private Process process;
+ private NiFiPythonGateway gateway;
+ private final Map<String, Boolean> processorPrefersIsolation = new ConcurrentHashMap<>();
+
+
+ public PythonProcess(final PythonProcessConfig processConfig, final ControllerServiceTypeLookup controllerServiceTypeLookup, final File virtualEnvHome) {
+ this.processConfig = processConfig;
+ this.controllerServiceTypeLookup = controllerServiceTypeLookup;
+ this.virtualEnvHome = virtualEnvHome;
+ }
+
+ public PythonController getController() {
+ return controller;
+ }
+
+ public void start() throws IOException {
+ // TODO: Look into using configured TLS Certs to make this secure by default.
+ final ServerSocketFactory serverSocketFactory = ServerSocketFactory.getDefault();
+ final SocketFactory socketFactory = SocketFactory.getDefault();
+
+ final int timeoutMillis = (int) processConfig.getCommsTimeout().toMillis();
+ final String authToken = null;
+ final CallbackClient callbackClient = new CallbackClient(GatewayServer.DEFAULT_PYTHON_PORT, GatewayServer.defaultAddress(), authToken,
+ 50000L, TimeUnit.MILLISECONDS, socketFactory, false, timeoutMillis);
+
+ final JavaObjectBindings bindings = new JavaObjectBindings();
+ gateway = new NiFiPythonGateway(bindings, null, callbackClient);
+ gateway.startup();
+
+ server = new NiFiGatewayServer(gateway,
+ 0,
+ GatewayServer.defaultAddress(),
+ timeoutMillis,
+ timeoutMillis,
+ Collections.emptyList(),
+ serverSocketFactory);
+ server.start();
+
+ final int listeningPort = server.getListeningPort();
+
+ setupEnvironment();
+ this.process = launchPythonProcess(listeningPort);
+
+ final StandardPythonClient pythonClient = new StandardPythonClient(gateway);
+ controller = pythonClient.getController();
+
+ final long timeout = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(60L);
+ Exception lastException = null;
+ boolean pingSuccessful = false;
+ while (System.currentTimeMillis() < timeout) {
+ try {
+ final String pingResponse = controller.ping();
+ pingSuccessful = "pong".equals(pingResponse);
Review Comment:
The expectation of `pong` as the value seems a bit buried. Recommend at least promoting this to a static variable or perhaps moving it to an enum to make the expected communication clear. On further consideration, it seems like the `ping()` interface should be changed to return a structured object, or a simple `boolean`, depending on whether the response could contain some useful error information.
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/PythonController.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * 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.python;
+
+import org.apache.nifi.python.processor.PythonProcessorAdapter;
+
+import java.util.List;
+
+/**
+ * An interface that is implemented on the Python side. All method invocations are to be proxied via Py4J.
+ */
+public interface PythonController {
+ /**
+ * A simple test method. If the method is invoked within throwing an Exception, it implies a successful roundtrip invocation
+ */
+ String ping();
Review Comment:
As mentioned in the calling class, it seems like this return type should be changed to a `boolean`, or perhaps a complex object with a `boolean success` property if more context is required.
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcess.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.py4j;
+
+import org.apache.nifi.py4j.client.JavaObjectBindings;
+import org.apache.nifi.py4j.client.NiFiPythonGateway;
+import org.apache.nifi.py4j.client.StandardPythonClient;
+import org.apache.nifi.py4j.server.NiFiGatewayServer;
+import org.apache.nifi.python.ControllerServiceTypeLookup;
+import org.apache.nifi.python.PythonController;
+import org.apache.nifi.python.PythonProcessConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import py4j.CallbackClient;
+import py4j.GatewayServer;
+
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+// TODO / Figure Out for MVP:
+// MUST DO:
+// - Documentation
+// - Admin Guide
+// - JavaDocs
+// - Developer Guide
+// - Explain how communication between Java & Python work.
+// - Java is preferred, Python is slower and more expensive b/c of network
+// - Different Extension Points (FlowFileTransform, RecordTransform)
+// - What the API Looks like, Links to JavaDocs for ProcessContext, etc.
+// - Example Code
+// - Exposing properties
+// - Relationships
+// - Controller Services
+// - Need to update docs to show the interfaces that are exposed, explain how to get these...
+// - Design Doc
+// - Setup proper logging on the Python side: https://docs.python.org/2/howto/logging-cookbook.html#using-file-rotation
+// - For FlowFileTransform, allow the result to contain either a byte array or a String. If a String, just convert in the parent class.
+// - Figure out how to deal with Python Packaging
+// - Need to figure out how to deal with additionalDetails.html, docs directory in python project typically?
+// - Understand how to deal with versioning
+// - Look at performance improvements for Py4J - socket comms appear to be INCREDIBLY slow.
+// - Create test that calls Python 1M times. Just returns 'hello'. See how long it takes
+// - Create test that calls Python 1M times. Returns <java object>.toString() and see how long it takes.
+// - Will help to understand if it's the call from Java to Python that's slow, Python to Java, or both.
+// - Performance concern for TransformRecord
+// - Currently, triggering the transform() method is pretty fast. But then the Result object comes back and we have to call into the Python side to call the getters
+// over and over. Need to look into instead serializing the entire response as JSON and sending that back.
+// - Also, since this is heavy JSON processing, might want to consider ORJSON or something like that instead of inbuilt JSON parser/generator
+// - Test pip install nifi-my-proc, does nifi pick it up?
+// - When ran DetectObjectInImage with multiple threads, Python died. Need to figure out why.
+// - If Python Process dies, need to create a new process and need to then create all of the Processors that were in that Process and initialize them.
+// - Milestone 2 or 3, not Milestone 1.
+// - Remove test-pypi usage from ExtensionManager.py
+// - Additional Interfaces beyond just FlowFileTransform
+// - FlowFileSource
+// - Restructure Maven projects
+// - Should this all go under Framework?
+//
+//
+// CONSIDER:
+// - Clustering: Ensure component on all nodes?
+// - Consider "pip freeze" type of thing to ensure that python dependencies are same across nodes when joining cluster.
+// - Update python code to use python_style_method_names instead of javaStyleMethodNames
+// - Also add 'failure' and 'original' relationships to FlowFileTransform
+//
+//
+// Can punt for now:
+// - We have an issue with objects created from Processor calling into Java. Is fine when we provide objects to Python but when
+// it makes a callback, those objects are bound and never unbound!!!
+// *** This appears to be fine as long as on the Python side we set manage_memory=True ***
+// But it does cut perf in half. May be room for improvement somehow? By implementing a 'bulk delete' custom command?
+// - Appears to still be issues with timeout in nifi when set to 10 secs... initially will probably just leave default of 0sec.
+
+public class PythonProcess {
+ private static final Logger logger = LoggerFactory.getLogger(PythonProcess.class);
+ private static final String PYTHON_CONTROLLER_FILENAME = "Controller.py";
+
+ private final PythonProcessConfig processConfig;
+ private final ControllerServiceTypeLookup controllerServiceTypeLookup;
+ private final File virtualEnvHome;
+ private GatewayServer server;
+ private PythonController controller;
+ private Process process;
+ private NiFiPythonGateway gateway;
+ private final Map<String, Boolean> processorPrefersIsolation = new ConcurrentHashMap<>();
+
+
+ public PythonProcess(final PythonProcessConfig processConfig, final ControllerServiceTypeLookup controllerServiceTypeLookup, final File virtualEnvHome) {
+ this.processConfig = processConfig;
+ this.controllerServiceTypeLookup = controllerServiceTypeLookup;
+ this.virtualEnvHome = virtualEnvHome;
+ }
+
+ public PythonController getController() {
+ return controller;
+ }
+
+ public void start() throws IOException {
+ // TODO: Look into using configured TLS Certs to make this secure by default.
+ final ServerSocketFactory serverSocketFactory = ServerSocketFactory.getDefault();
+ final SocketFactory socketFactory = SocketFactory.getDefault();
+
+ final int timeoutMillis = (int) processConfig.getCommsTimeout().toMillis();
+ final String authToken = null;
+ final CallbackClient callbackClient = new CallbackClient(GatewayServer.DEFAULT_PYTHON_PORT, GatewayServer.defaultAddress(), authToken,
+ 50000L, TimeUnit.MILLISECONDS, socketFactory, false, timeoutMillis);
Review Comment:
Recommend either declaring a static variable for `50000` or adding a comment that it is the Minimum Connection Time. At first it seemed like it could be based on the communication timeout, but as it is completely different, it would be helpful to note what it is here without having to refer to the Py4J documentation. It does not seem like it needs to be a configurable property, so just documenting or making it a static named variable would help.
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/server/NiFiGatewayServer.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * 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.py4j.server;
+
+import py4j.Gateway;
+import py4j.GatewayServer;
+import py4j.Py4JServerConnection;
+import py4j.commands.Command;
+
+import javax.net.ServerSocketFactory;
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.Socket;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * A custom implementation of the GatewayServer that makes more sense for NiFi's use case.
+ * There are only two differences between this implementation and the default GatewayServer implementation:
+ * when a Connection is created, we use a {@link NiFiGatewayConnection} instead of a GatewayConnection,
+ * and when shutdown() is called, we set a boolean flag so that we can provide an isShutdown() method.
+ */
+public class NiFiGatewayServer extends GatewayServer {
+ private volatile boolean shutdown = false;
+
+ public NiFiGatewayServer(final Gateway gateway,
+ final int port,
+ final InetAddress address,
+ final int connectTimeout,
+ final int readTimeout,
+ final List<Class<? extends Command>> customCommands,
+ final ServerSocketFactory sSocketFactory) {
+ super(gateway, port, address, connectTimeout, readTimeout, customCommands, sSocketFactory);
+ }
+
+ protected Py4JServerConnection createConnection(final Gateway gateway, final Socket socket) throws IOException {
+ final NiFiGatewayConnection connection = new NiFiGatewayConnection(this, gateway, socket, authToken, Collections.emptyList(), getListeners());
+ connection.startConnection();
+ return connection;
+ }
+
+ @Override
+ public void shutdown(final boolean shutdownCallbackClient) {
Review Comment:
The default `start()` method creates an new `Thread` instance for running the server, and the thread name is just `Thread-N`, which makes troubleshooting difficult. Overriding the `start(boolean fork)` method would allow customizing the Thread name, which would be very helpful for troubleshooting runtime problems.
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java:
##########
@@ -823,6 +850,80 @@ private static FlowFileRepository createFlowFileRepository(final NiFiProperties
}
}
+
+ private PythonBridge createPythonBridge(final NiFiProperties nifiProperties, final ControllerServiceProvider serviceProvider) {
+ final String pythonCommand = nifiProperties.getProperty(NiFiProperties.PYTHON_COMMAND);
+ if (pythonCommand == null) {
+ LOG.warn("Will not enable Python Extensions because the nifi.python.command property has not been configured in nifi.properties");
Review Comment:
Recommend changing this to an info message since it is the default behavior and does not indicate a problem.
```suggestion
LOG.info("Python Extensions disabled because the property [nifi.python.command] is not configured nifi.properties");
```
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework/src/main/python/framework/ExtensionManager.py:
##########
@@ -0,0 +1,531 @@
+# 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.
+
+import os
+import importlib
+import sys
+import importlib.util # Note requires Python 3.4+
+import inspect
+import logging
+import subprocess
+import ast
+import pkgutil
+from pathlib import Path
+
+logger = logging.getLogger("org.apache.nifi.py4j.ExtensionManager")
+
+# A simple wrapper class to encompass a processor type and its version
+class ExtensionId:
+ def __init__(self, classname=None, version=None):
+ self.classname = classname
+ self.version = version
+
+ def __hash__(self):
+ return hash((self.classname, self.version))
+
+ def __eq__(self, other):
+ return (self.classname, self.version) == (other.classname, other.version)
+
+
+class ExtensionDetails:
+ class Java:
+ implements = ['org.apache.nifi.python.PythonProcessorDetails']
+
+ def __init__(self, gateway, type, version='Unknown', dependencies=None, source_location=None, package_name=None, description=None, tags=None):
+ self.gateway = gateway
+ if dependencies is None:
+ dependencies = []
+ if tags is None:
+ tags = []
+
+ self.type = type
+ self.version = version
+ self.dependencies = dependencies
+ self.source_location = source_location
+ self.package_name = package_name
+ self.description = description
+ self.tags = tags
+
+ def getProcessorType(self):
+ return self.type
+
+ def getProcessorVersion(self):
+ return self.version
+
+ def getSourceLocation(self):
+ return self.source_location
+
+ def getPyPiPackageName(self):
+ return self.package_name
+
+ def getDependencies(self):
+ list = self.gateway.jvm.java.util.ArrayList()
+ for dep in self.dependencies:
+ list.add(dep)
+
+ return list
+
+ def getCapabilityDescription(self):
+ return self.description
+
+ def getTags(self):
+ list = self.gateway.jvm.java.util.ArrayList()
+ for tag in self.tags:
+ list.add(tag)
+
+ return list
+
+
+
+
+class ExtensionManager:
+ """
+ ExtensionManager is responsible for discovery of extensions types and the lifecycle management of those extension types.
+ Discovery of extension types includes finding what extension types are available
+ (e.g., which Processor types exist on the system), as well as information about those extension types, such as
+ the extension's documentation (tags and capability description).
+
+ Lifecycle management includes determining the third-party dependencies that an extension has and ensuring that those
+ third-party dependencies have been imported.
+ """
+
+ processorInterfaces = ['org.apache.nifi.python.processor.FlowFileTransform', 'org.apache.nifi.python.processor.RecordTransform']
+ processor_details = {}
+ processor_class_by_name = {}
+ module_files_by_extension_type = {}
+ dependency_directories = {}
+
+ def __init__(self, gateway):
+ self.gateway = gateway
+
+
+ def getProcessorTypes(self):
+ """
+ :return: a list of Processor types that have been discovered by the #discoverExtensions method
+ """
+ return self.processor_details.values()
+
+ def getProcessorClass(self, type, version, work_dir):
+ """
+ Returns the Python class that can be used to instantiate a processor of the given type.
+ Additionally, it ensures that the required third-party dependencies are on the system path in order to ensure that
+ the necessary libraries are available to the Processor so that it can be instantiated and used.
+
+ :param type: the type of Processor
+ :param version: the version of the Processor
+ :param work_dir: the working directory for extensions
+ :return: the Python class that can be used to instantiate a Processor of the given type and version
+
+ :raises ValueError: if there is no known Processor with the given type and version
+ """
+ id = ExtensionId(classname=type, version=version)
+ if id in self.processor_class_by_name:
+ return self.processor_class_by_name[id]
+
+ if id not in self.module_files_by_extension_type:
+ raise ValueError('Invalid Processor Type: No module is known to contain Processor of type ' + type + ' version ' + version)
+ module_file = self.module_files_by_extension_type[id]
+
+ if id in self.processor_details:
+ extension_working_dir = os.path.join(work_dir, 'extensions', type, version)
+ sys.path.insert(0, extension_working_dir)
+
+ details = self.processor_details[id]
+ processor_class = self.__load_extension_module__(module_file, details.local_dependencies)
+ self.processor_class_by_name[id] = processor_class
+ return processor_class
+
+
+ def reload_processor(self, processor_type, version, work_dir):
+ """
+ Reloads the class definition for the given processor type. This is used in order to ensure that any changes that have
+ been made to the Processor are reloaded and will take effect.
+
+ :param processor_type: the type of the processor whose class definition should be reloaded
+ :param version: the version of the processor whose class definition should be reloaded
+ :param work_dir: the working directory
+ :return: the new class definition
+ """
+ id = ExtensionId(classname=processor_type, version=version)
+
+ # get the python module file that contains the specified processor
+ module_file = self.get_module_file(processor_type, version)
+
+ # Call load_extension to ensure that we load all necessary dependencies, in case they have changed
+ self.__gather_extension_details__(module_file, work_dir)
+
+ # Reload the processor class itself
+ details = self.processor_details[id]
+ processor_class = self.__load_extension_module__(module_file, details.local_dependencies)
+
+ # Update our cache so that when the processor is created again, the new class will be used
+ self.processor_class_by_name[id] = processor_class
+
+
+ def get_module_file(self, processor_type, version):
+ """
+ Returns the module file that contains the source for the given Processor type and version
+ :param processor_type: the Processor type
+ :param version: the version of the Processor
+ :return: the file that contains the source for the given Processor
+
+ :raises ValueError: if no Processor type is known for the given type and version
+ """
+ id = ExtensionId(processor_type, version)
+ if id not in self.module_files_by_extension_type:
+ raise ValueError('Invalid Processor Type: No module is known to contain Processor of type ' + processor_type + ' version ' + version)
+
+ return self.module_files_by_extension_type[id]
+
+
+ # Discover extensions using the 'prefix' method described in
+ # https://packaging.python.org/en/latest/guides/creating-and-discovering-plugins/
+ def discoverExtensions(self, dirs, work_dir):
+ """
+ Discovers any extensions that are available in any of the given directories, as well as any extensions that are available
+ from PyPi repositories
+
+ :param dirs: the directories to check for any local extensions
+ :param work_dir: the working directory
+ """
+ self.__discover_local_extensions__(dirs, work_dir)
+ self.__discover_extensions_from_pypi__(work_dir)
+
+ def __discover_extensions_from_pypi__(self, work_dir):
Review Comment:
I agree, the trailing `__` should be removed from this and other methods. The prefix `__` is sufficient to denote these methods as private.
```suggestion
def __discover_extensions_from_pypi(self, work_dir):
```
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/StandardExtensionDiscoveringManager.java:
##########
@@ -169,6 +181,75 @@ public void discoverExtensions(final Set<Bundle> narBundles, final boolean logDe
}
}
+ public void setPythonBridge(final PythonBridge pythonBridge) {
+ this.pythonBridge = pythonBridge;
+ }
+
+ @Override
+ public void discoverPythonExtensions(final Bundle pythonBundle) {
+ logger.info("Scanning to discover which Python extensions are available and importing any necessary dependencies. This may take a few minutes. See python logs for more details.");
+ final long start = System.currentTimeMillis();
+ pythonBridge.discoverExtensions();
+
+ bundleCoordinateBundleLookup.putIfAbsent(pythonBundle.getBundleDetails().getCoordinate(), pythonBundle);
+
+ final Set<ExtensionDefinition> processorDefinitions = definitionMap.get(Processor.class);
+ final List<PythonProcessorDetails> pythonProcessorDetails = pythonBridge.getProcessorTypes();
+
+ int processorsFound = 0;
+ for (final PythonProcessorDetails details : pythonProcessorDetails) {
+ final BundleDetails bundleDetails = createBundleDetailsWithOverriddenVersion(pythonBundle.getBundleDetails(), details.getProcessorVersion());
+ final Bundle bundle = new Bundle(bundleDetails, pythonBundle.getClassLoader());
+
+ // TODO: This is a workaround because the UI has a bug that causes it not to work properly if the type doesn't have a '.' in it
+ final String className = "python." + details.getProcessorType();
+ final ExtensionDefinition extensionDefinition = new ExtensionDefinition.Builder()
+ .implementationClassName(className)
+ .runtime(ExtensionRuntime.PYTHON)
+ .bundle(bundle)
+ .extensionType(Processor.class)
+ .description(details.getCapabilityDescription())
+ .tags(details.getTags())
+ .version(details.getProcessorVersion())
+ .build();
+
+ final boolean added = processorDefinitions.add(extensionDefinition);
+ if (added) {
+ processorsFound++;
+ final List<Bundle> bundlesForClass = classNameBundleLookup.computeIfAbsent(className, key -> new ArrayList<>());
+ bundlesForClass.add(bundle);
+ bundleCoordinateBundleLookup.putIfAbsent(bundleDetails.getCoordinate(), bundle);
+ logger.info("Discovered Python Processor {}", details.getProcessorType());
+ } else {
+ logger.debug("Python Processor {} is already known", details.getProcessorType());
Review Comment:
```suggestion
logger.debug("Python Processor [{}] already known", details.getProcessorType());
```
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/StandardExtensionDiscoveringManager.java:
##########
@@ -169,6 +181,75 @@ public void discoverExtensions(final Set<Bundle> narBundles, final boolean logDe
}
}
+ public void setPythonBridge(final PythonBridge pythonBridge) {
+ this.pythonBridge = pythonBridge;
+ }
+
+ @Override
+ public void discoverPythonExtensions(final Bundle pythonBundle) {
+ logger.info("Scanning to discover which Python extensions are available and importing any necessary dependencies. This may take a few minutes. See python logs for more details.");
Review Comment:
This log and this method produce a lot of output for standard behavior. Recommend changing this log to `debug`. The "python logs" phrase is not quite clear, mentioning the specific filename would be helpful. I recommend removing the phrase `This may take a few minutes` because it can also be very fast.
```suggestion
logger.debug("Scanning for available Python extensions and dependencies. See [python.log] for details");
```
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/StandardExtensionDiscoveringManager.java:
##########
@@ -169,6 +181,75 @@ public void discoverExtensions(final Set<Bundle> narBundles, final boolean logDe
}
}
+ public void setPythonBridge(final PythonBridge pythonBridge) {
+ this.pythonBridge = pythonBridge;
+ }
+
+ @Override
+ public void discoverPythonExtensions(final Bundle pythonBundle) {
+ logger.info("Scanning to discover which Python extensions are available and importing any necessary dependencies. This may take a few minutes. See python logs for more details.");
+ final long start = System.currentTimeMillis();
+ pythonBridge.discoverExtensions();
+
+ bundleCoordinateBundleLookup.putIfAbsent(pythonBundle.getBundleDetails().getCoordinate(), pythonBundle);
+
+ final Set<ExtensionDefinition> processorDefinitions = definitionMap.get(Processor.class);
+ final List<PythonProcessorDetails> pythonProcessorDetails = pythonBridge.getProcessorTypes();
+
+ int processorsFound = 0;
+ for (final PythonProcessorDetails details : pythonProcessorDetails) {
+ final BundleDetails bundleDetails = createBundleDetailsWithOverriddenVersion(pythonBundle.getBundleDetails(), details.getProcessorVersion());
+ final Bundle bundle = new Bundle(bundleDetails, pythonBundle.getClassLoader());
+
+ // TODO: This is a workaround because the UI has a bug that causes it not to work properly if the type doesn't have a '.' in it
+ final String className = "python." + details.getProcessorType();
+ final ExtensionDefinition extensionDefinition = new ExtensionDefinition.Builder()
+ .implementationClassName(className)
+ .runtime(ExtensionRuntime.PYTHON)
+ .bundle(bundle)
+ .extensionType(Processor.class)
+ .description(details.getCapabilityDescription())
+ .tags(details.getTags())
+ .version(details.getProcessorVersion())
+ .build();
+
+ final boolean added = processorDefinitions.add(extensionDefinition);
+ if (added) {
+ processorsFound++;
+ final List<Bundle> bundlesForClass = classNameBundleLookup.computeIfAbsent(className, key -> new ArrayList<>());
+ bundlesForClass.add(bundle);
+ bundleCoordinateBundleLookup.putIfAbsent(bundleDetails.getCoordinate(), bundle);
+ logger.info("Discovered Python Processor {}", details.getProcessorType());
Review Comment:
Recommend adjusting the wording to parallel the already known log:
```suggestion
logger.info("Python Processor [{}] discovered", details.getProcessorType());
```
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework/src/main/python/framework/Controller.py:
##########
@@ -0,0 +1,102 @@
+# 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.
+
+import os
+import logging
+import ExtensionManager
+from py4j.java_gateway import JavaGateway, CallbackServerParameters, GatewayParameters
+
+import PythonProcessorAdapter
+
+# Initialize logging
+logger = logging.getLogger("org.apache.nifi.py4j.Controller")
+logger.setLevel(logging.INFO)
+
+logging.getLogger("py4j").setLevel(logging.WARN)
+
+logsDir = os.getenv('LOGS_DIR')
+logging.basicConfig(filename=logsDir + '/python.log',
+ format='%(asctime)s %(levelname)s %(name)s %(message)s',
+ encoding='utf-8',
+ level=logging.INFO)
+
+
+class Controller:
+
+ def ping(self):
+ return "pong"
+
+ def getProcessorTypes(self):
+ types = self.extensionManager.getProcessorTypes()
+ typesList = self.gateway.jvm.java.util.ArrayList()
+ for type in types:
+ typesList.add(type)
+ return typesList
+
+ def discoverExtensions(self, dirs, work_dir):
+ self.extensionManager.discoverExtensions(dirs, work_dir)
+
+ def createProcessor(self, processorType, version, work_dir):
+ processorClass = self.extensionManager.getProcessorClass(processorType, version, work_dir)
+ processor = processorClass(jvm=self.gateway.jvm)
+ adapter = PythonProcessorAdapter.PythonProcessorAdapter(self.gateway, processor, self.extensionManager, self.controllerServiceTypeLookup)
+ return adapter
+
+ def reloadProcessor(self, processorType, version, workDirectory):
+ self.extensionManager.reload_processor(processorType, version, workDirectory)
+
+ def getModuleFile(self, processorType, version):
+ module_file = self.extensionManager.get_module_file(processorType, version)
+ return module_file
+
+ def getProcessorDependencies(self, processorType, version):
+ deps = self.extensionManager.__get_dependencies_for_extension_type__(processorType, version)
+ dependencyList = self.gateway.jvm.java.util.ArrayList()
+ for dep in deps:
+ dependencyList.add(dep)
+
+ return dependencyList
+
+ def setGateway(self, gateway):
+ self.gateway = gateway
+ self.extensionManager = ExtensionManager.ExtensionManager(gateway)
+
+ def setControllerServiceTypeLookup(self, typeLookup):
+ self.controllerServiceTypeLookup = typeLookup
+
+ class Java:
+ implements = ["org.apache.nifi.py4j.PythonController"]
+
+
+if __name__ == "__main__":
+ # Create the Controller
+ controller = Controller()
+
+ # Create the Java Gateway for communicating with NiFi Java process
+ java_port = int(os.getenv('JAVA_PORT'))
+ gateway = JavaGateway(
+ callback_server_parameters=CallbackServerParameters(port=0),
Review Comment:
It looks like this is where the authentication token could be pulled from an environment variable and passed to the CallbackServerParameters.
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/client/PythonProxyInvocationHandler.java:
##########
@@ -0,0 +1,174 @@
+/*
+ * 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.py4j.client;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import py4j.Protocol;
+import py4j.Py4JException;
+import py4j.StringUtil;
+import py4j.reflection.MethodInvoker;
+import py4j.reflection.ReflectionEngine;
+import py4j.reflection.TypeConverter;
+
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Method;
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public class PythonProxyInvocationHandler implements InvocationHandler {
+ public static final String VOID_ARGUMENT_LINE = String.valueOf(Protocol.VOID);
+ public static final String METHOD_INVOCATION_START_CLAUSE = "c\n";
+ public static final String METHOD_INVOCATION_END_CLAUSE = "e\n";
+ private static final Logger logger = LoggerFactory.getLogger(PythonProxyInvocationHandler.class);
+
+ private final String objectId;
+ private final NiFiPythonGateway gateway;
+ private final JavaObjectBindings bindings;
+
+ public PythonProxyInvocationHandler(final NiFiPythonGateway gateway, final String objectId) {
+ this.objectId = objectId;
+ this.gateway = gateway;
+ this.bindings = gateway.getObjectBindings();
+ }
+
+ @Override
+ public Object invoke(final Object proxy, final Method method, final Object[] args) throws Throwable {
+ if (args == null && method.getName().equals("toString")) {
+ return "PythonProxy[targetObjectId=" + objectId + "]";
+ }
+
+ final CommandBuilder commandBuilder = new CommandBuilder(bindings, objectId, method.getName());
+ final String command = commandBuilder.buildCommand(args);
+
+ if (logger.isDebugEnabled()) {
+ final List<Object> argList = args == null ? Collections.emptyList() : Arrays.asList(args);
+ logger.debug("Invoking {} on {} with args {} using command {}", method, proxy, argList, command);
+ }
+
+ gateway.beginInvocation(this.objectId, method, args);
+
+ final String response = gateway.getCallbackClient().sendCommand(command);
+ final Object output = Protocol.getReturnValue(response, gateway);
+ final Object convertedOutput = convertOutput(method, output);
+
+ // TODO: While we're waiting for this call to return, the Python side may call back into the Java side and create objects.
+ // When that happens we need to throw those onto the stack also!
+ if (gateway.isUnbind(method)) {
+ commandBuilder.getBoundIds().forEach(bindings::unbind);
+ commandBuilder.getBoundIds().forEach(i -> logger.debug("For method invocation {} unbound {} (from command builder)", method.getName(), i));
+ } else {
+ commandBuilder.getBoundIds().forEach(i -> logger.debug("For method invocation {} will not unbind {} (from command builder) because arguments of this method are not to be unbound",
+ method.getName(), i));
+ }
+
+ gateway.endInvocation(this.objectId, method, args);
+
+ return convertedOutput;
+ }
+
+
+ private String buildInvocationCommand(final String methodName, final Object[] args) {
+ final StringBuilder sb = new StringBuilder();
+ sb.append(METHOD_INVOCATION_START_CLAUSE);
+ sb.append(objectId).append("\n");
+ sb.append(methodName).append("\n");
+
+ if (args != null) {
+ for (final Object arg : args) {
+ final String argumentLine = getArgumentLine(arg);
+ sb.append(argumentLine).append("\n");
+ }
+ }
+
+ sb.append(METHOD_INVOCATION_END_CLAUSE);
+ return sb.toString();
+ }
+
+ private String getArgumentLine(final Object arg) {
Review Comment:
This method appears to be the same as the one in `CommandBuilder`. Given the importance of proper argument formatting, can these implementations be pulled out and reused?
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/client/NiFiPythonGateway.java:
##########
@@ -0,0 +1,203 @@
+/*
+ * 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.py4j.client;
+
+import org.apache.nifi.python.processor.PreserveJavaBinding;
+import org.apache.nifi.python.processor.PythonProcessor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import py4j.CallbackClient;
+import py4j.Gateway;
+import py4j.reflection.PythonProxyHandler;
+
+import java.lang.reflect.Method;
+import java.net.InetAddress;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Stack;
+import java.util.concurrent.ConcurrentHashMap;
+
+
+// TODO: Switch to thread-local instead of a ConcurrentHashMap<Long>...
+// TODO: Document what this is / why it exists
Review Comment:
Are these TODO notes for this PR, or subsequent work? It would be helpful to at least add a sentence or two on why this class exists, and then the ThreadLocal change could come later.
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework/src/main/python/framework/ExtensionManager.py:
##########
@@ -0,0 +1,531 @@
+# 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.
+
+import os
+import importlib
+import sys
+import importlib.util # Note requires Python 3.4+
+import inspect
+import logging
+import subprocess
+import ast
+import pkgutil
+from pathlib import Path
+
+logger = logging.getLogger("org.apache.nifi.py4j.ExtensionManager")
+
+# A simple wrapper class to encompass a processor type and its version
+class ExtensionId:
+ def __init__(self, classname=None, version=None):
+ self.classname = classname
+ self.version = version
+
+ def __hash__(self):
+ return hash((self.classname, self.version))
+
+ def __eq__(self, other):
+ return (self.classname, self.version) == (other.classname, other.version)
+
+
+class ExtensionDetails:
+ class Java:
+ implements = ['org.apache.nifi.python.PythonProcessorDetails']
+
+ def __init__(self, gateway, type, version='Unknown', dependencies=None, source_location=None, package_name=None, description=None, tags=None):
+ self.gateway = gateway
+ if dependencies is None:
+ dependencies = []
+ if tags is None:
+ tags = []
+
+ self.type = type
+ self.version = version
+ self.dependencies = dependencies
+ self.source_location = source_location
+ self.package_name = package_name
+ self.description = description
+ self.tags = tags
+
+ def getProcessorType(self):
+ return self.type
+
+ def getProcessorVersion(self):
+ return self.version
+
+ def getSourceLocation(self):
+ return self.source_location
+
+ def getPyPiPackageName(self):
+ return self.package_name
+
+ def getDependencies(self):
+ list = self.gateway.jvm.java.util.ArrayList()
+ for dep in self.dependencies:
+ list.add(dep)
+
+ return list
+
+ def getCapabilityDescription(self):
+ return self.description
+
+ def getTags(self):
+ list = self.gateway.jvm.java.util.ArrayList()
+ for tag in self.tags:
+ list.add(tag)
+
+ return list
+
+
+
+
+class ExtensionManager:
+ """
+ ExtensionManager is responsible for discovery of extensions types and the lifecycle management of those extension types.
+ Discovery of extension types includes finding what extension types are available
+ (e.g., which Processor types exist on the system), as well as information about those extension types, such as
+ the extension's documentation (tags and capability description).
+
+ Lifecycle management includes determining the third-party dependencies that an extension has and ensuring that those
+ third-party dependencies have been imported.
+ """
+
+ processorInterfaces = ['org.apache.nifi.python.processor.FlowFileTransform', 'org.apache.nifi.python.processor.RecordTransform']
+ processor_details = {}
+ processor_class_by_name = {}
+ module_files_by_extension_type = {}
+ dependency_directories = {}
+
+ def __init__(self, gateway):
+ self.gateway = gateway
+
+
+ def getProcessorTypes(self):
+ """
+ :return: a list of Processor types that have been discovered by the #discoverExtensions method
+ """
+ return self.processor_details.values()
+
+ def getProcessorClass(self, type, version, work_dir):
+ """
+ Returns the Python class that can be used to instantiate a processor of the given type.
+ Additionally, it ensures that the required third-party dependencies are on the system path in order to ensure that
+ the necessary libraries are available to the Processor so that it can be instantiated and used.
+
+ :param type: the type of Processor
+ :param version: the version of the Processor
+ :param work_dir: the working directory for extensions
+ :return: the Python class that can be used to instantiate a Processor of the given type and version
+
+ :raises ValueError: if there is no known Processor with the given type and version
+ """
+ id = ExtensionId(classname=type, version=version)
+ if id in self.processor_class_by_name:
+ return self.processor_class_by_name[id]
+
+ if id not in self.module_files_by_extension_type:
+ raise ValueError('Invalid Processor Type: No module is known to contain Processor of type ' + type + ' version ' + version)
+ module_file = self.module_files_by_extension_type[id]
+
+ if id in self.processor_details:
+ extension_working_dir = os.path.join(work_dir, 'extensions', type, version)
+ sys.path.insert(0, extension_working_dir)
+
+ details = self.processor_details[id]
+ processor_class = self.__load_extension_module__(module_file, details.local_dependencies)
+ self.processor_class_by_name[id] = processor_class
+ return processor_class
+
+
+ def reload_processor(self, processor_type, version, work_dir):
+ """
+ Reloads the class definition for the given processor type. This is used in order to ensure that any changes that have
+ been made to the Processor are reloaded and will take effect.
+
+ :param processor_type: the type of the processor whose class definition should be reloaded
+ :param version: the version of the processor whose class definition should be reloaded
+ :param work_dir: the working directory
+ :return: the new class definition
+ """
+ id = ExtensionId(classname=processor_type, version=version)
+
+ # get the python module file that contains the specified processor
+ module_file = self.get_module_file(processor_type, version)
+
+ # Call load_extension to ensure that we load all necessary dependencies, in case they have changed
+ self.__gather_extension_details__(module_file, work_dir)
+
+ # Reload the processor class itself
+ details = self.processor_details[id]
+ processor_class = self.__load_extension_module__(module_file, details.local_dependencies)
+
+ # Update our cache so that when the processor is created again, the new class will be used
+ self.processor_class_by_name[id] = processor_class
+
+
+ def get_module_file(self, processor_type, version):
+ """
+ Returns the module file that contains the source for the given Processor type and version
+ :param processor_type: the Processor type
+ :param version: the version of the Processor
+ :return: the file that contains the source for the given Processor
+
+ :raises ValueError: if no Processor type is known for the given type and version
+ """
+ id = ExtensionId(processor_type, version)
+ if id not in self.module_files_by_extension_type:
+ raise ValueError('Invalid Processor Type: No module is known to contain Processor of type ' + processor_type + ' version ' + version)
+
+ return self.module_files_by_extension_type[id]
+
+
+ # Discover extensions using the 'prefix' method described in
+ # https://packaging.python.org/en/latest/guides/creating-and-discovering-plugins/
+ def discoverExtensions(self, dirs, work_dir):
+ """
+ Discovers any extensions that are available in any of the given directories, as well as any extensions that are available
+ from PyPi repositories
+
+ :param dirs: the directories to check for any local extensions
+ :param work_dir: the working directory
+ """
+ self.__discover_local_extensions__(dirs, work_dir)
+ self.__discover_extensions_from_pypi__(work_dir)
+
+ def __discover_extensions_from_pypi__(self, work_dir):
+ self.__discover_extensions_from_paths__(None, work_dir, True)
+
+ def __discover_local_extensions__(self, dirs, work_dir):
+ self.__discover_extensions_from_paths__(dirs, work_dir, False)
+
+
+ def __discover_extensions_from_paths__(self, paths, work_dir, require_nifi_prefix):
+ for finder, name, ispkg in pkgutil.iter_modules(paths):
+ if not require_nifi_prefix or name.startswith('nifi_'):
+ module_file = '<Unknown Module File>'
+ try:
+ module = finder.find_module(name)
+ module_file = module.path
+ logger.info('Discovered extension %s' % module_file)
+
+ self.__gather_extension_details__(module_file, work_dir)
+ except Exception:
+ logger.error("Failed to load Python extensions from module file {0}. This module will be ignored.".format(module_file), exc_info=True)
+
+
+
+ def __gather_extension_details__(self, module_file, work_dir, local_dependencies=None):
+ path = Path(module_file)
+ basename = os.path.basename(module_file)
+
+ # If no local_dependencies have been provided, check to see if there are any.
+ # We consider any Python module in the same directory as a local dependency
+ if local_dependencies is None:
+ local_dependencies = []
+
+ if basename == '__init__.py':
+ dir = path.parent
+ for filename in os.listdir(dir):
+ if not filename.endswith('.py'):
+ continue
+ if filename == '__init__.py':
+ continue
+
+ child_module_file = os.path.join(dir, filename)
+ local_dependencies.append(child_module_file)
+
+ # If the module file is an __init__.py file, we check all module files in the same directory.
+ if basename == '__init__.py':
+ dir = path.parent
+ for filename in os.listdir(dir):
+ if not filename.endswith('.py'):
+ continue
+ if filename == '__init__.py':
+ continue
+
+ child_module_file = os.path.join(dir, filename)
+ self.__gather_extension_details__(child_module_file, work_dir, local_dependencies=local_dependencies)
+
+ classes_and_details = self.__get_processor_classes_and_details__(module_file)
+ for classname, details in classes_and_details.items():
+ id = ExtensionId(classname, details.version)
+ logger.info("Found local dependencies {0} for {1}".format(local_dependencies, classname))
+
+ details.local_dependencies = local_dependencies
+
+ # Add class name to processor types only if not there already
+ if id not in self.processor_details:
+ self.processor_details[id] = details
+
+ self.module_files_by_extension_type[id] = module_file
+
+ # If there are any dependencies, use pip to install them
+ extension_working_dir = os.path.join(work_dir, 'extensions', classname, details.version)
+ self.__import_external_dependencies__(classname, details.dependencies, extension_working_dir)
+
+
+ def __get_dependencies_for_extension_type__(self, extension_type, version):
+ id = ExtensionId(extension_type, version)
+ return self.processor_details[id].dependencies
+
+ def __get_processor_classes_and_details__(self, module_file):
+ # Parse the python file
+ with open(module_file) as file:
+ root_node = ast.parse(file.read())
+
+ details_by_class = {}
+
+ # Get top-level class nodes (e.g., MyProcessor)
+ class_nodes = self.__get_class_nodes__(root_node)
+
+ for class_node in class_nodes:
+ logger.debug("Checking if class %s is a processor" % class_node.name)
+ if self.__is_processor_class_node__(class_node):
+ logger.info("Discovered Processor class {0} in module {1}".format(class_node.name, module_file))
+ details = self.__get_processor_details__(class_node, module_file)
+ details_by_class[class_node.name] = details
+
+ return details_by_class
+
+
+ def __is_processor_class_node__(self, class_node):
+ """
+ Checks if the Abstract Syntax Tree (AST) Node represents a Processor class.
+ We are looking for any classes within the given module file that look like:
+
+ class MyProcessor:
+ ...
+ class Java:
+ implements = ['org.apache.nifi.python.processor.FlowFileTransform']
+
+ :param class_node: the abstract syntax tree (AST) node
+ :return: True if the AST Node represents a Python Class that is a Processor, False otherwise
+ """
+
+ # Look for a 'Java' sub-class
+ child_class_nodes = self.__get_class_nodes__(class_node)
+
+ for child_class_node in child_class_nodes:
+ if child_class_node.name == 'Java':
+ # Look for an assignment that assigns values to the `implements` keyword
+ assignment_nodes = self.__get_assignment_nodes__(child_class_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == 'implements'):
+ assigned_values = assignment_node.value.elts
+ for assigned_value in assigned_values:
+ if assigned_value.value in self.processorInterfaces:
+ return True
+ return False
+
+
+ def __get_processor_details__(self, class_node, module_file):
+ # Look for a 'ProcessorDetails' class
+ child_class_nodes = self.__get_class_nodes__(class_node)
+
+ for child_class_node in child_class_nodes:
+ if child_class_node.name == 'ProcessorDetails':
+ logger.debug('Found ProcessorDetails class in %s' % class_node.name)
+ version = self.__get_processor_version__(child_class_node, class_node.name)
+ dependencies = self.__get_processor_dependencies__(child_class_node, class_node.name)
+ description = self.__get_processor_description__(child_class_node, class_node.name)
+ tags = self.__get_processor_tags__(child_class_node, class_node.name)
+
+ return ExtensionDetails(gateway=self.gateway,
+ type=class_node.name,
+ version=version,
+ dependencies=dependencies,
+ source_location=module_file,
+ description=description,
+ tags=tags)
+
+ return ExtensionDetails(gateway=self.gateway,
+ type=class_node.name,
+ version='Unknown',
+ dependencies=[],
+ source_location=module_file)
+
+
+ def __get_processor_version__(self, details_node, class_name):
+ assignment_nodes = self.__get_assignment_nodes__(details_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == 'version'):
+ assigned_values = assignment_node.value.value
+ logger.info("Found version of {0} to be {1}".format(class_name, assigned_values))
+ return assigned_values
+
+ # No dependencies found
+ logger.info("Found no version information for {0}".format(class_name))
+ return 'Unknown'
+
+
+ def __get_processor_dependencies__(self, details_node, class_name):
+ deps = self.__get_assigned_list__(details_node, class_name, 'dependencies')
+ if len(deps) == 0:
+ logger.info("Found no external dependencies that are required for class %s" % class_name)
+ else:
+ logger.info("Found the following external dependencies that are required for class {0}: {1}".format(class_name, deps))
+
+ return deps
+
+
+ def __get_processor_tags__(self, details_node, class_name):
+ return self.__get_assigned_list__(details_node, class_name, 'tags')
+
+
+ def __get_assigned_list__(self, details_node, class_name, element_name):
+ assignment_nodes = self.__get_assignment_nodes__(details_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == element_name):
+ assigned_values = assignment_node.value.elts
+ declared_dependencies = []
+ for assigned_value in assigned_values:
+ declared_dependencies.append(assigned_value.value)
+
+ return declared_dependencies
+
+ # No values found
+ return []
+
+
+ def __get_processor_description__(self, details_node, class_name):
+ assignment_nodes = self.__get_assignment_nodes__(details_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == 'description'):
+ return assignment_node.value.value
+
+ # No description found
+ logger.debug("Found no description for class %s" % class_name)
+ return None
+
+
+
+ def __get_class_nodes__(self, node):
+ class_nodes = [n for n in node.body if isinstance(n, ast.ClassDef)]
+ return class_nodes
+
+
+ def __get_assignment_nodes__(self, node):
+ assignment_nodes = [n for n in node.body if isinstance(n, ast.Assign)]
+ return assignment_nodes
+
+
+ def __import_external_dependencies__(self, class_name, dependencies, target_dir):
+ if not os.path.exists(target_dir):
+ os.makedirs(target_dir)
+
+ completion_marker_file = os.path.join(target_dir, 'dependency-download.complete')
Review Comment:
Does this block changing dependencies when reloading a Processor at runtime?
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcess.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.py4j;
+
+import org.apache.nifi.py4j.client.JavaObjectBindings;
+import org.apache.nifi.py4j.client.NiFiPythonGateway;
+import org.apache.nifi.py4j.client.StandardPythonClient;
+import org.apache.nifi.py4j.server.NiFiGatewayServer;
+import org.apache.nifi.python.ControllerServiceTypeLookup;
+import org.apache.nifi.python.PythonController;
+import org.apache.nifi.python.PythonProcessConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import py4j.CallbackClient;
+import py4j.GatewayServer;
+
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+// TODO / Figure Out for MVP:
+// MUST DO:
+// - Documentation
+// - Admin Guide
+// - JavaDocs
+// - Developer Guide
+// - Explain how communication between Java & Python work.
+// - Java is preferred, Python is slower and more expensive b/c of network
+// - Different Extension Points (FlowFileTransform, RecordTransform)
+// - What the API Looks like, Links to JavaDocs for ProcessContext, etc.
+// - Example Code
+// - Exposing properties
+// - Relationships
+// - Controller Services
+// - Need to update docs to show the interfaces that are exposed, explain how to get these...
+// - Design Doc
+// - Setup proper logging on the Python side: https://docs.python.org/2/howto/logging-cookbook.html#using-file-rotation
+// - For FlowFileTransform, allow the result to contain either a byte array or a String. If a String, just convert in the parent class.
+// - Figure out how to deal with Python Packaging
+// - Need to figure out how to deal with additionalDetails.html, docs directory in python project typically?
+// - Understand how to deal with versioning
+// - Look at performance improvements for Py4J - socket comms appear to be INCREDIBLY slow.
+// - Create test that calls Python 1M times. Just returns 'hello'. See how long it takes
+// - Create test that calls Python 1M times. Returns <java object>.toString() and see how long it takes.
+// - Will help to understand if it's the call from Java to Python that's slow, Python to Java, or both.
+// - Performance concern for TransformRecord
+// - Currently, triggering the transform() method is pretty fast. But then the Result object comes back and we have to call into the Python side to call the getters
+// over and over. Need to look into instead serializing the entire response as JSON and sending that back.
+// - Also, since this is heavy JSON processing, might want to consider ORJSON or something like that instead of inbuilt JSON parser/generator
+// - Test pip install nifi-my-proc, does nifi pick it up?
+// - When ran DetectObjectInImage with multiple threads, Python died. Need to figure out why.
+// - If Python Process dies, need to create a new process and need to then create all of the Processors that were in that Process and initialize them.
+// - Milestone 2 or 3, not Milestone 1.
+// - Remove test-pypi usage from ExtensionManager.py
+// - Additional Interfaces beyond just FlowFileTransform
+// - FlowFileSource
+// - Restructure Maven projects
+// - Should this all go under Framework?
+//
+//
+// CONSIDER:
+// - Clustering: Ensure component on all nodes?
+// - Consider "pip freeze" type of thing to ensure that python dependencies are same across nodes when joining cluster.
+// - Update python code to use python_style_method_names instead of javaStyleMethodNames
+// - Also add 'failure' and 'original' relationships to FlowFileTransform
+//
+//
+// Can punt for now:
+// - We have an issue with objects created from Processor calling into Java. Is fine when we provide objects to Python but when
+// it makes a callback, those objects are bound and never unbound!!!
+// *** This appears to be fine as long as on the Python side we set manage_memory=True ***
+// But it does cut perf in half. May be room for improvement somehow? By implementing a 'bulk delete' custom command?
+// - Appears to still be issues with timeout in nifi when set to 10 secs... initially will probably just leave default of 0sec.
+
+public class PythonProcess {
+ private static final Logger logger = LoggerFactory.getLogger(PythonProcess.class);
+ private static final String PYTHON_CONTROLLER_FILENAME = "Controller.py";
+
+ private final PythonProcessConfig processConfig;
+ private final ControllerServiceTypeLookup controllerServiceTypeLookup;
+ private final File virtualEnvHome;
+ private GatewayServer server;
+ private PythonController controller;
+ private Process process;
+ private NiFiPythonGateway gateway;
+ private final Map<String, Boolean> processorPrefersIsolation = new ConcurrentHashMap<>();
+
+
+ public PythonProcess(final PythonProcessConfig processConfig, final ControllerServiceTypeLookup controllerServiceTypeLookup, final File virtualEnvHome) {
+ this.processConfig = processConfig;
+ this.controllerServiceTypeLookup = controllerServiceTypeLookup;
+ this.virtualEnvHome = virtualEnvHome;
+ }
+
+ public PythonController getController() {
+ return controller;
+ }
+
+ public void start() throws IOException {
+ // TODO: Look into using configured TLS Certs to make this secure by default.
+ final ServerSocketFactory serverSocketFactory = ServerSocketFactory.getDefault();
+ final SocketFactory socketFactory = SocketFactory.getDefault();
+
+ final int timeoutMillis = (int) processConfig.getCommsTimeout().toMillis();
+ final String authToken = null;
+ final CallbackClient callbackClient = new CallbackClient(GatewayServer.DEFAULT_PYTHON_PORT, GatewayServer.defaultAddress(), authToken,
+ 50000L, TimeUnit.MILLISECONDS, socketFactory, false, timeoutMillis);
+
+ final JavaObjectBindings bindings = new JavaObjectBindings();
+ gateway = new NiFiPythonGateway(bindings, null, callbackClient);
+ gateway.startup();
+
+ server = new NiFiGatewayServer(gateway,
+ 0,
+ GatewayServer.defaultAddress(),
+ timeoutMillis,
+ timeoutMillis,
+ Collections.emptyList(),
+ serverSocketFactory);
+ server.start();
+
+ final int listeningPort = server.getListeningPort();
+
+ setupEnvironment();
+ this.process = launchPythonProcess(listeningPort);
+
+ final StandardPythonClient pythonClient = new StandardPythonClient(gateway);
+ controller = pythonClient.getController();
+
+ final long timeout = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(60L);
+ Exception lastException = null;
+ boolean pingSuccessful = false;
+ while (System.currentTimeMillis() < timeout) {
+ try {
+ final String pingResponse = controller.ping();
+ pingSuccessful = "pong".equals(pingResponse);
+ if (pingSuccessful) {
+ break;
+ } else {
+ logger.debug("Got unexpected response from Py4J Server during ping: {}", pingResponse);
+ }
+ } catch (final Exception e) {
+ lastException = e;
+ logger.debug("Failed to start Py4J Server", e);
+ }
+
+ try {
+ Thread.sleep(50L);
+ } catch (final InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+
+ if (!pingSuccessful && lastException != null) {
+ throw new RuntimeException("Failed to start Python Bridge", lastException);
+ }
+
+ controller.setControllerServiceTypeLookup(controllerServiceTypeLookup);
+ logger.info("Successfully started and pinged Python Server. Python Process = {}", process);
+ }
+
+ private Process launchPythonProcess(final int listeningPort) throws IOException {
+ final File pythonFrameworkDirectory = processConfig.getPythonFrameworkDirectory();
+ final File pythonApiDirectory = new File(pythonFrameworkDirectory.getParentFile(), "api");
+ final File pythonLogsDirectory = processConfig.getPythonLogsDirectory();
+ final File pythonCmdFile = new File(processConfig.getPythonCommand());
+ final String pythonCmd = pythonCmdFile.getName();
+ final File pythonCommandFile = new File(virtualEnvHome, "bin/" + pythonCmd);
+ final String pythonCommand = pythonCommandFile.getAbsolutePath();
+
+ final File controllerPyFile = new File(pythonFrameworkDirectory, PYTHON_CONTROLLER_FILENAME);
+ final ProcessBuilder processBuilder = new ProcessBuilder(pythonCommand, controllerPyFile.getAbsolutePath());
+ processBuilder.environment().put("JAVA_PORT", String.valueOf(listeningPort));
+ processBuilder.environment().put("LOGS_DIR", pythonLogsDirectory.getAbsolutePath());
+ processBuilder.environment().put("ENV_HOME", virtualEnvHome.getAbsolutePath());
+ processBuilder.environment().put("PYTHONPATH", pythonApiDirectory.getAbsolutePath());
+ processBuilder.environment().put("PYTHON_CMD", pythonCommandFile.getAbsolutePath());
+ processBuilder.inheritIO();
+
+ logger.info("Launching Python Process {} {} with working directory {} to communicate with Java on Port {}",
+ pythonCommand, controllerPyFile.getAbsolutePath(), virtualEnvHome, listeningPort);
+ return processBuilder.start();
+ }
+
+
+ private void setupEnvironment() throws IOException {
+ final File environmentCreationCompleteFile = new File(virtualEnvHome, "env-creation-complete.txt");
+ if (environmentCreationCompleteFile.exists()) {
+ logger.debug("Environment has already been created for {}; will not recreate", virtualEnvHome);
+ return;
+ }
+
+ logger.info("Creating Python Virtual Environment {}", virtualEnvHome);
+
+ Files.createDirectories(virtualEnvHome.toPath());
+
+ final String pythonCommand = processConfig.getPythonCommand();
+ final String environmentPath = virtualEnvHome.getAbsolutePath();
+
+ final ProcessBuilder processBuilder = new ProcessBuilder(pythonCommand, "-m", "venv", environmentPath);
+ processBuilder.directory(virtualEnvHome.getParentFile());
+
+ final String command = String.join(" ", processBuilder.command());
+ logger.debug("Creating Python Virtual Environment {} using command {}", virtualEnvHome, command);
+ final Process process = processBuilder.start();
+
+ final int result;
+ try {
+ result = process.waitFor();
+ } catch (final InterruptedException e) {
+ throw new IOException("Interrupted while waiting for Python virtual environment to be created");
+ }
+
+ if (result != 0) {
+ throw new IOException("Failed to create Python Environment " + virtualEnvHome + ": process existed with code " + result);
+ }
+
+ // Create file so that we don't keep trying to recreate the virtual environment
+ environmentCreationCompleteFile.createNewFile();
+ logger.info("Successfully created Python Virtual Environment {}", virtualEnvHome);
+ }
Review Comment:
The `launchPythonProcess()` and `setupEnvironment()` methods include a lot of specialized knowledge about the environment variables and command arguments. I recommend pulling these out to interfaces with implementations that could be tested in isolation. That should help define the expected behavior of these spawned processes.
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/client/CommandBuilder.java:
##########
@@ -0,0 +1,119 @@
+/*
+ * 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.py4j.client;
+
+import py4j.Protocol;
+import py4j.StringUtil;
+import py4j.reflection.ReflectionEngine;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+public class CommandBuilder {
Review Comment:
Given the importance of this class, having a unit test that exercises the various argument types would be very helpful to ensure expected formatting behavior.
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework/src/main/python/framework/ExtensionManager.py:
##########
@@ -0,0 +1,531 @@
+# 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.
+
+import os
+import importlib
+import sys
+import importlib.util # Note requires Python 3.4+
+import inspect
+import logging
+import subprocess
+import ast
+import pkgutil
+from pathlib import Path
+
+logger = logging.getLogger("org.apache.nifi.py4j.ExtensionManager")
+
+# A simple wrapper class to encompass a processor type and its version
+class ExtensionId:
+ def __init__(self, classname=None, version=None):
+ self.classname = classname
+ self.version = version
+
+ def __hash__(self):
+ return hash((self.classname, self.version))
+
+ def __eq__(self, other):
+ return (self.classname, self.version) == (other.classname, other.version)
+
+
+class ExtensionDetails:
+ class Java:
+ implements = ['org.apache.nifi.python.PythonProcessorDetails']
+
+ def __init__(self, gateway, type, version='Unknown', dependencies=None, source_location=None, package_name=None, description=None, tags=None):
+ self.gateway = gateway
+ if dependencies is None:
+ dependencies = []
+ if tags is None:
+ tags = []
+
+ self.type = type
+ self.version = version
+ self.dependencies = dependencies
+ self.source_location = source_location
+ self.package_name = package_name
+ self.description = description
+ self.tags = tags
+
+ def getProcessorType(self):
+ return self.type
+
+ def getProcessorVersion(self):
+ return self.version
+
+ def getSourceLocation(self):
+ return self.source_location
+
+ def getPyPiPackageName(self):
+ return self.package_name
+
+ def getDependencies(self):
+ list = self.gateway.jvm.java.util.ArrayList()
+ for dep in self.dependencies:
+ list.add(dep)
+
+ return list
+
+ def getCapabilityDescription(self):
+ return self.description
+
+ def getTags(self):
+ list = self.gateway.jvm.java.util.ArrayList()
+ for tag in self.tags:
+ list.add(tag)
+
+ return list
+
+
+
+
+class ExtensionManager:
+ """
+ ExtensionManager is responsible for discovery of extensions types and the lifecycle management of those extension types.
+ Discovery of extension types includes finding what extension types are available
+ (e.g., which Processor types exist on the system), as well as information about those extension types, such as
+ the extension's documentation (tags and capability description).
+
+ Lifecycle management includes determining the third-party dependencies that an extension has and ensuring that those
+ third-party dependencies have been imported.
+ """
+
+ processorInterfaces = ['org.apache.nifi.python.processor.FlowFileTransform', 'org.apache.nifi.python.processor.RecordTransform']
+ processor_details = {}
+ processor_class_by_name = {}
+ module_files_by_extension_type = {}
+ dependency_directories = {}
+
+ def __init__(self, gateway):
+ self.gateway = gateway
+
+
+ def getProcessorTypes(self):
+ """
+ :return: a list of Processor types that have been discovered by the #discoverExtensions method
+ """
+ return self.processor_details.values()
+
+ def getProcessorClass(self, type, version, work_dir):
+ """
+ Returns the Python class that can be used to instantiate a processor of the given type.
+ Additionally, it ensures that the required third-party dependencies are on the system path in order to ensure that
+ the necessary libraries are available to the Processor so that it can be instantiated and used.
+
+ :param type: the type of Processor
+ :param version: the version of the Processor
+ :param work_dir: the working directory for extensions
+ :return: the Python class that can be used to instantiate a Processor of the given type and version
+
+ :raises ValueError: if there is no known Processor with the given type and version
+ """
+ id = ExtensionId(classname=type, version=version)
+ if id in self.processor_class_by_name:
+ return self.processor_class_by_name[id]
+
+ if id not in self.module_files_by_extension_type:
+ raise ValueError('Invalid Processor Type: No module is known to contain Processor of type ' + type + ' version ' + version)
+ module_file = self.module_files_by_extension_type[id]
+
+ if id in self.processor_details:
+ extension_working_dir = os.path.join(work_dir, 'extensions', type, version)
+ sys.path.insert(0, extension_working_dir)
+
+ details = self.processor_details[id]
+ processor_class = self.__load_extension_module__(module_file, details.local_dependencies)
+ self.processor_class_by_name[id] = processor_class
+ return processor_class
+
+
+ def reload_processor(self, processor_type, version, work_dir):
+ """
+ Reloads the class definition for the given processor type. This is used in order to ensure that any changes that have
+ been made to the Processor are reloaded and will take effect.
+
+ :param processor_type: the type of the processor whose class definition should be reloaded
+ :param version: the version of the processor whose class definition should be reloaded
+ :param work_dir: the working directory
+ :return: the new class definition
+ """
+ id = ExtensionId(classname=processor_type, version=version)
+
+ # get the python module file that contains the specified processor
+ module_file = self.get_module_file(processor_type, version)
+
+ # Call load_extension to ensure that we load all necessary dependencies, in case they have changed
+ self.__gather_extension_details__(module_file, work_dir)
+
+ # Reload the processor class itself
+ details = self.processor_details[id]
+ processor_class = self.__load_extension_module__(module_file, details.local_dependencies)
+
+ # Update our cache so that when the processor is created again, the new class will be used
+ self.processor_class_by_name[id] = processor_class
+
+
+ def get_module_file(self, processor_type, version):
+ """
+ Returns the module file that contains the source for the given Processor type and version
+ :param processor_type: the Processor type
+ :param version: the version of the Processor
+ :return: the file that contains the source for the given Processor
+
+ :raises ValueError: if no Processor type is known for the given type and version
+ """
+ id = ExtensionId(processor_type, version)
+ if id not in self.module_files_by_extension_type:
+ raise ValueError('Invalid Processor Type: No module is known to contain Processor of type ' + processor_type + ' version ' + version)
+
+ return self.module_files_by_extension_type[id]
+
+
+ # Discover extensions using the 'prefix' method described in
+ # https://packaging.python.org/en/latest/guides/creating-and-discovering-plugins/
+ def discoverExtensions(self, dirs, work_dir):
+ """
+ Discovers any extensions that are available in any of the given directories, as well as any extensions that are available
+ from PyPi repositories
+
+ :param dirs: the directories to check for any local extensions
+ :param work_dir: the working directory
+ """
+ self.__discover_local_extensions__(dirs, work_dir)
+ self.__discover_extensions_from_pypi__(work_dir)
+
+ def __discover_extensions_from_pypi__(self, work_dir):
+ self.__discover_extensions_from_paths__(None, work_dir, True)
+
+ def __discover_local_extensions__(self, dirs, work_dir):
+ self.__discover_extensions_from_paths__(dirs, work_dir, False)
+
+
+ def __discover_extensions_from_paths__(self, paths, work_dir, require_nifi_prefix):
+ for finder, name, ispkg in pkgutil.iter_modules(paths):
+ if not require_nifi_prefix or name.startswith('nifi_'):
+ module_file = '<Unknown Module File>'
+ try:
+ module = finder.find_module(name)
+ module_file = module.path
+ logger.info('Discovered extension %s' % module_file)
+
+ self.__gather_extension_details__(module_file, work_dir)
+ except Exception:
+ logger.error("Failed to load Python extensions from module file {0}. This module will be ignored.".format(module_file), exc_info=True)
+
+
+
+ def __gather_extension_details__(self, module_file, work_dir, local_dependencies=None):
+ path = Path(module_file)
+ basename = os.path.basename(module_file)
+
+ # If no local_dependencies have been provided, check to see if there are any.
+ # We consider any Python module in the same directory as a local dependency
+ if local_dependencies is None:
+ local_dependencies = []
+
+ if basename == '__init__.py':
+ dir = path.parent
+ for filename in os.listdir(dir):
+ if not filename.endswith('.py'):
+ continue
+ if filename == '__init__.py':
+ continue
+
+ child_module_file = os.path.join(dir, filename)
+ local_dependencies.append(child_module_file)
+
+ # If the module file is an __init__.py file, we check all module files in the same directory.
+ if basename == '__init__.py':
+ dir = path.parent
+ for filename in os.listdir(dir):
+ if not filename.endswith('.py'):
+ continue
+ if filename == '__init__.py':
+ continue
+
+ child_module_file = os.path.join(dir, filename)
+ self.__gather_extension_details__(child_module_file, work_dir, local_dependencies=local_dependencies)
+
+ classes_and_details = self.__get_processor_classes_and_details__(module_file)
+ for classname, details in classes_and_details.items():
+ id = ExtensionId(classname, details.version)
+ logger.info("Found local dependencies {0} for {1}".format(local_dependencies, classname))
+
+ details.local_dependencies = local_dependencies
+
+ # Add class name to processor types only if not there already
+ if id not in self.processor_details:
+ self.processor_details[id] = details
+
+ self.module_files_by_extension_type[id] = module_file
+
+ # If there are any dependencies, use pip to install them
+ extension_working_dir = os.path.join(work_dir, 'extensions', classname, details.version)
+ self.__import_external_dependencies__(classname, details.dependencies, extension_working_dir)
+
+
+ def __get_dependencies_for_extension_type__(self, extension_type, version):
+ id = ExtensionId(extension_type, version)
+ return self.processor_details[id].dependencies
+
+ def __get_processor_classes_and_details__(self, module_file):
+ # Parse the python file
+ with open(module_file) as file:
+ root_node = ast.parse(file.read())
+
+ details_by_class = {}
+
+ # Get top-level class nodes (e.g., MyProcessor)
+ class_nodes = self.__get_class_nodes__(root_node)
+
+ for class_node in class_nodes:
+ logger.debug("Checking if class %s is a processor" % class_node.name)
+ if self.__is_processor_class_node__(class_node):
+ logger.info("Discovered Processor class {0} in module {1}".format(class_node.name, module_file))
+ details = self.__get_processor_details__(class_node, module_file)
+ details_by_class[class_node.name] = details
+
+ return details_by_class
+
+
+ def __is_processor_class_node__(self, class_node):
+ """
+ Checks if the Abstract Syntax Tree (AST) Node represents a Processor class.
+ We are looking for any classes within the given module file that look like:
+
+ class MyProcessor:
+ ...
+ class Java:
+ implements = ['org.apache.nifi.python.processor.FlowFileTransform']
+
+ :param class_node: the abstract syntax tree (AST) node
+ :return: True if the AST Node represents a Python Class that is a Processor, False otherwise
+ """
+
+ # Look for a 'Java' sub-class
+ child_class_nodes = self.__get_class_nodes__(class_node)
+
+ for child_class_node in child_class_nodes:
+ if child_class_node.name == 'Java':
+ # Look for an assignment that assigns values to the `implements` keyword
+ assignment_nodes = self.__get_assignment_nodes__(child_class_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == 'implements'):
+ assigned_values = assignment_node.value.elts
+ for assigned_value in assigned_values:
+ if assigned_value.value in self.processorInterfaces:
+ return True
+ return False
+
+
+ def __get_processor_details__(self, class_node, module_file):
+ # Look for a 'ProcessorDetails' class
+ child_class_nodes = self.__get_class_nodes__(class_node)
+
+ for child_class_node in child_class_nodes:
+ if child_class_node.name == 'ProcessorDetails':
+ logger.debug('Found ProcessorDetails class in %s' % class_node.name)
+ version = self.__get_processor_version__(child_class_node, class_node.name)
+ dependencies = self.__get_processor_dependencies__(child_class_node, class_node.name)
+ description = self.__get_processor_description__(child_class_node, class_node.name)
+ tags = self.__get_processor_tags__(child_class_node, class_node.name)
+
+ return ExtensionDetails(gateway=self.gateway,
+ type=class_node.name,
+ version=version,
+ dependencies=dependencies,
+ source_location=module_file,
+ description=description,
+ tags=tags)
+
+ return ExtensionDetails(gateway=self.gateway,
+ type=class_node.name,
+ version='Unknown',
+ dependencies=[],
+ source_location=module_file)
+
+
+ def __get_processor_version__(self, details_node, class_name):
+ assignment_nodes = self.__get_assignment_nodes__(details_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == 'version'):
+ assigned_values = assignment_node.value.value
+ logger.info("Found version of {0} to be {1}".format(class_name, assigned_values))
+ return assigned_values
+
+ # No dependencies found
+ logger.info("Found no version information for {0}".format(class_name))
+ return 'Unknown'
+
+
+ def __get_processor_dependencies__(self, details_node, class_name):
+ deps = self.__get_assigned_list__(details_node, class_name, 'dependencies')
+ if len(deps) == 0:
+ logger.info("Found no external dependencies that are required for class %s" % class_name)
+ else:
+ logger.info("Found the following external dependencies that are required for class {0}: {1}".format(class_name, deps))
+
+ return deps
+
+
+ def __get_processor_tags__(self, details_node, class_name):
+ return self.__get_assigned_list__(details_node, class_name, 'tags')
+
+
+ def __get_assigned_list__(self, details_node, class_name, element_name):
+ assignment_nodes = self.__get_assignment_nodes__(details_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == element_name):
+ assigned_values = assignment_node.value.elts
+ declared_dependencies = []
+ for assigned_value in assigned_values:
+ declared_dependencies.append(assigned_value.value)
+
+ return declared_dependencies
+
+ # No values found
+ return []
+
+
+ def __get_processor_description__(self, details_node, class_name):
+ assignment_nodes = self.__get_assignment_nodes__(details_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == 'description'):
+ return assignment_node.value.value
+
+ # No description found
+ logger.debug("Found no description for class %s" % class_name)
+ return None
+
+
+
+ def __get_class_nodes__(self, node):
+ class_nodes = [n for n in node.body if isinstance(n, ast.ClassDef)]
+ return class_nodes
+
+
+ def __get_assignment_nodes__(self, node):
+ assignment_nodes = [n for n in node.body if isinstance(n, ast.Assign)]
+ return assignment_nodes
+
+
+ def __import_external_dependencies__(self, class_name, dependencies, target_dir):
+ if not os.path.exists(target_dir):
+ os.makedirs(target_dir)
+
+ completion_marker_file = os.path.join(target_dir, 'dependency-download.complete')
+ if os.path.exists(completion_marker_file):
+ logger.info("All dependencies have already been imported for {0}".format(class_name))
+ return True
+
+ success = False
+ if len(dependencies) > 0:
+ python_cmd = os.getenv("PYTHON_CMD")
+ args = [python_cmd, '-m', 'pip', 'install', '--target', target_dir]
+ for dep in dependencies:
+ args.append(dep)
+
+ logger.info("Importing dependencies {0} for {1} to {2} using command {3}".format(dependencies, class_name, target_dir, args))
+ result = subprocess.run(args)
+
+ if result.returncode == 0:
+ logger.info("Successfully imported requirements for {0} to {1}".format(class_name, target_dir))
+ success = True
+ else:
+ logger.error("Failed to import requirements for {0}: process exited with status code {1}".format(class_name, result))
+ return False
+ else:
+ logger.info("No dependencies to import for {0}".format(class_name))
+ success = True
+
+ if success:
+ # Write a completion Marker File
+ file = open(completion_marker_file, "w")
Review Comment:
I agree this should be changed to follow recommended Python conventions for file handling.
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework/src/main/python/framework/ExtensionManager.py:
##########
@@ -0,0 +1,531 @@
+# 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.
+
+import os
+import importlib
+import sys
+import importlib.util # Note requires Python 3.4+
+import inspect
+import logging
+import subprocess
+import ast
+import pkgutil
+from pathlib import Path
+
+logger = logging.getLogger("org.apache.nifi.py4j.ExtensionManager")
+
+# A simple wrapper class to encompass a processor type and its version
+class ExtensionId:
+ def __init__(self, classname=None, version=None):
+ self.classname = classname
+ self.version = version
+
+ def __hash__(self):
+ return hash((self.classname, self.version))
+
+ def __eq__(self, other):
+ return (self.classname, self.version) == (other.classname, other.version)
+
+
+class ExtensionDetails:
+ class Java:
+ implements = ['org.apache.nifi.python.PythonProcessorDetails']
+
+ def __init__(self, gateway, type, version='Unknown', dependencies=None, source_location=None, package_name=None, description=None, tags=None):
+ self.gateway = gateway
+ if dependencies is None:
+ dependencies = []
+ if tags is None:
+ tags = []
+
+ self.type = type
+ self.version = version
+ self.dependencies = dependencies
+ self.source_location = source_location
+ self.package_name = package_name
+ self.description = description
+ self.tags = tags
+
+ def getProcessorType(self):
+ return self.type
+
+ def getProcessorVersion(self):
+ return self.version
+
+ def getSourceLocation(self):
+ return self.source_location
+
+ def getPyPiPackageName(self):
+ return self.package_name
+
+ def getDependencies(self):
+ list = self.gateway.jvm.java.util.ArrayList()
+ for dep in self.dependencies:
+ list.add(dep)
+
+ return list
+
+ def getCapabilityDescription(self):
+ return self.description
+
+ def getTags(self):
+ list = self.gateway.jvm.java.util.ArrayList()
+ for tag in self.tags:
+ list.add(tag)
+
+ return list
+
+
+
+
+class ExtensionManager:
+ """
+ ExtensionManager is responsible for discovery of extensions types and the lifecycle management of those extension types.
+ Discovery of extension types includes finding what extension types are available
+ (e.g., which Processor types exist on the system), as well as information about those extension types, such as
+ the extension's documentation (tags and capability description).
+
+ Lifecycle management includes determining the third-party dependencies that an extension has and ensuring that those
+ third-party dependencies have been imported.
+ """
+
+ processorInterfaces = ['org.apache.nifi.python.processor.FlowFileTransform', 'org.apache.nifi.python.processor.RecordTransform']
+ processor_details = {}
+ processor_class_by_name = {}
+ module_files_by_extension_type = {}
+ dependency_directories = {}
+
+ def __init__(self, gateway):
+ self.gateway = gateway
+
+
+ def getProcessorTypes(self):
+ """
+ :return: a list of Processor types that have been discovered by the #discoverExtensions method
+ """
+ return self.processor_details.values()
+
+ def getProcessorClass(self, type, version, work_dir):
+ """
+ Returns the Python class that can be used to instantiate a processor of the given type.
+ Additionally, it ensures that the required third-party dependencies are on the system path in order to ensure that
+ the necessary libraries are available to the Processor so that it can be instantiated and used.
+
+ :param type: the type of Processor
+ :param version: the version of the Processor
+ :param work_dir: the working directory for extensions
+ :return: the Python class that can be used to instantiate a Processor of the given type and version
+
+ :raises ValueError: if there is no known Processor with the given type and version
+ """
+ id = ExtensionId(classname=type, version=version)
+ if id in self.processor_class_by_name:
+ return self.processor_class_by_name[id]
+
+ if id not in self.module_files_by_extension_type:
+ raise ValueError('Invalid Processor Type: No module is known to contain Processor of type ' + type + ' version ' + version)
+ module_file = self.module_files_by_extension_type[id]
+
+ if id in self.processor_details:
+ extension_working_dir = os.path.join(work_dir, 'extensions', type, version)
+ sys.path.insert(0, extension_working_dir)
+
+ details = self.processor_details[id]
+ processor_class = self.__load_extension_module__(module_file, details.local_dependencies)
+ self.processor_class_by_name[id] = processor_class
+ return processor_class
+
+
+ def reload_processor(self, processor_type, version, work_dir):
+ """
+ Reloads the class definition for the given processor type. This is used in order to ensure that any changes that have
+ been made to the Processor are reloaded and will take effect.
+
+ :param processor_type: the type of the processor whose class definition should be reloaded
+ :param version: the version of the processor whose class definition should be reloaded
+ :param work_dir: the working directory
+ :return: the new class definition
+ """
+ id = ExtensionId(classname=processor_type, version=version)
+
+ # get the python module file that contains the specified processor
+ module_file = self.get_module_file(processor_type, version)
+
+ # Call load_extension to ensure that we load all necessary dependencies, in case they have changed
+ self.__gather_extension_details__(module_file, work_dir)
+
+ # Reload the processor class itself
+ details = self.processor_details[id]
+ processor_class = self.__load_extension_module__(module_file, details.local_dependencies)
+
+ # Update our cache so that when the processor is created again, the new class will be used
+ self.processor_class_by_name[id] = processor_class
+
+
+ def get_module_file(self, processor_type, version):
+ """
+ Returns the module file that contains the source for the given Processor type and version
+ :param processor_type: the Processor type
+ :param version: the version of the Processor
+ :return: the file that contains the source for the given Processor
+
+ :raises ValueError: if no Processor type is known for the given type and version
+ """
+ id = ExtensionId(processor_type, version)
+ if id not in self.module_files_by_extension_type:
+ raise ValueError('Invalid Processor Type: No module is known to contain Processor of type ' + processor_type + ' version ' + version)
+
+ return self.module_files_by_extension_type[id]
+
+
+ # Discover extensions using the 'prefix' method described in
+ # https://packaging.python.org/en/latest/guides/creating-and-discovering-plugins/
+ def discoverExtensions(self, dirs, work_dir):
+ """
+ Discovers any extensions that are available in any of the given directories, as well as any extensions that are available
+ from PyPi repositories
+
+ :param dirs: the directories to check for any local extensions
+ :param work_dir: the working directory
+ """
+ self.__discover_local_extensions__(dirs, work_dir)
+ self.__discover_extensions_from_pypi__(work_dir)
+
+ def __discover_extensions_from_pypi__(self, work_dir):
+ self.__discover_extensions_from_paths__(None, work_dir, True)
+
+ def __discover_local_extensions__(self, dirs, work_dir):
+ self.__discover_extensions_from_paths__(dirs, work_dir, False)
+
+
+ def __discover_extensions_from_paths__(self, paths, work_dir, require_nifi_prefix):
+ for finder, name, ispkg in pkgutil.iter_modules(paths):
+ if not require_nifi_prefix or name.startswith('nifi_'):
+ module_file = '<Unknown Module File>'
+ try:
+ module = finder.find_module(name)
+ module_file = module.path
+ logger.info('Discovered extension %s' % module_file)
+
+ self.__gather_extension_details__(module_file, work_dir)
+ except Exception:
+ logger.error("Failed to load Python extensions from module file {0}. This module will be ignored.".format(module_file), exc_info=True)
+
+
+
+ def __gather_extension_details__(self, module_file, work_dir, local_dependencies=None):
+ path = Path(module_file)
+ basename = os.path.basename(module_file)
+
+ # If no local_dependencies have been provided, check to see if there are any.
+ # We consider any Python module in the same directory as a local dependency
+ if local_dependencies is None:
+ local_dependencies = []
+
+ if basename == '__init__.py':
+ dir = path.parent
+ for filename in os.listdir(dir):
+ if not filename.endswith('.py'):
+ continue
+ if filename == '__init__.py':
+ continue
+
+ child_module_file = os.path.join(dir, filename)
+ local_dependencies.append(child_module_file)
+
+ # If the module file is an __init__.py file, we check all module files in the same directory.
+ if basename == '__init__.py':
+ dir = path.parent
+ for filename in os.listdir(dir):
+ if not filename.endswith('.py'):
+ continue
+ if filename == '__init__.py':
+ continue
+
+ child_module_file = os.path.join(dir, filename)
+ self.__gather_extension_details__(child_module_file, work_dir, local_dependencies=local_dependencies)
+
+ classes_and_details = self.__get_processor_classes_and_details__(module_file)
+ for classname, details in classes_and_details.items():
+ id = ExtensionId(classname, details.version)
+ logger.info("Found local dependencies {0} for {1}".format(local_dependencies, classname))
+
+ details.local_dependencies = local_dependencies
+
+ # Add class name to processor types only if not there already
+ if id not in self.processor_details:
+ self.processor_details[id] = details
+
+ self.module_files_by_extension_type[id] = module_file
+
+ # If there are any dependencies, use pip to install them
+ extension_working_dir = os.path.join(work_dir, 'extensions', classname, details.version)
+ self.__import_external_dependencies__(classname, details.dependencies, extension_working_dir)
+
+
+ def __get_dependencies_for_extension_type__(self, extension_type, version):
+ id = ExtensionId(extension_type, version)
+ return self.processor_details[id].dependencies
+
+ def __get_processor_classes_and_details__(self, module_file):
+ # Parse the python file
+ with open(module_file) as file:
+ root_node = ast.parse(file.read())
+
+ details_by_class = {}
+
+ # Get top-level class nodes (e.g., MyProcessor)
+ class_nodes = self.__get_class_nodes__(root_node)
+
+ for class_node in class_nodes:
+ logger.debug("Checking if class %s is a processor" % class_node.name)
+ if self.__is_processor_class_node__(class_node):
+ logger.info("Discovered Processor class {0} in module {1}".format(class_node.name, module_file))
+ details = self.__get_processor_details__(class_node, module_file)
+ details_by_class[class_node.name] = details
+
+ return details_by_class
+
+
+ def __is_processor_class_node__(self, class_node):
+ """
+ Checks if the Abstract Syntax Tree (AST) Node represents a Processor class.
+ We are looking for any classes within the given module file that look like:
+
+ class MyProcessor:
+ ...
+ class Java:
+ implements = ['org.apache.nifi.python.processor.FlowFileTransform']
+
+ :param class_node: the abstract syntax tree (AST) node
+ :return: True if the AST Node represents a Python Class that is a Processor, False otherwise
+ """
+
+ # Look for a 'Java' sub-class
+ child_class_nodes = self.__get_class_nodes__(class_node)
+
+ for child_class_node in child_class_nodes:
+ if child_class_node.name == 'Java':
+ # Look for an assignment that assigns values to the `implements` keyword
+ assignment_nodes = self.__get_assignment_nodes__(child_class_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == 'implements'):
+ assigned_values = assignment_node.value.elts
+ for assigned_value in assigned_values:
+ if assigned_value.value in self.processorInterfaces:
+ return True
+ return False
+
+
+ def __get_processor_details__(self, class_node, module_file):
+ # Look for a 'ProcessorDetails' class
+ child_class_nodes = self.__get_class_nodes__(class_node)
+
+ for child_class_node in child_class_nodes:
+ if child_class_node.name == 'ProcessorDetails':
+ logger.debug('Found ProcessorDetails class in %s' % class_node.name)
+ version = self.__get_processor_version__(child_class_node, class_node.name)
+ dependencies = self.__get_processor_dependencies__(child_class_node, class_node.name)
+ description = self.__get_processor_description__(child_class_node, class_node.name)
+ tags = self.__get_processor_tags__(child_class_node, class_node.name)
+
+ return ExtensionDetails(gateway=self.gateway,
+ type=class_node.name,
+ version=version,
+ dependencies=dependencies,
+ source_location=module_file,
+ description=description,
+ tags=tags)
+
+ return ExtensionDetails(gateway=self.gateway,
+ type=class_node.name,
+ version='Unknown',
+ dependencies=[],
+ source_location=module_file)
+
+
+ def __get_processor_version__(self, details_node, class_name):
+ assignment_nodes = self.__get_assignment_nodes__(details_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == 'version'):
+ assigned_values = assignment_node.value.value
+ logger.info("Found version of {0} to be {1}".format(class_name, assigned_values))
+ return assigned_values
+
+ # No dependencies found
+ logger.info("Found no version information for {0}".format(class_name))
+ return 'Unknown'
+
+
+ def __get_processor_dependencies__(self, details_node, class_name):
+ deps = self.__get_assigned_list__(details_node, class_name, 'dependencies')
+ if len(deps) == 0:
+ logger.info("Found no external dependencies that are required for class %s" % class_name)
+ else:
+ logger.info("Found the following external dependencies that are required for class {0}: {1}".format(class_name, deps))
+
+ return deps
+
+
+ def __get_processor_tags__(self, details_node, class_name):
+ return self.__get_assigned_list__(details_node, class_name, 'tags')
+
+
+ def __get_assigned_list__(self, details_node, class_name, element_name):
+ assignment_nodes = self.__get_assignment_nodes__(details_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == element_name):
+ assigned_values = assignment_node.value.elts
+ declared_dependencies = []
+ for assigned_value in assigned_values:
+ declared_dependencies.append(assigned_value.value)
+
+ return declared_dependencies
+
+ # No values found
+ return []
+
+
+ def __get_processor_description__(self, details_node, class_name):
+ assignment_nodes = self.__get_assignment_nodes__(details_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == 'description'):
+ return assignment_node.value.value
+
+ # No description found
+ logger.debug("Found no description for class %s" % class_name)
+ return None
+
+
+
+ def __get_class_nodes__(self, node):
+ class_nodes = [n for n in node.body if isinstance(n, ast.ClassDef)]
+ return class_nodes
+
+
+ def __get_assignment_nodes__(self, node):
+ assignment_nodes = [n for n in node.body if isinstance(n, ast.Assign)]
+ return assignment_nodes
+
+
+ def __import_external_dependencies__(self, class_name, dependencies, target_dir):
+ if not os.path.exists(target_dir):
+ os.makedirs(target_dir)
+
+ completion_marker_file = os.path.join(target_dir, 'dependency-download.complete')
+ if os.path.exists(completion_marker_file):
+ logger.info("All dependencies have already been imported for {0}".format(class_name))
+ return True
+
+ success = False
+ if len(dependencies) > 0:
Review Comment:
These are some good questions, I think we need to take a closer look at using a standard format for declaring additional dependencies as opposed to implementing a new way.
Supporting alternative repositories is also very important for some environments with either private repositories or lack of Internet access.
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java:
##########
@@ -823,6 +850,80 @@ private static FlowFileRepository createFlowFileRepository(final NiFiProperties
}
}
+
+ private PythonBridge createPythonBridge(final NiFiProperties nifiProperties, final ControllerServiceProvider serviceProvider) {
+ final String pythonCommand = nifiProperties.getProperty(NiFiProperties.PYTHON_COMMAND);
+ if (pythonCommand == null) {
+ LOG.warn("Will not enable Python Extensions because the nifi.python.command property has not been configured in nifi.properties");
+ return new DisabledPythonBridge();
+ }
+
+ final String commsTimeout = nifiProperties.getProperty(NiFiProperties.PYTHON_COMMS_TIMEOUT);
+ final File pythonFrameworkSourceDirectory = nifiProperties.getPythonFrameworkSourceDirectory();
+ final List<File> pythonExtensionsDirectories = nifiProperties.getPythonExtensionsDirectories();
+ final File pythonWorkingDirectory = new File(nifiProperties.getProperty(NiFiProperties.PYTHON_WORKING_DIRECTORY));
+ final File pythonLogsDirectory = new File(nifiProperties.getProperty(NiFiProperties.PYTHON_LOGS_DIRECTORY));
+
+ int maxProcesses = nifiProperties.getIntegerProperty(NiFiProperties.PYTHON_MAX_PROCESSES, 20);
+ int maxProcessesPerType = nifiProperties.getIntegerProperty(NiFiProperties.PYTHON_MAX_PROCESSES_PER_TYPE, 2);
+
+ // Validate configuration for max numbers of processes.
+ if (maxProcessesPerType < 1) {
+ LOG.warn("Configured value for {} in nifi.properties is {}, which is invalid. Defaulting to 2.", NiFiProperties.PYTHON_MAX_PROCESSES_PER_TYPE, maxProcessesPerType);
+ maxProcessesPerType = 2;
+ }
+ if (maxProcesses < 0) {
+ LOG.warn("Configured value for {} in nifi.properties is {}, which is invalid. Defaulting to 20.", NiFiProperties.PYTHON_MAX_PROCESSES, maxProcessesPerType);
+ maxProcesses = 20;
+ }
+ if (maxProcesses == 0) {
+ LOG.warn("Will not enable Python Extensions because the {} property in nifi.properties is set to 0.", NiFiProperties.PYTHON_MAX_PROCESSES);
+ return new DisabledPythonBridge();
+ }
+ if (maxProcessesPerType > maxProcesses) {
+ LOG.warn("Configured values for {} and {} in nifi.properties are {} and {} (respectively), which is invalid. " +
+ "Cannot set max process count per extension type greater than the max number of processors. Setting both to {}",
+ NiFiProperties.PYTHON_MAX_PROCESSES_PER_TYPE, NiFiProperties.PYTHON_MAX_PROCESSES, maxProcessesPerType, maxProcesses, maxProcesses);
+
+ maxProcessesPerType = maxProcesses;
+ }
+
+ final PythonProcessConfig pythonProcessConfig = new PythonProcessConfig.Builder()
+ .pythonCommand(pythonCommand)
+ .pythonFrameworkDirectory(pythonFrameworkSourceDirectory)
+ .pythonExtensionsDirectories(pythonExtensionsDirectories)
+ .pythonLogsDirectory(pythonLogsDirectory)
+ .pythonWorkingDirectory(pythonWorkingDirectory)
+ .commsTimeout(commsTimeout == null ? null : Duration.ofMillis(FormatUtils.getTimeDuration(commsTimeout, TimeUnit.MILLISECONDS)))
+ .maxPythonProcesses(maxProcesses)
+ .maxPythonProcessesPerType(maxProcessesPerType)
+ .build();
+
+ final ControllerServiceTypeLookup serviceTypeLookup = serviceProvider::getControllerServiceType;
+
+ try {
+ final PythonBridge bridge = NarThreadContextClassLoader.createInstance(extensionManager, STANDARD_PYTHON_BRIDGE_IMPLEMENTATION_CLASS, PythonBridge.class, null);
+
+ final PythonBridgeInitializationContext initializationContext = new PythonBridgeInitializationContext() {
+ @Override
+ public PythonProcessConfig getPythonProcessConfig() {
+ return pythonProcessConfig;
+ }
+
+ @Override
+ public ControllerServiceTypeLookup getControllerServiceTypeLookup() {
+ return serviceTypeLookup;
+ }
+ };
+
+ bridge.initialize(initializationContext);
+ return bridge;
+ } catch (final Exception e) {
+ throw new RuntimeException(e);
Review Comment:
Recommend adding a message to this exception:
```suggestion
throw new RuntimeException("Python Bridge initialization failed", e);
```
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcess.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.py4j;
+
+import org.apache.nifi.py4j.client.JavaObjectBindings;
+import org.apache.nifi.py4j.client.NiFiPythonGateway;
+import org.apache.nifi.py4j.client.StandardPythonClient;
+import org.apache.nifi.py4j.server.NiFiGatewayServer;
+import org.apache.nifi.python.ControllerServiceTypeLookup;
+import org.apache.nifi.python.PythonController;
+import org.apache.nifi.python.PythonProcessConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import py4j.CallbackClient;
+import py4j.GatewayServer;
+
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+// TODO / Figure Out for MVP:
+// MUST DO:
+// - Documentation
+// - Admin Guide
+// - JavaDocs
+// - Developer Guide
+// - Explain how communication between Java & Python work.
+// - Java is preferred, Python is slower and more expensive b/c of network
+// - Different Extension Points (FlowFileTransform, RecordTransform)
+// - What the API Looks like, Links to JavaDocs for ProcessContext, etc.
+// - Example Code
+// - Exposing properties
+// - Relationships
+// - Controller Services
+// - Need to update docs to show the interfaces that are exposed, explain how to get these...
+// - Design Doc
+// - Setup proper logging on the Python side: https://docs.python.org/2/howto/logging-cookbook.html#using-file-rotation
+// - For FlowFileTransform, allow the result to contain either a byte array or a String. If a String, just convert in the parent class.
+// - Figure out how to deal with Python Packaging
+// - Need to figure out how to deal with additionalDetails.html, docs directory in python project typically?
+// - Understand how to deal with versioning
+// - Look at performance improvements for Py4J - socket comms appear to be INCREDIBLY slow.
+// - Create test that calls Python 1M times. Just returns 'hello'. See how long it takes
+// - Create test that calls Python 1M times. Returns <java object>.toString() and see how long it takes.
+// - Will help to understand if it's the call from Java to Python that's slow, Python to Java, or both.
+// - Performance concern for TransformRecord
+// - Currently, triggering the transform() method is pretty fast. But then the Result object comes back and we have to call into the Python side to call the getters
+// over and over. Need to look into instead serializing the entire response as JSON and sending that back.
+// - Also, since this is heavy JSON processing, might want to consider ORJSON or something like that instead of inbuilt JSON parser/generator
+// - Test pip install nifi-my-proc, does nifi pick it up?
+// - When ran DetectObjectInImage with multiple threads, Python died. Need to figure out why.
+// - If Python Process dies, need to create a new process and need to then create all of the Processors that were in that Process and initialize them.
+// - Milestone 2 or 3, not Milestone 1.
+// - Remove test-pypi usage from ExtensionManager.py
+// - Additional Interfaces beyond just FlowFileTransform
+// - FlowFileSource
+// - Restructure Maven projects
+// - Should this all go under Framework?
+//
+//
+// CONSIDER:
+// - Clustering: Ensure component on all nodes?
+// - Consider "pip freeze" type of thing to ensure that python dependencies are same across nodes when joining cluster.
+// - Update python code to use python_style_method_names instead of javaStyleMethodNames
+// - Also add 'failure' and 'original' relationships to FlowFileTransform
+//
+//
+// Can punt for now:
+// - We have an issue with objects created from Processor calling into Java. Is fine when we provide objects to Python but when
+// it makes a callback, those objects are bound and never unbound!!!
+// *** This appears to be fine as long as on the Python side we set manage_memory=True ***
+// But it does cut perf in half. May be room for improvement somehow? By implementing a 'bulk delete' custom command?
+// - Appears to still be issues with timeout in nifi when set to 10 secs... initially will probably just leave default of 0sec.
Review Comment:
These notes are helpful, but it looks like some of these comments should be captured in Jira or other design documentation, rather than this class.
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework/src/main/python/framework/ExtensionManager.py:
##########
@@ -0,0 +1,531 @@
+# 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.
+
+import os
+import importlib
+import sys
+import importlib.util # Note requires Python 3.4+
+import inspect
+import logging
+import subprocess
+import ast
+import pkgutil
+from pathlib import Path
+
+logger = logging.getLogger("org.apache.nifi.py4j.ExtensionManager")
+
+# A simple wrapper class to encompass a processor type and its version
+class ExtensionId:
+ def __init__(self, classname=None, version=None):
+ self.classname = classname
+ self.version = version
+
+ def __hash__(self):
+ return hash((self.classname, self.version))
+
+ def __eq__(self, other):
+ return (self.classname, self.version) == (other.classname, other.version)
+
+
+class ExtensionDetails:
+ class Java:
+ implements = ['org.apache.nifi.python.PythonProcessorDetails']
+
+ def __init__(self, gateway, type, version='Unknown', dependencies=None, source_location=None, package_name=None, description=None, tags=None):
+ self.gateway = gateway
+ if dependencies is None:
+ dependencies = []
+ if tags is None:
+ tags = []
+
+ self.type = type
+ self.version = version
+ self.dependencies = dependencies
+ self.source_location = source_location
+ self.package_name = package_name
+ self.description = description
+ self.tags = tags
+
+ def getProcessorType(self):
+ return self.type
+
+ def getProcessorVersion(self):
+ return self.version
+
+ def getSourceLocation(self):
+ return self.source_location
+
+ def getPyPiPackageName(self):
+ return self.package_name
+
+ def getDependencies(self):
+ list = self.gateway.jvm.java.util.ArrayList()
+ for dep in self.dependencies:
+ list.add(dep)
+
+ return list
+
+ def getCapabilityDescription(self):
+ return self.description
+
+ def getTags(self):
+ list = self.gateway.jvm.java.util.ArrayList()
+ for tag in self.tags:
+ list.add(tag)
+
+ return list
+
+
+
+
+class ExtensionManager:
+ """
+ ExtensionManager is responsible for discovery of extensions types and the lifecycle management of those extension types.
+ Discovery of extension types includes finding what extension types are available
+ (e.g., which Processor types exist on the system), as well as information about those extension types, such as
+ the extension's documentation (tags and capability description).
+
+ Lifecycle management includes determining the third-party dependencies that an extension has and ensuring that those
+ third-party dependencies have been imported.
+ """
+
+ processorInterfaces = ['org.apache.nifi.python.processor.FlowFileTransform', 'org.apache.nifi.python.processor.RecordTransform']
+ processor_details = {}
+ processor_class_by_name = {}
+ module_files_by_extension_type = {}
+ dependency_directories = {}
+
+ def __init__(self, gateway):
+ self.gateway = gateway
+
+
+ def getProcessorTypes(self):
+ """
+ :return: a list of Processor types that have been discovered by the #discoverExtensions method
+ """
+ return self.processor_details.values()
+
+ def getProcessorClass(self, type, version, work_dir):
+ """
+ Returns the Python class that can be used to instantiate a processor of the given type.
+ Additionally, it ensures that the required third-party dependencies are on the system path in order to ensure that
+ the necessary libraries are available to the Processor so that it can be instantiated and used.
+
+ :param type: the type of Processor
+ :param version: the version of the Processor
+ :param work_dir: the working directory for extensions
+ :return: the Python class that can be used to instantiate a Processor of the given type and version
+
+ :raises ValueError: if there is no known Processor with the given type and version
+ """
+ id = ExtensionId(classname=type, version=version)
+ if id in self.processor_class_by_name:
+ return self.processor_class_by_name[id]
+
+ if id not in self.module_files_by_extension_type:
+ raise ValueError('Invalid Processor Type: No module is known to contain Processor of type ' + type + ' version ' + version)
+ module_file = self.module_files_by_extension_type[id]
+
+ if id in self.processor_details:
+ extension_working_dir = os.path.join(work_dir, 'extensions', type, version)
+ sys.path.insert(0, extension_working_dir)
+
+ details = self.processor_details[id]
+ processor_class = self.__load_extension_module__(module_file, details.local_dependencies)
+ self.processor_class_by_name[id] = processor_class
+ return processor_class
+
+
+ def reload_processor(self, processor_type, version, work_dir):
+ """
+ Reloads the class definition for the given processor type. This is used in order to ensure that any changes that have
+ been made to the Processor are reloaded and will take effect.
+
+ :param processor_type: the type of the processor whose class definition should be reloaded
+ :param version: the version of the processor whose class definition should be reloaded
+ :param work_dir: the working directory
+ :return: the new class definition
+ """
+ id = ExtensionId(classname=processor_type, version=version)
+
+ # get the python module file that contains the specified processor
+ module_file = self.get_module_file(processor_type, version)
+
+ # Call load_extension to ensure that we load all necessary dependencies, in case they have changed
+ self.__gather_extension_details__(module_file, work_dir)
+
+ # Reload the processor class itself
+ details = self.processor_details[id]
+ processor_class = self.__load_extension_module__(module_file, details.local_dependencies)
+
+ # Update our cache so that when the processor is created again, the new class will be used
+ self.processor_class_by_name[id] = processor_class
+
+
+ def get_module_file(self, processor_type, version):
+ """
+ Returns the module file that contains the source for the given Processor type and version
+ :param processor_type: the Processor type
+ :param version: the version of the Processor
+ :return: the file that contains the source for the given Processor
+
+ :raises ValueError: if no Processor type is known for the given type and version
+ """
+ id = ExtensionId(processor_type, version)
+ if id not in self.module_files_by_extension_type:
+ raise ValueError('Invalid Processor Type: No module is known to contain Processor of type ' + processor_type + ' version ' + version)
+
+ return self.module_files_by_extension_type[id]
+
+
+ # Discover extensions using the 'prefix' method described in
+ # https://packaging.python.org/en/latest/guides/creating-and-discovering-plugins/
+ def discoverExtensions(self, dirs, work_dir):
+ """
+ Discovers any extensions that are available in any of the given directories, as well as any extensions that are available
+ from PyPi repositories
+
+ :param dirs: the directories to check for any local extensions
+ :param work_dir: the working directory
+ """
+ self.__discover_local_extensions__(dirs, work_dir)
+ self.__discover_extensions_from_pypi__(work_dir)
+
+ def __discover_extensions_from_pypi__(self, work_dir):
+ self.__discover_extensions_from_paths__(None, work_dir, True)
+
+ def __discover_local_extensions__(self, dirs, work_dir):
+ self.__discover_extensions_from_paths__(dirs, work_dir, False)
+
+
+ def __discover_extensions_from_paths__(self, paths, work_dir, require_nifi_prefix):
+ for finder, name, ispkg in pkgutil.iter_modules(paths):
+ if not require_nifi_prefix or name.startswith('nifi_'):
+ module_file = '<Unknown Module File>'
+ try:
+ module = finder.find_module(name)
+ module_file = module.path
+ logger.info('Discovered extension %s' % module_file)
+
+ self.__gather_extension_details__(module_file, work_dir)
+ except Exception:
+ logger.error("Failed to load Python extensions from module file {0}. This module will be ignored.".format(module_file), exc_info=True)
+
+
+
+ def __gather_extension_details__(self, module_file, work_dir, local_dependencies=None):
+ path = Path(module_file)
+ basename = os.path.basename(module_file)
+
+ # If no local_dependencies have been provided, check to see if there are any.
+ # We consider any Python module in the same directory as a local dependency
+ if local_dependencies is None:
+ local_dependencies = []
+
+ if basename == '__init__.py':
+ dir = path.parent
+ for filename in os.listdir(dir):
+ if not filename.endswith('.py'):
+ continue
+ if filename == '__init__.py':
+ continue
+
+ child_module_file = os.path.join(dir, filename)
+ local_dependencies.append(child_module_file)
+
+ # If the module file is an __init__.py file, we check all module files in the same directory.
+ if basename == '__init__.py':
+ dir = path.parent
+ for filename in os.listdir(dir):
+ if not filename.endswith('.py'):
+ continue
+ if filename == '__init__.py':
+ continue
+
+ child_module_file = os.path.join(dir, filename)
+ self.__gather_extension_details__(child_module_file, work_dir, local_dependencies=local_dependencies)
+
+ classes_and_details = self.__get_processor_classes_and_details__(module_file)
+ for classname, details in classes_and_details.items():
+ id = ExtensionId(classname, details.version)
+ logger.info("Found local dependencies {0} for {1}".format(local_dependencies, classname))
+
+ details.local_dependencies = local_dependencies
+
+ # Add class name to processor types only if not there already
+ if id not in self.processor_details:
+ self.processor_details[id] = details
+
+ self.module_files_by_extension_type[id] = module_file
+
+ # If there are any dependencies, use pip to install them
+ extension_working_dir = os.path.join(work_dir, 'extensions', classname, details.version)
+ self.__import_external_dependencies__(classname, details.dependencies, extension_working_dir)
+
+
+ def __get_dependencies_for_extension_type__(self, extension_type, version):
+ id = ExtensionId(extension_type, version)
+ return self.processor_details[id].dependencies
+
+ def __get_processor_classes_and_details__(self, module_file):
+ # Parse the python file
+ with open(module_file) as file:
+ root_node = ast.parse(file.read())
+
+ details_by_class = {}
+
+ # Get top-level class nodes (e.g., MyProcessor)
+ class_nodes = self.__get_class_nodes__(root_node)
+
+ for class_node in class_nodes:
+ logger.debug("Checking if class %s is a processor" % class_node.name)
+ if self.__is_processor_class_node__(class_node):
+ logger.info("Discovered Processor class {0} in module {1}".format(class_node.name, module_file))
+ details = self.__get_processor_details__(class_node, module_file)
+ details_by_class[class_node.name] = details
+
+ return details_by_class
+
+
+ def __is_processor_class_node__(self, class_node):
+ """
+ Checks if the Abstract Syntax Tree (AST) Node represents a Processor class.
+ We are looking for any classes within the given module file that look like:
+
+ class MyProcessor:
+ ...
+ class Java:
+ implements = ['org.apache.nifi.python.processor.FlowFileTransform']
+
+ :param class_node: the abstract syntax tree (AST) node
+ :return: True if the AST Node represents a Python Class that is a Processor, False otherwise
+ """
+
+ # Look for a 'Java' sub-class
+ child_class_nodes = self.__get_class_nodes__(class_node)
+
+ for child_class_node in child_class_nodes:
+ if child_class_node.name == 'Java':
+ # Look for an assignment that assigns values to the `implements` keyword
+ assignment_nodes = self.__get_assignment_nodes__(child_class_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == 'implements'):
+ assigned_values = assignment_node.value.elts
+ for assigned_value in assigned_values:
+ if assigned_value.value in self.processorInterfaces:
+ return True
+ return False
+
+
+ def __get_processor_details__(self, class_node, module_file):
+ # Look for a 'ProcessorDetails' class
+ child_class_nodes = self.__get_class_nodes__(class_node)
+
+ for child_class_node in child_class_nodes:
+ if child_class_node.name == 'ProcessorDetails':
+ logger.debug('Found ProcessorDetails class in %s' % class_node.name)
+ version = self.__get_processor_version__(child_class_node, class_node.name)
+ dependencies = self.__get_processor_dependencies__(child_class_node, class_node.name)
+ description = self.__get_processor_description__(child_class_node, class_node.name)
+ tags = self.__get_processor_tags__(child_class_node, class_node.name)
+
+ return ExtensionDetails(gateway=self.gateway,
+ type=class_node.name,
+ version=version,
+ dependencies=dependencies,
+ source_location=module_file,
+ description=description,
+ tags=tags)
+
+ return ExtensionDetails(gateway=self.gateway,
+ type=class_node.name,
+ version='Unknown',
+ dependencies=[],
+ source_location=module_file)
+
+
+ def __get_processor_version__(self, details_node, class_name):
+ assignment_nodes = self.__get_assignment_nodes__(details_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == 'version'):
+ assigned_values = assignment_node.value.value
+ logger.info("Found version of {0} to be {1}".format(class_name, assigned_values))
+ return assigned_values
+
+ # No dependencies found
+ logger.info("Found no version information for {0}".format(class_name))
+ return 'Unknown'
+
+
+ def __get_processor_dependencies__(self, details_node, class_name):
+ deps = self.__get_assigned_list__(details_node, class_name, 'dependencies')
+ if len(deps) == 0:
+ logger.info("Found no external dependencies that are required for class %s" % class_name)
+ else:
+ logger.info("Found the following external dependencies that are required for class {0}: {1}".format(class_name, deps))
+
+ return deps
+
+
+ def __get_processor_tags__(self, details_node, class_name):
+ return self.__get_assigned_list__(details_node, class_name, 'tags')
+
+
+ def __get_assigned_list__(self, details_node, class_name, element_name):
+ assignment_nodes = self.__get_assignment_nodes__(details_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == element_name):
+ assigned_values = assignment_node.value.elts
+ declared_dependencies = []
+ for assigned_value in assigned_values:
+ declared_dependencies.append(assigned_value.value)
+
+ return declared_dependencies
+
+ # No values found
+ return []
+
+
+ def __get_processor_description__(self, details_node, class_name):
+ assignment_nodes = self.__get_assignment_nodes__(details_node)
+ for assignment_node in assignment_nodes:
+ if (len(assignment_node.targets) == 1 and assignment_node.targets[0].id == 'description'):
+ return assignment_node.value.value
+
+ # No description found
+ logger.debug("Found no description for class %s" % class_name)
+ return None
+
+
+
+ def __get_class_nodes__(self, node):
+ class_nodes = [n for n in node.body if isinstance(n, ast.ClassDef)]
+ return class_nodes
+
+
+ def __get_assignment_nodes__(self, node):
+ assignment_nodes = [n for n in node.body if isinstance(n, ast.Assign)]
+ return assignment_nodes
+
+
+ def __import_external_dependencies__(self, class_name, dependencies, target_dir):
+ if not os.path.exists(target_dir):
+ os.makedirs(target_dir)
+
+ completion_marker_file = os.path.join(target_dir, 'dependency-download.complete')
+ if os.path.exists(completion_marker_file):
+ logger.info("All dependencies have already been imported for {0}".format(class_name))
+ return True
+
+ success = False
+ if len(dependencies) > 0:
Review Comment:
Although having inline dependency declaration is convenient, following a standard convention seems better than introducing a new approach. With support for directory-based modules, what about leveraging the [requirements.txt](https://pip.pypa.io/en/stable/reference/requirements-file-format/) format in that directory?
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcess.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.py4j;
+
+import org.apache.nifi.py4j.client.JavaObjectBindings;
+import org.apache.nifi.py4j.client.NiFiPythonGateway;
+import org.apache.nifi.py4j.client.StandardPythonClient;
+import org.apache.nifi.py4j.server.NiFiGatewayServer;
+import org.apache.nifi.python.ControllerServiceTypeLookup;
+import org.apache.nifi.python.PythonController;
+import org.apache.nifi.python.PythonProcessConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import py4j.CallbackClient;
+import py4j.GatewayServer;
+
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+// TODO / Figure Out for MVP:
+// MUST DO:
+// - Documentation
+// - Admin Guide
+// - JavaDocs
+// - Developer Guide
+// - Explain how communication between Java & Python work.
+// - Java is preferred, Python is slower and more expensive b/c of network
+// - Different Extension Points (FlowFileTransform, RecordTransform)
+// - What the API Looks like, Links to JavaDocs for ProcessContext, etc.
+// - Example Code
+// - Exposing properties
+// - Relationships
+// - Controller Services
+// - Need to update docs to show the interfaces that are exposed, explain how to get these...
+// - Design Doc
+// - Setup proper logging on the Python side: https://docs.python.org/2/howto/logging-cookbook.html#using-file-rotation
+// - For FlowFileTransform, allow the result to contain either a byte array or a String. If a String, just convert in the parent class.
+// - Figure out how to deal with Python Packaging
+// - Need to figure out how to deal with additionalDetails.html, docs directory in python project typically?
+// - Understand how to deal with versioning
+// - Look at performance improvements for Py4J - socket comms appear to be INCREDIBLY slow.
+// - Create test that calls Python 1M times. Just returns 'hello'. See how long it takes
+// - Create test that calls Python 1M times. Returns <java object>.toString() and see how long it takes.
+// - Will help to understand if it's the call from Java to Python that's slow, Python to Java, or both.
+// - Performance concern for TransformRecord
+// - Currently, triggering the transform() method is pretty fast. But then the Result object comes back and we have to call into the Python side to call the getters
+// over and over. Need to look into instead serializing the entire response as JSON and sending that back.
+// - Also, since this is heavy JSON processing, might want to consider ORJSON or something like that instead of inbuilt JSON parser/generator
+// - Test pip install nifi-my-proc, does nifi pick it up?
+// - When ran DetectObjectInImage with multiple threads, Python died. Need to figure out why.
+// - If Python Process dies, need to create a new process and need to then create all of the Processors that were in that Process and initialize them.
+// - Milestone 2 or 3, not Milestone 1.
+// - Remove test-pypi usage from ExtensionManager.py
+// - Additional Interfaces beyond just FlowFileTransform
+// - FlowFileSource
+// - Restructure Maven projects
+// - Should this all go under Framework?
+//
+//
+// CONSIDER:
+// - Clustering: Ensure component on all nodes?
+// - Consider "pip freeze" type of thing to ensure that python dependencies are same across nodes when joining cluster.
+// - Update python code to use python_style_method_names instead of javaStyleMethodNames
+// - Also add 'failure' and 'original' relationships to FlowFileTransform
+//
+//
+// Can punt for now:
+// - We have an issue with objects created from Processor calling into Java. Is fine when we provide objects to Python but when
+// it makes a callback, those objects are bound and never unbound!!!
+// *** This appears to be fine as long as on the Python side we set manage_memory=True ***
+// But it does cut perf in half. May be room for improvement somehow? By implementing a 'bulk delete' custom command?
+// - Appears to still be issues with timeout in nifi when set to 10 secs... initially will probably just leave default of 0sec.
+
+public class PythonProcess {
+ private static final Logger logger = LoggerFactory.getLogger(PythonProcess.class);
+ private static final String PYTHON_CONTROLLER_FILENAME = "Controller.py";
+
+ private final PythonProcessConfig processConfig;
+ private final ControllerServiceTypeLookup controllerServiceTypeLookup;
+ private final File virtualEnvHome;
+ private GatewayServer server;
+ private PythonController controller;
+ private Process process;
+ private NiFiPythonGateway gateway;
+ private final Map<String, Boolean> processorPrefersIsolation = new ConcurrentHashMap<>();
+
+
+ public PythonProcess(final PythonProcessConfig processConfig, final ControllerServiceTypeLookup controllerServiceTypeLookup, final File virtualEnvHome) {
+ this.processConfig = processConfig;
+ this.controllerServiceTypeLookup = controllerServiceTypeLookup;
+ this.virtualEnvHome = virtualEnvHome;
+ }
+
+ public PythonController getController() {
+ return controller;
+ }
+
+ public void start() throws IOException {
+ // TODO: Look into using configured TLS Certs to make this secure by default.
+ final ServerSocketFactory serverSocketFactory = ServerSocketFactory.getDefault();
+ final SocketFactory socketFactory = SocketFactory.getDefault();
+
+ final int timeoutMillis = (int) processConfig.getCommsTimeout().toMillis();
+ final String authToken = null;
+ final CallbackClient callbackClient = new CallbackClient(GatewayServer.DEFAULT_PYTHON_PORT, GatewayServer.defaultAddress(), authToken,
+ 50000L, TimeUnit.MILLISECONDS, socketFactory, false, timeoutMillis);
+
+ final JavaObjectBindings bindings = new JavaObjectBindings();
+ gateway = new NiFiPythonGateway(bindings, null, callbackClient);
+ gateway.startup();
+
+ server = new NiFiGatewayServer(gateway,
+ 0,
+ GatewayServer.defaultAddress(),
+ timeoutMillis,
+ timeoutMillis,
+ Collections.emptyList(),
+ serverSocketFactory);
+ server.start();
Review Comment:
As commented on the class itself, the `start()` method spawns a new Thread with a generic name. It would be very helpful to adjust that behavior to name the thread for tracking.
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework/src/main/python/framework/Controller.py:
##########
@@ -0,0 +1,102 @@
+# 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.
+
+import os
+import logging
+import ExtensionManager
+from py4j.java_gateway import JavaGateway, CallbackServerParameters, GatewayParameters
+
+import PythonProcessorAdapter
+
+# Initialize logging
+logger = logging.getLogger("org.apache.nifi.py4j.Controller")
+logger.setLevel(logging.INFO)
+
+logging.getLogger("py4j").setLevel(logging.WARN)
+
+logsDir = os.getenv('LOGS_DIR')
+logging.basicConfig(filename=logsDir + '/python.log',
Review Comment:
Should this be named `nifi-python.log` to follow the convention of other logs?
```suggestion
logging.basicConfig(filename=logsDir + '/nifi-python.log',
```
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/StandardExtensionDiscoveringManager.java:
##########
@@ -616,8 +703,34 @@ public synchronized ConfigurableComponent getTempComponent(final String classTyp
final ClassLoader bundleClassLoader = bundle.getClassLoader();
try (final NarCloseable narCloseable = NarCloseable.withComponentNarLoader(bundleClassLoader)) {
- final Class<?> componentClass = Class.forName(classType, true, bundleClassLoader);
- final ConfigurableComponent tempComponent = (ConfigurableComponent) componentClass.newInstance();
+ final ConfigurableComponent tempComponent;
+ if (PythonBundle.isPythonCoordinate(bundle.getBundleDetails().getCoordinate())) {
+ // TODO: This is a workaround due to bug in UI. Fix bug in UI.
+ final String type = classType.startsWith("python.") ? classType.substring("python.".length()) : classType;
Review Comment:
Recommend defining a static `PYTHON_TYPE_PREFIX` variable and reusing for `python.` references.
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-nar-utils/src/main/java/org/apache/nifi/nar/StandardExtensionDiscoveringManager.java:
##########
@@ -169,6 +181,75 @@ public void discoverExtensions(final Set<Bundle> narBundles, final boolean logDe
}
}
+ public void setPythonBridge(final PythonBridge pythonBridge) {
+ this.pythonBridge = pythonBridge;
+ }
+
+ @Override
+ public void discoverPythonExtensions(final Bundle pythonBundle) {
+ logger.info("Scanning to discover which Python extensions are available and importing any necessary dependencies. This may take a few minutes. See python logs for more details.");
+ final long start = System.currentTimeMillis();
+ pythonBridge.discoverExtensions();
+
+ bundleCoordinateBundleLookup.putIfAbsent(pythonBundle.getBundleDetails().getCoordinate(), pythonBundle);
+
+ final Set<ExtensionDefinition> processorDefinitions = definitionMap.get(Processor.class);
+ final List<PythonProcessorDetails> pythonProcessorDetails = pythonBridge.getProcessorTypes();
+
+ int processorsFound = 0;
+ for (final PythonProcessorDetails details : pythonProcessorDetails) {
+ final BundleDetails bundleDetails = createBundleDetailsWithOverriddenVersion(pythonBundle.getBundleDetails(), details.getProcessorVersion());
+ final Bundle bundle = new Bundle(bundleDetails, pythonBundle.getClassLoader());
+
+ // TODO: This is a workaround because the UI has a bug that causes it not to work properly if the type doesn't have a '.' in it
+ final String className = "python." + details.getProcessorType();
+ final ExtensionDefinition extensionDefinition = new ExtensionDefinition.Builder()
+ .implementationClassName(className)
+ .runtime(ExtensionRuntime.PYTHON)
+ .bundle(bundle)
+ .extensionType(Processor.class)
+ .description(details.getCapabilityDescription())
+ .tags(details.getTags())
+ .version(details.getProcessorVersion())
+ .build();
+
+ final boolean added = processorDefinitions.add(extensionDefinition);
+ if (added) {
+ processorsFound++;
+ final List<Bundle> bundlesForClass = classNameBundleLookup.computeIfAbsent(className, key -> new ArrayList<>());
+ bundlesForClass.add(bundle);
+ bundleCoordinateBundleLookup.putIfAbsent(bundleDetails.getCoordinate(), bundle);
+ logger.info("Discovered Python Processor {}", details.getProcessorType());
+ } else {
+ logger.debug("Python Processor {} is already known", details.getProcessorType());
+ }
+ }
+
+ logger.info("Discovered {} new/updated Python Processors in {} millis", processorsFound, System.currentTimeMillis() - start);
Review Comment:
Similar to the start message, this message fills the logs. The info log for discovering a new Processor is helpful, but this log seems better at the debug level.
```suggestion
logger.debug("Discovered or updated [{}] Python Processors in {} ms", processorsFound, System.currentTimeMillis() - start);
```
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/server/NiFiGatewayConnection.java:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.py4j.server;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import py4j.Gateway;
+import py4j.GatewayConnection;
+import py4j.GatewayServerListener;
+import py4j.commands.Command;
+
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.util.List;
+
+/**
+ * A customized version of the GatewayConnection that is more suitable for NiFi's needs.
+ *
+ * With the Py4J GatewayConnection, If there's a SocketTimeoutException, py4j will kill the Connection.
+ * This results in the Python side not being able to send any information back to the Java side.
+ * Perhaps this makes sense for some use cases, but it does not make sense for our use case.
+ *
+ * With the default behavior, if we go N seconds (where N is the timeout) without communicating with the Python
+ * side, the Python Process essentially becomes defunct because it can no longer communicate with the NiFi side.
+ *
+ * This means that if the timeout is configured for, say 30 seconds, and we go 30 seconds on a NiFi instance with no
+ * Python processor, we cannot create a Python Processor. Or, if we go 30 seconds without sending a FlowFile to a
+ * Processor, then we can no longer make use of the Processor.
+ *
+ * Because this doesn't make sense for our use case, we extend the existing GatewayConnection, and modify it in a couple
+ * of ways:
+ * <ul>
+ * <li>
+ * When <code>run()</code> is called, we execute super.run() continually until the connection gets poisoned
+ * (due to some Exception other than SocketTimeoutException). Or until the GatewayServer has been shutdown.
+ * This way, even when a SocketTimeoutException occurs, we continue trying to read from the socket and leave
+ * the connection open.
+ * </li>
+ * <li>
+ * Because the super class calls <code>shutdown(boolean)</code> from within <code>run()</code> for any Exception
+ * that occurs, even SocketTimeoutException, we have to engineer a way around this. Before calling <code>shutdown</code>
+ * it calls <code>quietSendFatalError(BufferedWriter, Exception)</code> with the Exception that caused it to shutdown.
+ * We use this to mark the connection as poisoned if and only if the Exception is not a SocketTimeoutException. If a
+ * SocketTimeoutException occurs, we simply ignore it. Then, in the <code>shutdown(boolean)</code> method, we do not
+ * perform the shutdown unless either the connection is poisoned or the GatewayServer has been shutdown (which indicates
+ * that communications with the Process are to be shutdown).
+ * </li>
+ * </ul>
+ * -
+ */
+public class NiFiGatewayConnection extends GatewayConnection {
Review Comment:
Similar to `GatewayServer`, the `GatewayConnection` class creates a new instance of `Thread` in the `startConnection` method. It would be very helpful to override this method and assign a thread name using a ThreadFactory.
##########
nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcess.java:
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.py4j;
+
+import org.apache.nifi.py4j.client.JavaObjectBindings;
+import org.apache.nifi.py4j.client.NiFiPythonGateway;
+import org.apache.nifi.py4j.client.StandardPythonClient;
+import org.apache.nifi.py4j.server.NiFiGatewayServer;
+import org.apache.nifi.python.ControllerServiceTypeLookup;
+import org.apache.nifi.python.PythonController;
+import org.apache.nifi.python.PythonProcessConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import py4j.CallbackClient;
+import py4j.GatewayServer;
+
+import javax.net.ServerSocketFactory;
+import javax.net.SocketFactory;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+
+// TODO / Figure Out for MVP:
+// MUST DO:
+// - Documentation
+// - Admin Guide
+// - JavaDocs
+// - Developer Guide
+// - Explain how communication between Java & Python work.
+// - Java is preferred, Python is slower and more expensive b/c of network
+// - Different Extension Points (FlowFileTransform, RecordTransform)
+// - What the API Looks like, Links to JavaDocs for ProcessContext, etc.
+// - Example Code
+// - Exposing properties
+// - Relationships
+// - Controller Services
+// - Need to update docs to show the interfaces that are exposed, explain how to get these...
+// - Design Doc
+// - Setup proper logging on the Python side: https://docs.python.org/2/howto/logging-cookbook.html#using-file-rotation
+// - For FlowFileTransform, allow the result to contain either a byte array or a String. If a String, just convert in the parent class.
+// - Figure out how to deal with Python Packaging
+// - Need to figure out how to deal with additionalDetails.html, docs directory in python project typically?
+// - Understand how to deal with versioning
+// - Look at performance improvements for Py4J - socket comms appear to be INCREDIBLY slow.
+// - Create test that calls Python 1M times. Just returns 'hello'. See how long it takes
+// - Create test that calls Python 1M times. Returns <java object>.toString() and see how long it takes.
+// - Will help to understand if it's the call from Java to Python that's slow, Python to Java, or both.
+// - Performance concern for TransformRecord
+// - Currently, triggering the transform() method is pretty fast. But then the Result object comes back and we have to call into the Python side to call the getters
+// over and over. Need to look into instead serializing the entire response as JSON and sending that back.
+// - Also, since this is heavy JSON processing, might want to consider ORJSON or something like that instead of inbuilt JSON parser/generator
+// - Test pip install nifi-my-proc, does nifi pick it up?
+// - When ran DetectObjectInImage with multiple threads, Python died. Need to figure out why.
+// - If Python Process dies, need to create a new process and need to then create all of the Processors that were in that Process and initialize them.
+// - Milestone 2 or 3, not Milestone 1.
+// - Remove test-pypi usage from ExtensionManager.py
+// - Additional Interfaces beyond just FlowFileTransform
+// - FlowFileSource
+// - Restructure Maven projects
+// - Should this all go under Framework?
+//
+//
+// CONSIDER:
+// - Clustering: Ensure component on all nodes?
+// - Consider "pip freeze" type of thing to ensure that python dependencies are same across nodes when joining cluster.
+// - Update python code to use python_style_method_names instead of javaStyleMethodNames
+// - Also add 'failure' and 'original' relationships to FlowFileTransform
+//
+//
+// Can punt for now:
+// - We have an issue with objects created from Processor calling into Java. Is fine when we provide objects to Python but when
+// it makes a callback, those objects are bound and never unbound!!!
+// *** This appears to be fine as long as on the Python side we set manage_memory=True ***
+// But it does cut perf in half. May be room for improvement somehow? By implementing a 'bulk delete' custom command?
+// - Appears to still be issues with timeout in nifi when set to 10 secs... initially will probably just leave default of 0sec.
+
+public class PythonProcess {
+ private static final Logger logger = LoggerFactory.getLogger(PythonProcess.class);
+ private static final String PYTHON_CONTROLLER_FILENAME = "Controller.py";
+
+ private final PythonProcessConfig processConfig;
+ private final ControllerServiceTypeLookup controllerServiceTypeLookup;
+ private final File virtualEnvHome;
+ private GatewayServer server;
+ private PythonController controller;
+ private Process process;
+ private NiFiPythonGateway gateway;
+ private final Map<String, Boolean> processorPrefersIsolation = new ConcurrentHashMap<>();
+
+
+ public PythonProcess(final PythonProcessConfig processConfig, final ControllerServiceTypeLookup controllerServiceTypeLookup, final File virtualEnvHome) {
+ this.processConfig = processConfig;
+ this.controllerServiceTypeLookup = controllerServiceTypeLookup;
+ this.virtualEnvHome = virtualEnvHome;
+ }
+
+ public PythonController getController() {
+ return controller;
+ }
+
+ public void start() throws IOException {
+ // TODO: Look into using configured TLS Certs to make this secure by default.
+ final ServerSocketFactory serverSocketFactory = ServerSocketFactory.getDefault();
+ final SocketFactory socketFactory = SocketFactory.getDefault();
+
+ final int timeoutMillis = (int) processConfig.getCommsTimeout().toMillis();
+ final String authToken = null;
+ final CallbackClient callbackClient = new CallbackClient(GatewayServer.DEFAULT_PYTHON_PORT, GatewayServer.defaultAddress(), authToken,
+ 50000L, TimeUnit.MILLISECONDS, socketFactory, false, timeoutMillis);
+
+ final JavaObjectBindings bindings = new JavaObjectBindings();
+ gateway = new NiFiPythonGateway(bindings, null, callbackClient);
+ gateway.startup();
+
+ server = new NiFiGatewayServer(gateway,
+ 0,
+ GatewayServer.defaultAddress(),
+ timeoutMillis,
+ timeoutMillis,
+ Collections.emptyList(),
+ serverSocketFactory);
+ server.start();
+
+ final int listeningPort = server.getListeningPort();
+
+ setupEnvironment();
+ this.process = launchPythonProcess(listeningPort);
+
+ final StandardPythonClient pythonClient = new StandardPythonClient(gateway);
+ controller = pythonClient.getController();
+
+ final long timeout = System.currentTimeMillis() + TimeUnit.SECONDS.toMillis(60L);
+ Exception lastException = null;
+ boolean pingSuccessful = false;
+ while (System.currentTimeMillis() < timeout) {
+ try {
+ final String pingResponse = controller.ping();
+ pingSuccessful = "pong".equals(pingResponse);
+ if (pingSuccessful) {
+ break;
+ } else {
+ logger.debug("Got unexpected response from Py4J Server during ping: {}", pingResponse);
+ }
+ } catch (final Exception e) {
+ lastException = e;
+ logger.debug("Failed to start Py4J Server", e);
+ }
+
+ try {
+ Thread.sleep(50L);
+ } catch (final InterruptedException ie) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ }
+
+ if (!pingSuccessful && lastException != null) {
+ throw new RuntimeException("Failed to start Python Bridge", lastException);
+ }
+
+ controller.setControllerServiceTypeLookup(controllerServiceTypeLookup);
+ logger.info("Successfully started and pinged Python Server. Python Process = {}", process);
+ }
+
+ private Process launchPythonProcess(final int listeningPort) throws IOException {
+ final File pythonFrameworkDirectory = processConfig.getPythonFrameworkDirectory();
+ final File pythonApiDirectory = new File(pythonFrameworkDirectory.getParentFile(), "api");
+ final File pythonLogsDirectory = processConfig.getPythonLogsDirectory();
+ final File pythonCmdFile = new File(processConfig.getPythonCommand());
+ final String pythonCmd = pythonCmdFile.getName();
+ final File pythonCommandFile = new File(virtualEnvHome, "bin/" + pythonCmd);
+ final String pythonCommand = pythonCommandFile.getAbsolutePath();
+
+ final File controllerPyFile = new File(pythonFrameworkDirectory, PYTHON_CONTROLLER_FILENAME);
+ final ProcessBuilder processBuilder = new ProcessBuilder(pythonCommand, controllerPyFile.getAbsolutePath());
+ processBuilder.environment().put("JAVA_PORT", String.valueOf(listeningPort));
+ processBuilder.environment().put("LOGS_DIR", pythonLogsDirectory.getAbsolutePath());
+ processBuilder.environment().put("ENV_HOME", virtualEnvHome.getAbsolutePath());
+ processBuilder.environment().put("PYTHONPATH", pythonApiDirectory.getAbsolutePath());
+ processBuilder.environment().put("PYTHON_CMD", pythonCommandFile.getAbsolutePath());
+ processBuilder.inheritIO();
+
+ logger.info("Launching Python Process {} {} with working directory {} to communicate with Java on Port {}",
+ pythonCommand, controllerPyFile.getAbsolutePath(), virtualEnvHome, listeningPort);
+ return processBuilder.start();
+ }
+
+
+ private void setupEnvironment() throws IOException {
+ final File environmentCreationCompleteFile = new File(virtualEnvHome, "env-creation-complete.txt");
+ if (environmentCreationCompleteFile.exists()) {
+ logger.debug("Environment has already been created for {}; will not recreate", virtualEnvHome);
+ return;
+ }
+
+ logger.info("Creating Python Virtual Environment {}", virtualEnvHome);
+
+ Files.createDirectories(virtualEnvHome.toPath());
+
+ final String pythonCommand = processConfig.getPythonCommand();
+ final String environmentPath = virtualEnvHome.getAbsolutePath();
+
+ final ProcessBuilder processBuilder = new ProcessBuilder(pythonCommand, "-m", "venv", environmentPath);
+ processBuilder.directory(virtualEnvHome.getParentFile());
+
+ final String command = String.join(" ", processBuilder.command());
+ logger.debug("Creating Python Virtual Environment {} using command {}", virtualEnvHome, command);
+ final Process process = processBuilder.start();
+
+ final int result;
+ try {
+ result = process.waitFor();
+ } catch (final InterruptedException e) {
+ throw new IOException("Interrupted while waiting for Python virtual environment to be created");
+ }
+
+ if (result != 0) {
+ throw new IOException("Failed to create Python Environment " + virtualEnvHome + ": process existed with code " + result);
+ }
+
+ // Create file so that we don't keep trying to recreate the virtual environment
+ environmentCreationCompleteFile.createNewFile();
+ logger.info("Successfully created Python Virtual Environment {}", virtualEnvHome);
+ }
+
+ public void shutdown() {
+ logger.info("Shutting down Python Process {}", process);
+
+ if (server != null) {
+ try {
+ server.shutdown();
+ } catch (final Exception e) {
+ logger.error("Failed to cleanly shutdown Py4J server", e);
+ }
+ }
+
+ if (gateway != null) {
+ try {
+ gateway.shutdown(true);
+ } catch (final Exception e) {
+ logger.error("Failed to cleanly shutdown Py4J Gateway", e);
+ }
+ }
+
+ if (process != null) {
+ try {
+ process.destroy();
Review Comment:
It seems like `Process.destroyForcibly()` might be a better option to help ensure process cleanup.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@nifi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org