This is an automated email from the ASF dual-hosted git repository.
markap14 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi.git
The following commit(s) were added to refs/heads/main by this push:
new 26f5fa2be0 NIFI-11443 Route Python Framework Logging to SLF4J (#8407)
26f5fa2be0 is described below
commit 26f5fa2be044accc6ae8403daf62be621e47f33e
Author: David Handermann <[email protected]>
AuthorDate: Tue Mar 5 15:55:04 2024 -0600
NIFI-11443 Route Python Framework Logging to SLF4J (#8407)
* NIFI-11443 Routed Python Framework Logging to SLF4J
- Changed Python logging to use standard output stream
- Adjusted Python logging format for simplified processing
- Updated PythonProcess to pipe standard error and standard output streams
to reader
- Added Log Reader command with Virtual Thread for each Python Process
- Removed Python log properties from NiFi Properties configuration
---
.../java/org/apache/nifi/util/NiFiProperties.java | 2 -
.../src/main/asciidoc/administration-guide.adoc | 3 -
.../org/apache/nifi/controller/FlowController.java | 4 -
.../nifi-framework/nifi-resources/pom.xml | 1 -
.../src/main/resources/conf/logback.xml | 3 +
.../src/main/resources/conf/nifi.properties | 1 -
.../nifi-py4j-bundle/nifi-py4j-bridge/pom.xml | 6 +
.../java/org/apache/nifi/py4j/PythonProcess.java | 46 ++++-
.../apache/nifi/py4j/PythonProcessLogReader.java | 176 +++++++++++++++++
.../org/apache/nifi/py4j/StandardPythonBridge.java | 6 +
.../nifi/py4j/logback/LevelChangeListener.java | 145 ++++++++++++++
.../nifi/py4j/logging/LogLevelChangeHandler.java | 37 ++++
.../nifi/py4j/logging/LogLevelChangeListener.java | 32 +++
.../apache/nifi/py4j/logging/PythonLogLevel.java | 67 +++++++
.../logging/StandardLogLevelChangeHandler.java | 100 ++++++++++
.../nifi/py4j/PythonProcessLogReaderTest.java | 220 +++++++++++++++++++++
.../nifi/py4j/logback/LevelChangeListenerTest.java | 67 +++++++
.../PythonControllerInteractionIT.java | 1 -
.../org/apache/nifi/python/PythonController.java | 8 +
.../apache/nifi/python/PythonProcessConfig.java | 25 ---
.../src/main/python/framework/Controller.py | 19 +-
.../resources/conf/clustered/node1/nifi.properties | 1 -
.../resources/conf/clustered/node2/nifi.properties | 1 -
.../test/resources/conf/default/nifi.properties | 1 -
.../src/test/resources/conf/pythonic/logback.xml | 3 +
.../test/resources/conf/pythonic/nifi.properties | 1 -
26 files changed, 919 insertions(+), 57 deletions(-)
diff --git
a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
index 7109600631..01dd887190 100644
---
a/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
+++
b/nifi-commons/nifi-properties/src/main/java/org/apache/nifi/util/NiFiProperties.java
@@ -319,7 +319,6 @@ public class NiFiProperties extends ApplicationProperties {
public static final String PYTHON_FRAMEWORK_SOURCE_DIRECTORY =
"nifi.python.framework.source.directory";
public static final String PYTHON_EXTENSION_DIRECTORY_PREFIX =
"nifi.python.extensions.source.directory.";
public static final String PYTHON_WORKING_DIRECTORY =
"nifi.python.working.directory";
- public static final String PYTHON_LOGS_DIRECTORY =
"nifi.python.logs.directory";
public static final String PYTHON_MAX_PROCESSES =
"nifi.python.max.processes";
public static final String PYTHON_MAX_PROCESSES_PER_TYPE =
"nifi.python.max.processes.per.extension.type";
public static final String PYTHON_COMMS_TIMEOUT =
"nifi.python.comms.timeout";
@@ -327,7 +326,6 @@ public class NiFiProperties extends ApplicationProperties {
public static final String PYTHON_CONTROLLER_DEBUGPY_ENABLED =
"nifi.python.controller.debugpy.enabled";
public static final String PYTHON_CONTROLLER_DEBUGPY_PORT =
"nifi.python.controller.debugpy.port";
public static final String PYTHON_CONTROLLER_DEBUGPY_HOST =
"nifi.python.controller.debugpy.host";
- public static final String PYTHON_CONTROLLER_DEBUGPY_LOGS_DIR =
"nifi.python.controller.debugpy.logs.directory";
// kubernetes properties
public static final String CLUSTER_LEADER_ELECTION_KUBERNETES_LEASE_PREFIX
= "nifi.cluster.leader.election.kubernetes.lease.prefix";
diff --git a/nifi-docs/src/main/asciidoc/administration-guide.adoc
b/nifi-docs/src/main/asciidoc/administration-guide.adoc
index db31ffa366..72e073c652 100644
--- a/nifi-docs/src/main/asciidoc/administration-guide.adoc
+++ b/nifi-docs/src/main/asciidoc/administration-guide.adoc
@@ -247,9 +247,6 @@ located as a sibling of this directory. For example, if the
value of this proper
by default, but multiple Python Extension directories can be added by adding
additional properties with the prefix
`nifi.python.extensions.source.directory.`.
| nifi.python.working.directory | ./work/python | The working directory where
NiFi should store artifacts, such as any third-party libraries that are
downloaded as dependencies
for the Python Processors.
-| nifi.python.logs.directory | ./logs | The directory where NiFi should store
the logs for the Python processes. If Python Processors use the in-built
logger, the messages will be logged to the
-app-log. However, if they create their own Python logger, the messages will be
written to the Python-specific log file(s). Additionally, the Python framework
will write to the Python-specific log
-file(s).
| nifi.python.max.processes.per.extension.type | 10 | The maximum number of
Python processes that should be spawned for any one type of Processor. Because
Python does not scale vertically,
adding many NiFi Processors within the same Python process would yield very
poor performance. Instead, NiFi creates a Python process for every Python
Processor that is added to the canvas,
within limits. This property indicates the maximum number of Python processes
that can be created for any particular type of Processor. For example, if there
are 5 instances of the
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
index 17e8c9e013..727f71f25f 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/FlowController.java
@@ -871,7 +871,6 @@ public class FlowController implements
ReportingTaskProvider, FlowAnalysisRulePr
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);
@@ -879,7 +878,6 @@ public class FlowController implements
ReportingTaskProvider, FlowAnalysisRulePr
final boolean enableControllerDebug =
Boolean.parseBoolean(nifiProperties.getProperty(NiFiProperties.PYTHON_CONTROLLER_DEBUGPY_ENABLED,
"false"));
final int debugPort =
nifiProperties.getIntegerProperty(NiFiProperties.PYTHON_CONTROLLER_DEBUGPY_PORT,
5678);
final String debugHost =
nifiProperties.getProperty(NiFiProperties.PYTHON_CONTROLLER_DEBUGPY_HOST,
"localhost");
- final String debugLogs =
nifiProperties.getProperty(NiFiProperties.PYTHON_CONTROLLER_DEBUGPY_LOGS_DIR,
"logs");
// Validate configuration for max numbers of processes.
if (maxProcessesPerType < 1) {
@@ -906,7 +904,6 @@ public class FlowController implements
ReportingTaskProvider, FlowAnalysisRulePr
.pythonCommand(pythonCommand)
.pythonFrameworkDirectory(pythonFrameworkSourceDirectory)
.pythonExtensionsDirectories(pythonExtensionsDirectories)
- .pythonLogsDirectory(pythonLogsDirectory)
.pythonWorkingDirectory(pythonWorkingDirectory)
.commsTimeout(commsTimeout == null ? null :
Duration.ofMillis(FormatUtils.getTimeDuration(commsTimeout,
TimeUnit.MILLISECONDS)))
.maxPythonProcesses(maxProcesses)
@@ -914,7 +911,6 @@ public class FlowController implements
ReportingTaskProvider, FlowAnalysisRulePr
.enableControllerDebug(enableControllerDebug)
.debugPort(debugPort)
.debugHost(debugHost)
- .debugLogsDirectory(new File(debugLogs))
.build();
final ControllerServiceTypeLookup serviceTypeLookup =
serviceProvider::getControllerServiceType;
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml
index b973224863..56aeebfd4a 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml
@@ -268,7 +268,6 @@
<nifi.python.working.directory>./work/python</nifi.python.working.directory>
<nifi.python.max.processes>100</nifi.python.max.processes>
<nifi.python.max.processes.per.extension.type>10</nifi.python.max.processes.per.extension.type>
- <nifi.python.logs.dir>./logs</nifi.python.logs.dir>
<nifi.cluster.leader.election.kubernetes.lease.prefix />
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml
index ae33e0484e..cbeff47ef6 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/logback.xml
@@ -174,6 +174,9 @@
<logger
name="org.apache.nifi.controller.repository.StandardProcessSession"
level="WARN" />
<logger name="org.apache.parquet.hadoop.InternalParquetRecordReader"
level="WARN" />
+ <!-- Py4J set to WARN to avoid verbose socket communication messages -->
+ <logger name="py4j" level="WARN" />
+
<logger name="org.apache.zookeeper.ClientCnxn" level="ERROR" />
<logger name="org.apache.zookeeper.server.NIOServerCnxn" level="ERROR" />
<logger name="org.apache.zookeeper.server.NIOServerCnxnFactory"
level="ERROR" />
diff --git
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
index 922e4f9dea..3147152535 100644
---
a/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
+++
b/nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties
@@ -49,7 +49,6 @@
nifi.python.extensions.source.directory.default=${nifi.python.extensions.source.
nifi.python.working.directory=${nifi.python.working.directory}
nifi.python.max.processes=${nifi.python.max.processes}
nifi.python.max.processes.per.extension.type=${nifi.python.max.processes.per.extension.type}
-nifi.python.logs.directory=${nifi.python.logs.dir}
####################
# State Management #
diff --git a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/pom.xml
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/pom.xml
index 655abc75f2..c7c05cad9f 100644
--- a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/pom.xml
+++ b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/pom.xml
@@ -42,6 +42,12 @@
<artifactId>py4j</artifactId>
<version>0.10.9.7</version>
</dependency>
+ <!-- Logback provided dependency for Log Level Change notification -->
+ <dependency>
+ <groupId>ch.qos.logback</groupId>
+ <artifactId>logback-classic</artifactId>
+ <scope>provided</scope>
+ </dependency>
<dependency>
<groupId>org.apache.nifi</groupId>
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcess.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcess.java
index 2180681c0d..f34199b5bc 100644
---
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcess.java
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcess.java
@@ -17,6 +17,10 @@
package org.apache.nifi.py4j;
+import org.apache.nifi.py4j.logging.LogLevelChangeListener;
+import org.apache.nifi.py4j.logging.PythonLogLevel;
+import org.apache.nifi.py4j.logging.StandardLogLevelChangeHandler;
+import org.apache.nifi.logging.LogLevel;
import org.apache.nifi.py4j.client.JavaObjectBindings;
import org.apache.nifi.py4j.client.NiFiPythonGateway;
import org.apache.nifi.py4j.client.StandardPythonClient;
@@ -36,6 +40,7 @@ import javax.net.ServerSocketFactory;
import javax.net.SocketFactory;
import java.io.File;
import java.io.IOException;
+import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.security.SecureRandom;
import java.util.ArrayList;
@@ -52,6 +57,8 @@ public class PythonProcess {
private static final Logger logger =
LoggerFactory.getLogger(PythonProcess.class);
private static final String PYTHON_CONTROLLER_FILENAME = "Controller.py";
+ private static final String LOG_READER_THREAD_NAME_FORMAT =
"python-log-%d";
+
private final PythonProcessConfig processConfig;
private final ControllerServiceTypeLookup controllerServiceTypeLookup;
private final File virtualEnvHome;
@@ -66,6 +73,8 @@ public class PythonProcess {
private volatile boolean shutdown = false;
private volatile List<String> extensionDirs;
private volatile String workDir;
+ private Thread logReaderThread;
+ private String logListenerId;
public PythonProcess(final PythonProcessConfig processConfig, final
ControllerServiceTypeLookup controllerServiceTypeLookup, final File
virtualEnvHome,
@@ -118,6 +127,10 @@ public class PythonProcess {
this.process = launchPythonProcess(listeningPort, authToken);
this.process.onExit().thenAccept(this::handlePythonProcessDied);
+ final String logReaderThreadName =
LOG_READER_THREAD_NAME_FORMAT.formatted(process.pid());
+ final Runnable logReaderCommand = new
PythonProcessLogReader(process.inputReader(StandardCharsets.UTF_8));
+ this.logReaderThread =
Thread.ofVirtual().name(logReaderThreadName).start(logReaderCommand);
+
final StandardPythonClient pythonClient = new
StandardPythonClient(gateway);
controller = pythonClient.getController();
@@ -151,6 +164,9 @@ public class PythonProcess {
throw new RuntimeException("Failed to start Python Bridge",
lastException);
}
+ logListenerId = Long.toString(process.pid());
+ StandardLogLevelChangeHandler.getHandler().addListener(logListenerId,
new PythonProcessLogLevelChangeListener());
+
controller.setControllerServiceTypeLookup(controllerServiceTypeLookup);
logger.info("Successfully started and pinged Python Server. Python
Process = {}", process);
}
@@ -211,7 +227,6 @@ public class PythonProcess {
private Process launchPythonProcess(final int listeningPort, final String
authToken) 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);
@@ -225,14 +240,12 @@ public class PythonProcess {
String pythonPath = pythonApiDirectory.getAbsolutePath();
-
if (processConfig.isDebugController() &&
"Controller".equals(componentId)) {
commands.add("-m");
commands.add("debugpy");
commands.add("--listen");
commands.add(processConfig.getDebugHost() + ":" +
processConfig.getDebugPort());
- commands.add("--log-to");
-
commands.add(processConfig.getDebugLogsDirectory().getAbsolutePath());
+ commands.add("--log-to-stderr");
pythonPath = pythonPath + File.pathSeparator +
virtualEnvHome.getAbsolutePath();
}
@@ -241,12 +254,13 @@ public class PythonProcess {
processBuilder.command(commands);
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", pythonPath);
processBuilder.environment().put("PYTHON_CMD",
pythonCommandFile.getAbsolutePath());
processBuilder.environment().put("AUTH_TOKEN", authToken);
- processBuilder.inheritIO();
+
+ // Redirect error stream to standard output stream
+ processBuilder.redirectErrorStream(true);
logger.info("Launching Python Process {} {} with working directory {}
to communicate with Java on Port {}",
pythonCommand, controllerPyFile.getAbsolutePath(), virtualEnvHome,
listeningPort);
@@ -329,6 +343,8 @@ public class PythonProcess {
}
private synchronized void killProcess() {
+
StandardLogLevelChangeHandler.getHandler().removeListener(logListenerId);
+
if (server != null) {
try {
server.shutdown();
@@ -358,6 +374,10 @@ public class PythonProcess {
process = null;
}
+
+ if (logReaderThread != null) {
+ logReaderThread.interrupt();
+ }
}
public void discoverExtensions(final List<String> directories, final
String workDirectory) {
@@ -433,4 +453,18 @@ public class PythonProcess {
}
private record CreatedProcessor(String identifier, String type,
PythonProcessorBridge processorBridge) {}
+
+ private class PythonProcessLogLevelChangeListener implements
LogLevelChangeListener {
+ /**
+ * Publish log level changes to Python Controller with conversion from
framework log level to Python log level
+ *
+ * @param loggerName Name of logger with updated level
+ * @param logLevel New log level
+ */
+ @Override
+ public void onLevelChange(final String loggerName, final LogLevel
logLevel) {
+ final PythonLogLevel pythonLogLevel =
PythonLogLevel.valueOf(logLevel);
+ controller.setLoggerLevel(loggerName, pythonLogLevel.getLevel());
+ }
+ }
}
\ No newline at end of file
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcessLogReader.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcessLogReader.java
new file mode 100644
index 0000000000..4a52bc35db
--- /dev/null
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/PythonProcessLogReader.java
@@ -0,0 +1,176 @@
+/*
+ * 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.logging.PythonLogLevel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Queue;
+import java.util.stream.Collectors;
+
+/**
+ * Runnable Command for reading a line from Process Output Stream and writing
to a Logger
+ */
+class PythonProcessLogReader implements Runnable {
+ private static final int LOG_LEVEL_BEGIN_INDEX = 0;
+
+ private static final int LOG_LEVEL_END_INDEX = 2;
+
+ private static final int MESSAGE_BEGIN_INDEX = 3;
+
+ private static final char NAME_MESSAGE_SEPARATOR = ':';
+
+ private static final int MINIMUM_LOGGER_NAME_INDEX = 3;
+
+ private static final String LOG_PREFIX = "PY4JLOG";
+
+ private static final int PREFIXED_LOG_LEVEL_BEGIN_INDEX = 8;
+
+ private static final String LINE_SEPARATOR = System.lineSeparator();
+
+ private static final Map<String, PythonLogLevel> PYTHON_LOG_LEVELS =
Arrays.stream(PythonLogLevel.values()).collect(
+ Collectors.toUnmodifiableMap(
+ pythonLogLevel ->
Integer.toString(pythonLogLevel.getLevel()),
+ pythonLogLevel -> pythonLogLevel
+ )
+ );
+
+ private final Logger processLogger =
LoggerFactory.getLogger("org.apache.nifi.py4j.ProcessLog");
+
+ private final BufferedReader processReader;
+
+ /**
+ * Standard constructor with Buffered Reader connected to Python Process
Output Stream
+ *
+ * @param processReader Reader from Process Output Stream
+ */
+ PythonProcessLogReader(final BufferedReader processReader) {
+ this.processReader = Objects.requireNonNull(processReader, "Reader
required");
+ }
+
+ /**
+ * Read lines from Process Reader and write log messages based on parsed
level and named logger
+ */
+ @Override
+ public void run() {
+ final Queue<ParsedRecord> parsedRecords = new ArrayDeque<>();
+
+ try {
+ String line = processReader.readLine();
+ while (line != null) {
+ try {
+ processLine(line, parsedRecords);
+
+ if (parsedRecords.size() == 2) {
+ // Log previous record after creating a new record
+ final ParsedRecord parsedRecord =
parsedRecords.remove();
+ log(parsedRecord);
+ }
+
+ if (!processReader.ready()) {
+ // Log queued records when Process Reader is not ready
+ while (!parsedRecords.isEmpty()) {
+ final ParsedRecord parsedRecord =
parsedRecords.poll();
+ log(parsedRecord);
+ }
+ }
+ } catch (final Exception e) {
+ processLogger.error("Failed to handle log from Python
Process", e);
+ }
+
+ // Read and block for subsequent lines
+ line = processReader.readLine();
+ }
+ } catch (final IOException e) {
+ processLogger.error("Failed to read output of Python Process", e);
+ }
+
+ // Handle last buffered message following closure of process stream
+ for (final ParsedRecord parsedRecord : parsedRecords) {
+ log(parsedRecord);
+ }
+ }
+
+ private void processLine(final String line, final Queue<ParsedRecord>
parsedRecords) {
+ final int logPrefixIndex = line.indexOf(LOG_PREFIX);
+ if (logPrefixIndex == 0) {
+ final String levelLogLine =
line.substring(PREFIXED_LOG_LEVEL_BEGIN_INDEX);
+ final String levelNumber =
levelLogLine.substring(LOG_LEVEL_BEGIN_INDEX, LOG_LEVEL_END_INDEX);
+ final PythonLogLevel logLevel =
PYTHON_LOG_LEVELS.getOrDefault(levelNumber, PythonLogLevel.NOTSET);
+ final String loggerMessage =
levelLogLine.substring(MESSAGE_BEGIN_INDEX);
+
+ final int nameSeparatorIndex =
loggerMessage.indexOf(NAME_MESSAGE_SEPARATOR);
+ final String message;
+ final Logger logger;
+ if (nameSeparatorIndex < MINIMUM_LOGGER_NAME_INDEX) {
+ // Set ProcessLog when named logger not found
+ logger = processLogger;
+ message = loggerMessage;
+ } else {
+ final String loggerName = loggerMessage.substring(0,
nameSeparatorIndex);
+ logger = LoggerFactory.getLogger(loggerName);
+
+ final int messageBeginIndex = nameSeparatorIndex + 1;
+ message = loggerMessage.substring(messageBeginIndex);
+ }
+
+ final StringBuilder buffer = new StringBuilder(message);
+ final ParsedRecord parsedRecord = new ParsedRecord(logLevel,
logger, buffer);
+ parsedRecords.add(parsedRecord);
+ } else {
+ final ParsedRecord lastRecord = parsedRecords.peek();
+ if (lastRecord == null) {
+ final StringBuilder buffer = new StringBuilder(line);
+ final ParsedRecord parsedRecord = new
ParsedRecord(PythonLogLevel.INFO, processLogger, buffer);
+ parsedRecords.add(parsedRecord);
+ } else if (!line.isEmpty()) {
+ // Add line separator for buffering multiple lines in the same
record
+ lastRecord.buffer.append(LINE_SEPARATOR);
+ lastRecord.buffer.append(line);
+ }
+ }
+ }
+
+ private void log(final ParsedRecord parsedRecord) {
+ final PythonLogLevel logLevel = parsedRecord.level;
+ final Logger logger = parsedRecord.logger;
+ final String message = parsedRecord.buffer.toString();
+
+ if (PythonLogLevel.DEBUG == logLevel) {
+ logger.debug(message);
+ } else if (PythonLogLevel.INFO == logLevel) {
+ logger.info(message);
+ } else if (PythonLogLevel.WARNING == logLevel) {
+ logger.warn(message);
+ } else if (PythonLogLevel.ERROR == logLevel) {
+ logger.error(message);
+ } else if (PythonLogLevel.CRITICAL == logLevel) {
+ logger.error(message);
+ } else {
+ logger.warn(message);
+ }
+ }
+
+ private record ParsedRecord(PythonLogLevel level, Logger logger,
StringBuilder buffer) {}
+}
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/StandardPythonBridge.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/StandardPythonBridge.java
index bb3cbe496a..cc2a5d8489 100644
---
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/StandardPythonBridge.java
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/StandardPythonBridge.java
@@ -18,6 +18,9 @@
package org.apache.nifi.py4j;
import org.apache.nifi.components.AsyncLoadedProcessor;
+import org.apache.nifi.py4j.logback.LevelChangeListener;
+import org.apache.nifi.py4j.logging.LogLevelChangeHandler;
+import org.apache.nifi.py4j.logging.StandardLogLevelChangeHandler;
import org.apache.nifi.python.BoundObjectCounts;
import org.apache.nifi.python.ControllerServiceTypeLookup;
import org.apache.nifi.python.PythonBridge;
@@ -71,6 +74,9 @@ public class StandardPythonBridge implements PythonBridge {
logger.debug("{} launching Python Process", this);
try {
+ final LogLevelChangeHandler logLevelChangeHandler =
StandardLogLevelChangeHandler.getHandler();
+ LevelChangeListener.registerLogbackListener(logLevelChangeHandler);
+
final File envHome = new
File(processConfig.getPythonWorkingDirectory(), "controller");
controllerProcess = new PythonProcess(processConfig,
serviceTypeLookup, envHome, "Controller", "Controller");
controllerProcess.start();
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logback/LevelChangeListener.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logback/LevelChangeListener.java
new file mode 100644
index 0000000000..8efc49b36c
--- /dev/null
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logback/LevelChangeListener.java
@@ -0,0 +1,145 @@
+/*
+ * 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.logback;
+
+import ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import ch.qos.logback.classic.LoggerContext;
+import ch.qos.logback.classic.spi.LoggerContextListener;
+import org.apache.nifi.py4j.logging.LogLevelChangeListener;
+import org.apache.nifi.logging.LogLevel;
+import org.slf4j.ILoggerFactory;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Logback Listener implementation for tracking Logger Level changes and
notifying framework listeners
+ */
+public class LevelChangeListener implements LoggerContextListener {
+ private static final Map<Level, LogLevel> LEVELS = Map.of(
+ Level.ALL, LogLevel.TRACE,
+ Level.TRACE, LogLevel.TRACE,
+ Level.DEBUG, LogLevel.DEBUG,
+ Level.INFO, LogLevel.INFO,
+ Level.WARN, LogLevel.WARN,
+ Level.ERROR, LogLevel.ERROR,
+ Level.OFF, LogLevel.NONE
+ );
+
+ private static final List<Pattern> EXCLUDED_LOGGERS = List.of(
+ Pattern.compile("^$"),
+ Pattern.compile("^org|com|net$"),
+ Pattern.compile("^jetbrains.*"),
+ Pattern.compile("^org\\.apache.*"),
+ Pattern.compile("^org\\.eclipse.*"),
+ Pattern.compile("^org\\.glassfish.*"),
+ Pattern.compile("^org\\.springframework.*"),
+ Pattern.compile("^org\\.opensaml.*"),
+ Pattern.compile("^software\\.amazon.*")
+ );
+
+ private final LogLevelChangeListener logLevelChangeListener;
+
+ LevelChangeListener(final LogLevelChangeListener logLevelChangeListener) {
+ this.logLevelChangeListener =
Objects.requireNonNull(logLevelChangeListener, "Listener required");
+ }
+
+ /**
+ * Register an instance of Logback LevelChangeListener routing to
framework LogLevelChangeListener
+ */
+ public static void registerLogbackListener(final LogLevelChangeListener
logLevelChangeListener) {
+ final LevelChangeListener levelChangeListener = new
LevelChangeListener(logLevelChangeListener);
+ final ILoggerFactory loggerFactory = LoggerFactory.getILoggerFactory();
+ if (loggerFactory instanceof LoggerContext loggerContext) {
+ loggerContext.addListener(levelChangeListener);
+ levelChangeListener.onStart(loggerContext);
+ }
+ }
+
+ /**
+ * Reset Resistant enabled to support persistent behavior when the
LoggerContext is changed
+ *
+ * @return Reset Resistant enabled
+ */
+ @Override
+ public boolean isResetResistant() {
+ return true;
+ }
+
+ /**
+ * Set initial Logger Levels from registered loggers when starting
+ *
+ * @param loggerContext Logback Context
+ */
+ @Override
+ public void onStart(final LoggerContext loggerContext) {
+ for (final Logger logger : loggerContext.getLoggerList()) {
+ onLevelChange(logger, logger.getEffectiveLevel());
+ }
+ }
+
+ /**
+ * Set Logger Levels from registered loggers when resetting
+ *
+ * @param loggerContext Logback Context
+ */
+ @Override
+ public void onReset(final LoggerContext loggerContext) {
+ onStart(loggerContext);
+ }
+
+ @Override
+ public void onStop(final LoggerContext loggerContext) {
+
+ }
+
+ /**
+ * Send logger level changes to framework handler for subsequent
notification
+ *
+ * @param logger Logback Logger with new level
+ * @param level New level assigned for Logback Logger
+ */
+ @Override
+ public void onLevelChange(final Logger logger, final Level level) {
+ Objects.requireNonNull(level, "Level required");
+
+ final String loggerName = logger.getName();
+ if (isLoggerSupported(loggerName)) {
+ final LogLevel logLevel = LEVELS.getOrDefault(level,
LogLevel.WARN);
+ logLevelChangeListener.onLevelChange(loggerName, logLevel);
+ }
+ }
+
+ private boolean isLoggerSupported(final String loggerName) {
+ boolean supported = true;
+
+ for (final Pattern loggerPattern : EXCLUDED_LOGGERS) {
+ final Matcher matcher = loggerPattern.matcher(loggerName);
+ if (matcher.matches()) {
+ supported = false;
+ break;
+ }
+ }
+
+ return supported;
+ }
+}
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logging/LogLevelChangeHandler.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logging/LogLevelChangeHandler.java
new file mode 100644
index 0000000000..3cf50a99c0
--- /dev/null
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logging/LogLevelChangeHandler.java
@@ -0,0 +1,37 @@
+/*
+ * 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.logging;
+
+/**
+ * Handler abstraction for registering Listeners and invoking Listeners on
level changes
+ */
+public interface LogLevelChangeHandler extends LogLevelChangeListener {
+ /**
+ * Register Log Level Change Listener with provided identifier
+ *
+ * @param identifier Tracking identifier associated with Log Level Change
Listener
+ * @param listener Log Level Change Listener to be registered
+ */
+ void addListener(String identifier, LogLevelChangeListener listener);
+
+ /**
+ * Remove registered listener based on provided identifier
+ *
+ * @param identifier Tracking identifier of registered listener to be
removed
+ */
+ void removeListener(String identifier);
+}
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logging/LogLevelChangeListener.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logging/LogLevelChangeListener.java
new file mode 100644
index 0000000000..d9b2077f4d
--- /dev/null
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logging/LogLevelChangeListener.java
@@ -0,0 +1,32 @@
+/*
+ * 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.logging;
+
+import org.apache.nifi.logging.LogLevel;
+
+/**
+ * Listener abstraction for subscribing to notification of level changes for
named loggers
+ */
+public interface LogLevelChangeListener {
+ /**
+ * Handle level change notification for named logger
+ *
+ * @param loggerName Name of logger with updated level
+ * @param logLevel New log level
+ */
+ void onLevelChange(String loggerName, LogLevel logLevel);
+}
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logging/PythonLogLevel.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logging/PythonLogLevel.java
new file mode 100644
index 0000000000..f32e5245c5
--- /dev/null
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logging/PythonLogLevel.java
@@ -0,0 +1,67 @@
+/*
+ * 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.logging;
+
+import org.apache.nifi.logging.LogLevel;
+
+import java.util.Map;
+
+/**
+ * Python Log Level enumeration with level numbers according to Python logging
module documentation
+ */
+public enum PythonLogLevel {
+ NOTSET(0),
+
+ DEBUG(10),
+
+ INFO(20),
+
+ WARNING(30),
+
+ ERROR(40),
+
+ CRITICAL(50);
+
+ private static final Map<LogLevel, PythonLogLevel> FRAMEWORK_LEVELS =
Map.of(
+ LogLevel.NONE, NOTSET,
+ LogLevel.DEBUG, DEBUG,
+ LogLevel.INFO, INFO,
+ LogLevel.WARN, WARNING,
+ LogLevel.ERROR, ERROR,
+ LogLevel.FATAL, CRITICAL
+ );
+
+ private final int level;
+
+ PythonLogLevel(final int level) {
+ this.level = level;
+ }
+
+ public int getLevel() {
+ return level;
+ }
+
+ /**
+ * Get Python Log Level equivalent from framework Log Level
+ *
+ * @param logLevel Framework Log Level
+ * @return Python Log Level
+ */
+ public static PythonLogLevel valueOf(LogLevel logLevel) {
+ return FRAMEWORK_LEVELS.getOrDefault(logLevel, WARNING);
+ }
+}
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logging/StandardLogLevelChangeHandler.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logging/StandardLogLevelChangeHandler.java
new file mode 100644
index 0000000000..150798c317
--- /dev/null
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/main/java/org/apache/nifi/py4j/logging/StandardLogLevelChangeHandler.java
@@ -0,0 +1,100 @@
+/*
+ * 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.logging;
+
+import org.apache.nifi.logging.LogLevel;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Standard implementation of Log Level Change Handler with singleton instance
for shared collection of listeners
+ */
+public class StandardLogLevelChangeHandler implements LogLevelChangeHandler {
+ private static final StandardLogLevelChangeHandler HANDLER = new
StandardLogLevelChangeHandler();
+
+ private static final Logger handlerLogger =
LoggerFactory.getLogger(StandardLogLevelChangeHandler.class);
+
+ private final Map<String, LogLevelChangeListener> listeners = new
ConcurrentHashMap<>();
+
+ private final Map<String, LogLevel> loggerLevels = new
ConcurrentHashMap<>();
+
+ private StandardLogLevelChangeHandler() {
+
+ }
+
+ /**
+ * Get shared reference to Handler for Listener registration and
centralized notification
+ *
+ * @return Shared Log Level Change Handler
+ */
+ public static LogLevelChangeHandler getHandler() {
+ return HANDLER;
+ }
+
+ /**
+ * Register Log Level Change Listener with provided identifier
+ *
+ * @param identifier Tracking identifier associated with Log Level Change
Listener
+ * @param listener Log Level Change Listener to be registered
+ */
+ @Override
+ public void addListener(final String identifier, final
LogLevelChangeListener listener) {
+ Objects.requireNonNull(identifier, "Identifier required");
+ Objects.requireNonNull(listener, "Listener required");
+
+ for (final Map.Entry<String, LogLevel> loggerLevel :
loggerLevels.entrySet()) {
+ listener.onLevelChange(loggerLevel.getKey(),
loggerLevel.getValue());
+ }
+
+ listeners.put(identifier, listener);
+ handlerLogger.trace("Added Listener [{}]", identifier);
+ }
+
+ /**
+ * Remove registered listener based on provided identifier
+ *
+ * @param identifier Tracking identifier of registered listener to be
removed
+ */
+ @Override
+ public void removeListener(String identifier) {
+ Objects.requireNonNull(identifier, "Identifier required");
+ listeners.remove(identifier);
+ handlerLogger.trace("Removed Listener [{}]", identifier);
+ }
+
+ /**
+ * Handle level change notification for named logger and broadcast to
registered Listeners
+ *
+ * @param loggerName Name of logger with updated level
+ * @param logLevel New log level
+ */
+ @Override
+ public void onLevelChange(final String loggerName, final LogLevel
logLevel) {
+ Objects.requireNonNull(loggerName, "Logger Name required");
+ Objects.requireNonNull(logLevel, "Log Level required");
+ handlerLogger.trace("Logger [{}] Level [{}] changed", loggerName,
logLevel);
+
+ loggerLevels.put(loggerName, logLevel);
+ for (final LogLevelChangeListener listener : listeners.values()) {
+ listener.onLevelChange(loggerName, logLevel);
+ }
+ }
+}
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/test/java/org/apache/nifi/py4j/PythonProcessLogReaderTest.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/test/java/org/apache/nifi/py4j/PythonProcessLogReaderTest.java
new file mode 100644
index 0000000000..2db386df9a
--- /dev/null
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/test/java/org/apache/nifi/py4j/PythonProcessLogReaderTest.java
@@ -0,0 +1,220 @@
+/*
+ * 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.logging.PythonLogLevel;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.MockedStatic;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.StringReader;
+
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isA;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class PythonProcessLogReaderTest {
+
+ private static final String PROCESS_LOGGER =
"org.apache.nifi.py4j.ProcessLog";
+
+ private static final String CONTROLLER_LOGGER =
"org.apache.nifi.py4j.Controller";
+
+ private static final String LOG_MESSAGE = "Testing Python Processing";
+
+ private static final String LINE_FORMAT = "PY4JLOG %s %s:%s %d%n";
+
+ private static final String LINE_UNFORMATTED = "Testing message without
level or logger";
+
+ private static final String LINE_MISSING_SEPARATOR = "%s
%s".formatted(PythonLogLevel.INFO.getLevel(), LOG_MESSAGE);
+
+ private static final String LINE_EMPTY = "";
+
+ private static final String MESSAGE_TRACEBACK = String.format("Command
Failed%nTrackback (most recent call last):%n File: command.py, line 1%nError:
name is not defined");
+
+ private static final int FIRST_LOG = 1;
+
+ private static final String FIRST_LOG_MESSAGE = String.format("%s %d",
LOG_MESSAGE, FIRST_LOG);
+
+ private static final String NUMBERED_LOG_FORMAT = "%s %d";
+
+ @Mock
+ private Logger processLogger;
+
+ @Mock
+ private Logger controllerLogger;
+
+ @Test
+ void testDebug() {
+ runCommand(PythonLogLevel.DEBUG, controllerLogger);
+
+ verify(controllerLogger).debug(eq(FIRST_LOG_MESSAGE));
+ }
+
+ @Test
+ void testInfo() {
+ runCommand(PythonLogLevel.INFO, controllerLogger);
+
+ verify(controllerLogger).info(eq(FIRST_LOG_MESSAGE));
+ }
+
+ @Test
+ void testWarning() {
+ runCommand(PythonLogLevel.WARNING, controllerLogger);
+
+ verify(controllerLogger).warn(eq(FIRST_LOG_MESSAGE));
+ }
+
+ @Test
+ void testError() {
+ runCommand(PythonLogLevel.ERROR, controllerLogger);
+
+ verify(controllerLogger).error(eq(FIRST_LOG_MESSAGE));
+ }
+
+ @Test
+ void testCritical() {
+ runCommand(PythonLogLevel.CRITICAL, controllerLogger);
+
+ verify(controllerLogger).error(eq(FIRST_LOG_MESSAGE));
+ }
+
+ @Test
+ void testUnformatted() {
+ try (MockedStatic<LoggerFactory> loggerFactory =
mockStatic(LoggerFactory.class)) {
+ setupLogger(loggerFactory, processLogger, PROCESS_LOGGER);
+
+ final BufferedReader reader = new BufferedReader(new
StringReader(LINE_UNFORMATTED));
+ final Runnable command = new PythonProcessLogReader(reader);
+ command.run();
+
+ verify(processLogger).info(eq(LINE_UNFORMATTED));
+ }
+ }
+
+ @Test
+ void testMissingSeparator() {
+ try (MockedStatic<LoggerFactory> loggerFactory =
mockStatic(LoggerFactory.class)) {
+ setupLogger(loggerFactory, processLogger, PROCESS_LOGGER);
+
+ final BufferedReader reader = new BufferedReader(new
StringReader(LINE_MISSING_SEPARATOR));
+ final Runnable command = new PythonProcessLogReader(reader);
+ command.run();
+
+ verify(processLogger).info(eq(LINE_MISSING_SEPARATOR));
+ }
+ }
+
+ @Test
+ void testEmpty() {
+ try (MockedStatic<LoggerFactory> loggerFactory =
mockStatic(LoggerFactory.class)) {
+ setupLogger(loggerFactory, processLogger, PROCESS_LOGGER);
+
+ final BufferedReader reader = new BufferedReader(new
StringReader(LINE_EMPTY));
+ final Runnable command = new PythonProcessLogReader(reader);
+ command.run();
+
+ verifyNoInteractions(processLogger);
+ }
+ }
+
+ @Test
+ void testInfoMultipleMessages() {
+ final int messages = 2;
+
+ try (MockedStatic<LoggerFactory> loggerFactory =
mockStatic(LoggerFactory.class)) {
+ setupLogger(loggerFactory, controllerLogger, CONTROLLER_LOGGER);
+
+ final StringBuilder builder = new StringBuilder();
+ for (int i = 0; i < messages; i++) {
+ builder.append(getLine(PythonLogLevel.INFO, i));
+ builder.append(System.lineSeparator());
+ }
+
+ final String lines = builder.toString();
+ final BufferedReader reader = new BufferedReader(new
StringReader(lines));
+ final Runnable command = new PythonProcessLogReader(reader);
+ command.run();
+ }
+
+ for (int i = 0; i < messages; i++) {
+ final String expected = NUMBERED_LOG_FORMAT.formatted(LOG_MESSAGE,
i);
+ verify(controllerLogger).info(eq(expected));
+ }
+ }
+
+ @Test
+ void testErrorMultipleLines() {
+ try (MockedStatic<LoggerFactory> loggerFactory =
mockStatic(LoggerFactory.class)) {
+ setupLogger(loggerFactory, controllerLogger, CONTROLLER_LOGGER);
+
+ final String lines =
LINE_FORMAT.formatted(PythonLogLevel.ERROR.getLevel(), CONTROLLER_LOGGER,
MESSAGE_TRACEBACK, FIRST_LOG);
+
+ final BufferedReader reader = new BufferedReader(new
StringReader(lines));
+ final Runnable command = new PythonProcessLogReader(reader);
+ command.run();
+ }
+
+ final String expected =
NUMBERED_LOG_FORMAT.formatted(MESSAGE_TRACEBACK, FIRST_LOG);
+ verify(controllerLogger).error(eq(expected));
+ }
+
+ @Test
+ void testReaderException() throws IOException {
+ try (MockedStatic<LoggerFactory> loggerFactory =
mockStatic(LoggerFactory.class)) {
+ setupLogger(loggerFactory, processLogger, PROCESS_LOGGER);
+
+ final BufferedReader reader = mock(BufferedReader.class);
+ when(reader.readLine()).thenThrow(new IOException());
+
+ final Runnable command = new PythonProcessLogReader(reader);
+ command.run();
+
+ verify(processLogger).error(anyString(), isA(IOException.class));
+ }
+ }
+
+ private void runCommand(final PythonLogLevel logLevel, final Logger
logger) {
+ try (MockedStatic<LoggerFactory> loggerFactory =
mockStatic(LoggerFactory.class)) {
+ setupLogger(loggerFactory, logger, CONTROLLER_LOGGER);
+
+ final String line = getLine(logLevel, FIRST_LOG);
+ final BufferedReader reader = new BufferedReader(new
StringReader(line));
+ final Runnable command = new PythonProcessLogReader(reader);
+ command.run();
+ }
+ }
+
+ private void setupLogger(final MockedStatic<LoggerFactory> loggerFactory,
final Logger logger, final String loggerName) {
+ loggerFactory.when(() ->
LoggerFactory.getLogger(eq(loggerName))).thenReturn(logger);
+ }
+
+ private String getLine(final PythonLogLevel logLevel, final int number) {
+ return LINE_FORMAT.formatted(logLevel.getLevel(), CONTROLLER_LOGGER,
LOG_MESSAGE, number);
+ }
+}
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/test/java/org/apache/nifi/py4j/logback/LevelChangeListenerTest.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/test/java/org/apache/nifi/py4j/logback/LevelChangeListenerTest.java
new file mode 100644
index 0000000000..6e4a4c06bb
--- /dev/null
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-bridge/src/test/java/org/apache/nifi/py4j/logback/LevelChangeListenerTest.java
@@ -0,0 +1,67 @@
+package org.apache.nifi.py4j.logback;/*
+ * 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 ch.qos.logback.classic.Level;
+import ch.qos.logback.classic.Logger;
+import org.apache.nifi.py4j.logging.LogLevelChangeListener;
+import org.apache.nifi.py4j.logging.StandardLogLevelChangeHandler;
+import org.apache.nifi.logging.LogLevel;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class LevelChangeListenerTest {
+
+ private static final String LOGGER_NAME = Test.class.getName();
+
+ @Mock
+ private Logger logger;
+
+ @Mock
+ private LogLevelChangeListener logLevelChangeListener;
+
+ private LevelChangeListener listener;
+
+ @BeforeEach
+ void setListener() {
+ listener = new
LevelChangeListener(StandardLogLevelChangeHandler.getHandler());
+
+
StandardLogLevelChangeHandler.getHandler().addListener(LogLevelChangeListener.class.getSimpleName(),
logLevelChangeListener);
+ }
+
+ @AfterEach
+ void removeListener() {
+
StandardLogLevelChangeHandler.getHandler().removeListener(LogLevelChangeListener.class.getSimpleName());
+ }
+
+ @Test
+ void testOnLevelChange() {
+ when(logger.getName()).thenReturn(LOGGER_NAME);
+
+ listener.onLevelChange(logger, Level.INFO);
+
+ verify(logLevelChangeListener).onLevelChange(eq(LOGGER_NAME),
eq(LogLevel.INFO));
+ }
+}
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-integration-tests/src/test/java/org.apache.nifi.py4j/PythonControllerInteractionIT.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-integration-tests/src/test/java/org.apache.nifi.py4j/PythonControllerInteractionIT.java
index c41d459a17..7fb0399166 100644
---
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-integration-tests/src/test/java/org.apache.nifi.py4j/PythonControllerInteractionIT.java
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-py4j-integration-tests/src/test/java/org.apache.nifi.py4j/PythonControllerInteractionIT.java
@@ -89,7 +89,6 @@ public class PythonControllerInteractionIT {
.commsTimeout(Duration.ofSeconds(0))
.maxPythonProcessesPerType(25)
.maxPythonProcesses(100)
- .pythonLogsDirectory(logsDir)
.build();
Files.createDirectories(logsDir.toPath());
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/PythonController.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/PythonController.java
index c5ae45bde2..b840603b9e 100644
---
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/PythonController.java
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/PythonController.java
@@ -99,4 +99,12 @@ public interface PythonController {
* @return the details that have been discovered
*/
PythonProcessorDetails getProcessorDetails(String type, String version);
+
+ /**
+ * Set level for specified logger name in Python logging.Logger objects
+ *
+ * @param loggerName Python logger name to be updated
+ * @param level Python log level according to Python logging module
documentation
+ */
+ void setLoggerLevel(String loggerName, int level);
}
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/PythonProcessConfig.java
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/PythonProcessConfig.java
index 72c8c85d4c..4e32b5e89d 100644
---
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/PythonProcessConfig.java
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework-api/src/main/java/org/apache/nifi/python/PythonProcessConfig.java
@@ -30,20 +30,17 @@ public class PythonProcessConfig {
private final File pythonFrameworkDirectory;
private final List<File> pythonExtensionsDirectories;
private final File pythonWorkingDirectory;
- private final File pythonLogsDirectory;
private final Duration commsTimeout;
private final int maxPythonProcesses;
private final int maxPythonProcessesPerType;
private final boolean debugController;
private final String debugHost;
private final int debugPort;
- private final File debugLogsDirectory;
private PythonProcessConfig(final Builder builder) {
this.pythonCommand = builder.pythonCommand;
this.pythonFrameworkDirectory = builder.pythonFrameworkDirectory;
this.pythonExtensionsDirectories = builder.pythonExtensionsDirectories;
- this.pythonLogsDirectory = builder.pythonLogsDirectory;
this.pythonWorkingDirectory = builder.pythonWorkingDirectory;
this.commsTimeout = builder.commsTimeout;
this.maxPythonProcesses = builder.maxProcesses;
@@ -51,7 +48,6 @@ public class PythonProcessConfig {
this.debugController = builder.debugController;
this.debugPort = builder.debugPort;
this.debugHost = builder.debugHost;
- this.debugLogsDirectory = builder.debugLogsDirectory;
}
public String getPythonCommand() {
@@ -66,10 +62,6 @@ public class PythonProcessConfig {
return pythonExtensionsDirectories;
}
- public File getPythonLogsDirectory() {
- return pythonLogsDirectory;
- }
-
public File getPythonWorkingDirectory() {
return pythonWorkingDirectory;
}
@@ -98,15 +90,10 @@ public class PythonProcessConfig {
return debugPort;
}
- public File getDebugLogsDirectory() {
- return debugLogsDirectory;
- }
-
public static class Builder {
private String pythonCommand = "python3";
private File pythonFrameworkDirectory = new File("python/framework");
private List<File> pythonExtensionsDirectories =
Collections.singletonList(new File("python/extensions"));
- private File pythonLogsDirectory = new File("./logs");
private File pythonWorkingDirectory = new File("python");
private Duration commsTimeout = Duration.ofSeconds(0);
private int maxProcesses;
@@ -114,8 +101,6 @@ public class PythonProcessConfig {
private boolean debugController = false;
private String debugHost = "localhost";
private int debugPort = 5678;
- private File debugLogsDirectory = new File("logs/");
-
public Builder pythonCommand(final String command) {
this.pythonCommand = command;
@@ -164,11 +149,6 @@ public class PythonProcessConfig {
return this;
}
- public Builder pythonLogsDirectory(final File logsDirectory) {
- this.pythonLogsDirectory = logsDirectory;
- return this;
- }
-
public Builder enableControllerDebug(final boolean enableDebug) {
this.debugController = enableDebug;
return this;
@@ -184,11 +164,6 @@ public class PythonProcessConfig {
return this;
}
- public Builder debugLogsDirectory(final File debugLogsDirectory) {
- this.debugLogsDirectory = debugLogsDirectory;
- return this;
- }
-
public PythonProcessConfig build() {
return new PythonProcessConfig(this);
}
diff --git
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework/src/main/python/framework/Controller.py
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework/src/main/python/framework/Controller.py
index 8919924c06..05e0d4f6b0 100644
---
a/nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework/src/main/python/framework/Controller.py
+++
b/nifi-nar-bundles/nifi-py4j-bundle/nifi-python-framework/src/main/python/framework/Controller.py
@@ -15,6 +15,7 @@
import logging
import os
+import sys
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
from py4j.java_gateway import JavaGateway, CallbackServerParameters,
GatewayParameters
@@ -29,19 +30,14 @@ import ExtensionManager
threadpool_attrs = dir(ThreadPoolExecutor)
processpool_attrs = dir(ProcessPoolExecutor)
-
-# 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 + '/nifi-python.log',
- format='%(asctime)s %(levelname)s %(name)s %(message)s',
+# Set log format with level number and separator as expected in
PythonProcessReaderCommand
+logging.basicConfig(stream=sys.stderr,
+ format='PY4JLOG %(levelno)s %(name)s:%(message)s',
encoding='utf-8',
level=logging.INFO)
+logger = logging.getLogger("org.apache.nifi.py4j.Controller")
+
class Controller:
@@ -92,6 +88,9 @@ class Controller:
def setControllerServiceTypeLookup(self, typeLookup):
self.controllerServiceTypeLookup = typeLookup
+ def setLoggerLevel(self, loggerName, level):
+ logging.getLogger(loggerName).setLevel(level)
+
class Java:
implements = ["org.apache.nifi.py4j.PythonController"]
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties
index ce4aa9faf1..f9ef29179d 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node1/nifi.properties
@@ -48,7 +48,6 @@
nifi.python.extensions.source.directory.default=./python/extensions
nifi.python.working.directory=./work/python
nifi.python.max.processes=100
nifi.python.max.processes.per.extension.type=10
-nifi.python.logs.directory=./logs
####################
# State Management #
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties
index d507b92b22..8b45f32b17 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/clustered/node2/nifi.properties
@@ -48,7 +48,6 @@
nifi.python.extensions.source.directory.default=./python/extensions
nifi.python.working.directory=./work/python
nifi.python.max.processes=100
nifi.python.max.processes.per.extension.type=10
-nifi.python.logs.directory=./logs
####################
# State Management #
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties
index 6ce80350c4..4758d714dc 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/default/nifi.properties
@@ -48,7 +48,6 @@
nifi.python.extensions.source.directory.default=./python/extensions
nifi.python.working.directory=./work/python
nifi.python.max.processes=100
nifi.python.max.processes.per.extension.type=10
-nifi.python.logs.directory=./logs
####################
# State Management #
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/logback.xml
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/logback.xml
index 806958ddbf..a3047818dd 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/logback.xml
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/logback.xml
@@ -115,6 +115,9 @@
<logger name="org.apache.calcite.runtime.CalciteException" level="OFF" />
<logger name="deprecation" level="OFF" />
+ <!-- Py4J set to WARN to avoid verbose socket communication messages -->
+ <logger name="py4j" level="WARN" />
+
<logger name="org.apache.curator.framework.recipes.leader.LeaderSelector"
level="OFF" />
<logger name="org.apache.curator.ConnectionState" level="OFF" />
diff --git
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/nifi.properties
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/nifi.properties
index 0069027b3e..faf92c6a96 100644
---
a/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/nifi.properties
+++
b/nifi-system-tests/nifi-system-test-suite/src/test/resources/conf/pythonic/nifi.properties
@@ -52,7 +52,6 @@
nifi.python.extensions.source.directory.default=./python/extensions
nifi.python.working.directory=./work/python
nifi.python.max.processes=100
nifi.python.max.processes.per.extension.type=10
-nifi.python.logs.directory=./logs
####################
# State Management #
