You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@nifi.apache.org by joewitt <gi...@git.apache.org> on 2016/02/01 22:59:50 UTC
[GitHub] nifi pull request: NIFI-210: Add ExecuteScript and InvokeScriptedP...
Github user joewitt commented on a diff in the pull request:
https://github.com/apache/nifi/pull/185#discussion_r51488029
--- Diff: nifi-nar-bundles/nifi-scripting-bundle/nifi-scripting-processors/src/main/java/org/apache/nifi/processors/script/InvokeScriptedProcessor.java ---
@@ -0,0 +1,606 @@
+/*
+ * 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.processors.script;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.annotation.behavior.DynamicProperty;
+import org.apache.nifi.annotation.documentation.CapabilityDescription;
+import org.apache.nifi.annotation.documentation.SeeAlso;
+import org.apache.nifi.annotation.documentation.Tags;
+import org.apache.nifi.annotation.lifecycle.OnScheduled;
+import org.apache.nifi.annotation.lifecycle.OnStopped;
+import org.apache.nifi.components.PropertyDescriptor;
+import org.apache.nifi.components.ValidationContext;
+import org.apache.nifi.components.ValidationResult;
+import org.apache.nifi.controller.ControllerServiceLookup;
+import org.apache.nifi.logging.ProcessorLog;
+import org.apache.nifi.processor.ProcessContext;
+import org.apache.nifi.processor.ProcessSessionFactory;
+import org.apache.nifi.processor.Processor;
+import org.apache.nifi.processor.ProcessorInitializationContext;
+import org.apache.nifi.processor.Relationship;
+import org.apache.nifi.processor.exception.ProcessException;
+import org.apache.nifi.processor.util.StandardValidators;
+import org.apache.nifi.util.file.monitor.LastModifiedMonitor;
+import org.apache.nifi.util.file.monitor.SynchronousFileWatcher;
+
+import javax.script.Invocable;
+import java.io.FileInputStream;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+@Tags({"script", "invoke", "groovy", "python", "jython", "jruby", "ruby", "javascript", "js", "lua", "luaj"})
+@CapabilityDescription("Experimental - Invokes a script engine for a Processor defined in the given script. The script must define "
+ + "a valid class that implements the Processor interface, and it must set a variable 'processor' to an instance of "
+ + "the class. Processor methods such as onTrigger() will be delegated to the scripted Processor instance. Also any "
+ + "Relationships or PropertyDescriptors defined by the scripted processor will be added to the configuration dialog. "
+ + "Experimental: Impact of sustained usage not yet verified.")
+@DynamicProperty(name = "A script engine property to update", value = "The value to set it to", supportsExpressionLanguage = true,
+ description = "Updates a script engine property specified by the Dynamic Property's key with the value specified by the Dynamic Property's value")
+@SeeAlso({ExecuteScript.class})
+public class InvokeScriptedProcessor extends AbstractScriptProcessor {
+
+ private final AtomicReference<Processor> processor = new AtomicReference<>();
+ private final AtomicReference<Collection<ValidationResult>> validationResults =
+ new AtomicReference<>((Collection<ValidationResult>) new ArrayList<ValidationResult>());
+
+ private final Lock lock = new ReentrantLock();
+ private SynchronousFileWatcher scriptWatcher;
+
+ private ScheduledExecutorService reloadService = null;
+
+ /**
+ * Creates the resources needed by this processor. An attempt is made to also initialize the scripted processor,
+ * but unless the properties (such as script engine name and script file path) have already been specified, the
+ * script will not yet have been evaluated, so the script's initialize() method will not be called.
+ */
+ protected void createResources() {
+
+ // Set up script file reloader service. This checks to see if the script file has changed, and if so, tries
+ // to reload it
+ if (reloadService == null) {
+ reloadService = Executors.newScheduledThreadPool(1);
+
+ // monitor the script if configured for changes
+ reloadService.scheduleWithFixedDelay(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ final boolean hasLock = lock.tryLock();
+
+ // if a property is changing we don't need to reload this iteration
+ if (hasLock) {
+ try {
+ if (scriptWatcher != null && scriptWatcher.checkAndReset()) {
+ if (isFile(scriptPath)) {
+ // reload the actual script
+ final boolean reloaded = reloadScriptFile(scriptPath);
+
+ // log the script was reloaded
+ if (reloaded) {
+ getLogger().info("The configured script has been successfully reloaded.");
+ }
+ }
+ }
+ } finally {
+ lock.unlock();
+ }
+ }
+ } catch (final Throwable t) {
+ final ProcessorLog logger = getLogger();
+ final String message = "Unable to reload configured script Processor: " + t;
+
+ logger.error(message);
+ if (logger.isDebugEnabled()) {
+ logger.error(message, t);
+ }
+ }
+ }
+ }, 30, 10, TimeUnit.SECONDS);
+ }
+
+ super.createResources();
+ }
+
+ /**
+ * Returns the valid relationships for this processor. SUCCESS and FAILURE are always returned, and if the script
+ * processor has defined additional relationships, those will be added as well.
+ *
+ * @return a Set of Relationships supported by this processor
+ */
+ @Override
+ public Set<Relationship> getRelationships() {
+ final Set<Relationship> relationships = new HashSet<>();
+ final Processor instance = processor.get();
+ if (instance != null) {
+ try {
+ relationships.addAll(instance.getRelationships());
+ } catch (final Throwable t) {
+ final ProcessorLog logger = getLogger();
+ final String message = "Unable to get relationships from scripted Processor: " + t;
+
+ logger.error(message);
+ if (logger.isDebugEnabled()) {
+ logger.error(message, t);
+ }
+ }
+ } else {
+ // Return defaults for now
+ relationships.add(REL_SUCCESS);
+ relationships.add(REL_FAILURE);
+ }
+ return Collections.unmodifiableSet(relationships);
+ }
+
+ /**
+ * Returns a list of property descriptors supported by this processor. The list always includes properties such as
+ * script engine name, script file name, script body name, script arguments, and an external module path. If the
+ * scripted processor also defines supported properties, those are added to the list as well.
+ *
+ * @return a List of PropertyDescriptor objects supported by this processor
+ */
+ @Override
+ protected List<PropertyDescriptor> getSupportedPropertyDescriptors() {
+
+ synchronized (isInitialized) {
+ if (!isInitialized.get()) {
+ createResources();
+ }
+ }
+ List<PropertyDescriptor> supportedPropertyDescriptors = new ArrayList<>();
+ supportedPropertyDescriptors.addAll(descriptors);
+
+ final Processor instance = processor.get();
+ if (instance != null) {
+ try {
+ final List<PropertyDescriptor> instanceDescriptors = instance.getPropertyDescriptors();
+ if (instanceDescriptors != null) {
+ supportedPropertyDescriptors.addAll(instanceDescriptors);
+ }
+ } catch (final Throwable t) {
+ final ProcessorLog logger = getLogger();
+ final String message = "Unable to get property descriptors from Processor: " + t;
+
+ logger.error(message);
+ if (logger.isDebugEnabled()) {
+ logger.error(message, t);
+ }
+ }
+ }
+
+ return Collections.unmodifiableList(supportedPropertyDescriptors);
+ }
+
+ /**
+ * Returns a PropertyDescriptor for the given name. This is for the user to be able to define their own properties
+ * which will be available as variables in the script
+ *
+ * @param propertyDescriptorName used to lookup if any property descriptors exist for that name
+ * @return a PropertyDescriptor object corresponding to the specified dynamic property name
+ */
+ @Override
+ protected PropertyDescriptor getSupportedDynamicPropertyDescriptor(final String propertyDescriptorName) {
+ return new PropertyDescriptor.Builder()
+ .name(propertyDescriptorName)
+ .required(false)
+ .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
+ .expressionLanguageSupported(true)
+ .dynamic(true)
+ .build();
+ }
+
+ /**
+ * Performs setup operations when the processor is scheduled to run. This includes evaluating the processor's
+ * properties, as well as reloading the script (from file or the "Script Body" property)
+ *
+ * @param context the context in which to perform the setup operations
+ */
+ @OnScheduled
+ @Override
+ public void setup(final ProcessContext context) {
+ super.setup(context);
+
+ if (processor.get() == null) {
+ if (isFile(scriptPath)) {
+ reloadScriptFile(scriptPath);
+ } else {
+ reloadScriptBody(scriptBody);
+ }
+ }
+ }
+
+
+ /**
+ * Handles changes to this processor's properties. If changes are made to script- or engine-related properties,
+ * the script will be reloaded.
+ *
+ * @param descriptor of the modified property
+ * @param oldValue non-null property value (previous)
+ * @param newValue the new property value or if null indicates the property
+ */
+ @Override
+ public void onPropertyModified(final PropertyDescriptor descriptor, final String oldValue, final String newValue) {
+ final ProcessorLog logger = getLogger();
+ final Processor instance = processor.get();
+
+ if (SCRIPT_FILE.equals(descriptor)
+ || SCRIPT_BODY.equals(descriptor)
+ || MODULES.equals(descriptor)
+ || SCRIPT_ENGINE.equals(descriptor)) {
+ lock.lock();
+ try {
+ // if the script is changing we'll want to reload the instance
+ if (SCRIPT_FILE.equals(descriptor)) {
+ if (isFile(newValue)) {
+ reloadScriptFile(newValue);
+
+ // we're attempted to load the script so we need to watch for updates
+ scriptWatcher = new SynchronousFileWatcher(Paths.get(newValue), new LastModifiedMonitor());
+ } else {
+ // the doesn't appear to be a file
+ scriptWatcher = null;
+ }
+
+ // always want to record the configured value
+ scriptPath = newValue;
+ } else if (SCRIPT_BODY.equals(descriptor)) {
+
+ if (reloadScriptBody(newValue)) {
+ // always want to record the configured value
+ scriptBody = newValue;
+ }
+ } else if (MODULES.equals(descriptor)) {
+
+ // temporarily set new value (will be restored to oldValue if something goes wrong)
+ modulePath = newValue;
+ try {
+ setupEngine();
+
+ boolean reloaded = false;
+
+ // we only want to reload during a module change if the script is already loaded
+ if (scriptPath != null || scriptBody != null) {
+ if (isFile(scriptPath)) {
+ // reload the script
+ reloaded = reloadScriptFile(scriptPath);
+ } else if (scriptBody != null) {
+ reloaded = reloadScriptBody(scriptBody);
+ }
+ // log the script was reloaded
+ if (reloaded) {
+ logger.info("The configured script has been successfully reloaded.");
+ } else {
+ throw new ProcessException("The configured script could not be reloaded");
+ }
+ }
+ } catch (Throwable t) {
+ modulePath = oldValue;
+ logger.error(t.getLocalizedMessage(), t);
+ }
+
+ } else if (SCRIPT_ENGINE.equals(descriptor)) {
+ // The script engine has changed, so we need to set up a new instance for the selected
+ // engine name
+ scriptEngineName = newValue;
+ setupEngine();
+ }
+ } finally {
+ lock.unlock();
+ }
+ } else if (instance != null) {
+ // If the script provides a Processor, call its onPropertyModified() method
+ try {
+ instance.onPropertyModified(descriptor, oldValue, newValue);
+ } catch (final Exception e) {
+ final String message = "Unable to invoke onPropertyModified from script Processor: " + e;
+ logger.error(message, e);
+ }
+ }
+ }
+
+ /**
+ * Reloads the script located at the given path
+ *
+ * @param scriptPath the path to the script file to be loaded
+ * @return true if the script was loaded successfully; false otherwise
+ */
+ private boolean reloadScriptFile(final String scriptPath) {
+ final Collection<ValidationResult> results = new HashSet<>();
+
+ try (final FileInputStream scriptStream = new FileInputStream(scriptPath)) {
+ return reloadScript(IOUtils.toString(scriptStream));
+
+ } catch (final Exception e) {
+ final ProcessorLog logger = getLogger();
+ final String message = "Unable to load script: " + e;
+
+ // If the module path has not yet been set, then this script is likely being loaded too early and depends
+ // on modules the processor does not yet know about. If this is the case, it will be reloaded later on
+ // property change (modules) or when scheduled
+ if (modulePath != null) {
+ logger.error(message, e);
+ results.add(new ValidationResult.Builder()
+ .subject("ScriptValidation")
+ .valid(false)
+ .explanation("Unable to load script due to " + e)
+ .input(scriptPath)
+ .build());
+ }
+ }
+
+ // store the updated validation results
+ validationResults.set(results);
+
+ // return whether there was any issues loading the configured script
+ return results.isEmpty();
+ }
+
+ /**
+ * Reloads the script defined by the given string
+ *
+ * @param scriptBody the contents of the script to be loaded
+ * @return true if the script was loaded successfully; false otherwise
+ */
+ private boolean reloadScriptBody(final String scriptBody) {
+ final Collection<ValidationResult> results = new HashSet<>();
+ try {
+ return reloadScript(scriptBody);
+
+ } catch (final Exception e) {
+ final ProcessorLog logger = getLogger();
+ final String message = "Unable to load script: " + e;
+
+ // If the module path has not yet been set, then this script is likely being loaded too early and depends
+ // on modules the processor does not yet know about. If this is the case, it will be reloaded later on
+ // property change (modules) or when scheduled
+ if (modulePath != null) {
--- End diff --
seems like something a bit wonky is happening here. So if one enters a valid script then enters a buggy script the original value is set and not being replaced. Then because if there is an invalid script and no modulepath the error is being buried. Definitely need the old script/processor that was compiled to get wiped out and we need the errors to always make it to the log/bulletin.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---