You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by GitBox <gi...@apache.org> on 2022/06/10 20:36:23 UTC
[GitHub] [nifi] markap14 commented on a diff in pull request #6115: NIFI-10108 Processor scheduling via parameter
markap14 commented on code in PR #6115:
URL: https://github.com/apache/nifi/pull/6115#discussion_r894840416
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java:
##########
@@ -1197,6 +1208,34 @@ protected Collection<ValidationResult> computeValidationErrors(final ValidationC
return results;
}
+ @Override
+ protected List<ValidationResult> validateConfig() {
+
+ final List<ValidationResult> results = new ArrayList<>();
+ final ParameterContext parameterContext = getParameterContext();
+
+ if (parameterContext == null && !this.parameterReferences.isEmpty()) {
+ results.add(new ValidationResult.Builder()
+ .subject(getName())
+ .valid(false)
+ .explanation("Processor configuration references one or more Parameters but no Parameter Context is currently set on the Process Group.")
+ .build());
+ } else {
+ for (final ParameterReference paramRef : parameterReferences) {
+ if (!parameterContext.getParameter(paramRef.getParameterName()).isPresent() ) {
Review Comment:
It should also be invalid if it references a sensitive parameter.
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java:
##########
@@ -1197,6 +1208,34 @@ protected Collection<ValidationResult> computeValidationErrors(final ValidationC
return results;
}
+ @Override
+ protected List<ValidationResult> validateConfig() {
+
+ final List<ValidationResult> results = new ArrayList<>();
+ final ParameterContext parameterContext = getParameterContext();
+
+ if (parameterContext == null && !this.parameterReferences.isEmpty()) {
+ results.add(new ValidationResult.Builder()
+ .subject(getName())
Review Comment:
Subject here should be what is validated. So "Run Schedule" or similar.
Should also include `.input(runSchedule)` or something analogous to convey what the value is (but not expose the evaluated value)
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/service/StandardControllerServiceNode.java:
##########
@@ -504,6 +505,11 @@ public ValidationState performValidation(final ValidationContext validationConte
return state;
}
+ @Override
+ protected List<ValidationResult> validateConfig() {
+ return Collections.EMPTY_LIST;
Review Comment:
`Collections.EMPTY_LIST` is discouraged. We should instead use `Collections.emptyList()` as it adheres properly to generics
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java:
##########
@@ -505,24 +510,30 @@ public String getSchedulingPeriod() {
@Override
@SuppressWarnings("deprecation")
- public synchronized void setScheduldingPeriod(final String schedulingPeriod) {
+ public synchronized void setSchedulingPeriod(final String schedulingPeriod) {
if (isRunning()) {
throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
}
+ final String evaluatedSchedulingPeriod = evaluateSchedulingPeriodFromParameter(schedulingPeriod);
+ final ParameterParser parameterParser = new ExpressionLanguageAgnosticParameterParser();
+ final ParameterTokenList parameterTokenList = parameterParser.parseTokens(schedulingPeriod);
+
+ parameterReferences = new ArrayList<>(parameterTokenList.toReferenceList());
+
switch (schedulingStrategy) {
case CRON_DRIVEN: {
try {
- new CronExpression(schedulingPeriod);
+ new CronExpression(evaluatedSchedulingPeriod);
} catch (final Exception e) {
throw new IllegalArgumentException(
- "Scheduling Period is not a valid cron expression: " + schedulingPeriod);
+ "Scheduling Period is not a valid cron expression: " + evaluatedSchedulingPeriod);
}
}
break;
case PRIMARY_NODE_ONLY:
case TIMER_DRIVEN: {
- final long schedulingNanos = FormatUtils.getTimeDuration(requireNonNull(schedulingPeriod),
+ final long schedulingNanos = FormatUtils.getTimeDuration(requireNonNull(evaluatedSchedulingPeriod),
TimeUnit.NANOSECONDS);
if (schedulingNanos < 0) {
throw new IllegalArgumentException("Scheduling Period must be positive");
Review Comment:
Because these values can change outside of this method now, by changing a parameter, we need to perform these checks during validation. Because we'll be checking it there, and because we shouldn't throw an Exception if referencing a parameter that hasn't yet been set, the validation should now be removed from this method.
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java:
##########
@@ -1928,6 +1967,21 @@ public void setMaxBackoffPeriod(String maxBackoffPeriod) {
}
this.maxBackoffPeriod = maxBackoffPeriod;
}
+
+ @Override
+ public String evaluateSchedulingPeriodFromParameter(final String schedulingPeriod) {
+ final ParameterContext parameterContext = getParameterContext();
+ final ParameterParser parameterParser = new ExpressionLanguageAgnosticParameterParser();
+ final ParameterTokenList parameterTokenList = parameterParser.parseTokens(schedulingPeriod);
+ final String evaluatedValue = parameterTokenList.substitute(parameterContext);
+
+ if (evaluatedValue != null && !evaluatedValue.startsWith("#{")) {
+ return evaluatedValue;
+ } else {
+ return schedulingStrategy.getDefaultSchedulingPeriod();
+ }
Review Comment:
Not sure that I follow the intent here. Is the value is `null` or starts with `#{` then use the default scheduling period. Otherwise use `evaluatedValue`? Why would we ever use DefaultSchedulingPeriod? What if the value is set to `1#{param}`? In that case, we'd still use `evaluatedValue`. As we would if the parameter had a value of `hello`. So not sure that I'm understanding the distinction being made here.
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/test/java/org/apache/nifi/controller/TestAbstractComponentNode.java:
##########
@@ -403,6 +403,11 @@ public Resource getResource() {
return null;
}
+ @Override
+ protected List<ValidationResult> validateConfig() {
+ return Collections.EMPTY_LIST;
Review Comment:
Should use `Collections.emptyList()`
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/reporting/StandardReportingTaskNode.java:
##########
@@ -87,6 +91,11 @@ public ReportingContext getReportingContext() {
return new StandardReportingContext(flowController, flowController.getBulletinRepository(), getEffectivePropertyValues(), this, getVariableRegistry(), ParameterLookup.EMPTY);
}
+ @Override
+ protected List<ValidationResult> validateConfig() {
+ return Collections.EMPTY_LIST;
Review Comment:
`Collections.emptyList()`
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core-api/src/main/java/org/apache/nifi/controller/AbstractComponentNode.java:
##########
@@ -793,6 +795,8 @@ protected Collection<ValidationResult> computeValidationErrors(final ValidationC
.build());
}
+ protected abstract List<ValidationResult> validateConfig();
Review Comment:
This is an abstract method that we expect other classes to implement. Please add JavaDocs that explain what is expected of this implementations.
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/controller/scheduling/TimerDrivenSchedulingAgent.java:
##########
@@ -71,6 +71,9 @@ public void doSchedule(final Connectable connectable, final LifecycleState sched
final List<ScheduledFuture<?>> futures = new ArrayList<>();
final ConnectableTask connectableTask = new ConnectableTask(this, connectable, flowController, contextFactory, scheduleState, encryptor);
+ final long schedulingPeriodNanos = FormatUtils.getTimeDuration(connectable.evaluateSchedulingPeriodFromParameter(connectable.getSchedulingPeriod()),
+ TimeUnit.NANOSECONDS);
Review Comment:
This can be much simpler, as:
```
final long schedulingPeriodNanos = connectable.getSchedulingPeriod(TimeUnit.NANOSECONDS);
```
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java:
##########
@@ -1197,6 +1208,34 @@ protected Collection<ValidationResult> computeValidationErrors(final ValidationC
return results;
}
+ @Override
+ protected List<ValidationResult> validateConfig() {
+
+ final List<ValidationResult> results = new ArrayList<>();
+ final ParameterContext parameterContext = getParameterContext();
+
+ if (parameterContext == null && !this.parameterReferences.isEmpty()) {
+ results.add(new ValidationResult.Builder()
+ .subject(getName())
+ .valid(false)
+ .explanation("Processor configuration references one or more Parameters but no Parameter Context is currently set on the Process Group.")
+ .build());
+ } else {
+ for (final ParameterReference paramRef : parameterReferences) {
+ if (!parameterContext.getParameter(paramRef.getParameterName()).isPresent() ) {
+ results.add(new ValidationResult.Builder()
+ .subject(getName())
+ .valid(false)
+ .explanation("Processor configuration references Parameter '" + paramRef.getParameterName() +
+ "' but the currently selected Parameter Context does not have a Parameter with that name")
+ .build());
+ }
+ }
+ }
+
+ return results;
Review Comment:
I don't think we're done evaluating here. At this point, after substituting the values in, we need to validate that the resultant value is correct. For example, if I set a value of `#{greeting}` which evaluates to `hello` that shouldn't be valid here. I think the logic that is above in the `setSchedulingPeriod` method now needs to be moved down here.
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java:
##########
@@ -505,24 +510,30 @@ public String getSchedulingPeriod() {
@Override
@SuppressWarnings("deprecation")
- public synchronized void setScheduldingPeriod(final String schedulingPeriod) {
+ public synchronized void setSchedulingPeriod(final String schedulingPeriod) {
if (isRunning()) {
throw new IllegalStateException("Cannot modify Processor configuration while the Processor is running");
}
+ final String evaluatedSchedulingPeriod = evaluateSchedulingPeriodFromParameter(schedulingPeriod);
+ final ParameterParser parameterParser = new ExpressionLanguageAgnosticParameterParser();
+ final ParameterTokenList parameterTokenList = parameterParser.parseTokens(schedulingPeriod);
+
+ parameterReferences = new ArrayList<>(parameterTokenList.toReferenceList());
+
switch (schedulingStrategy) {
case CRON_DRIVEN: {
try {
- new CronExpression(schedulingPeriod);
+ new CronExpression(evaluatedSchedulingPeriod);
} catch (final Exception e) {
throw new IllegalArgumentException(
- "Scheduling Period is not a valid cron expression: " + schedulingPeriod);
+ "Scheduling Period is not a valid cron expression: " + evaluatedSchedulingPeriod);
Review Comment:
This could potentially be set to a sensitive parameter value. We cannot include the output here, as it means that a user could maliciously set the scheduling period to something like `#{password}` just to get back an error message that leaks the sensitive value. Should instead just log the configured value: `schedulingPeriod`
##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/StandardProcessorNode.java:
##########
@@ -1928,6 +1967,21 @@ public void setMaxBackoffPeriod(String maxBackoffPeriod) {
}
this.maxBackoffPeriod = maxBackoffPeriod;
}
+
+ @Override
+ public String evaluateSchedulingPeriodFromParameter(final String schedulingPeriod) {
+ final ParameterContext parameterContext = getParameterContext();
+ final ParameterParser parameterParser = new ExpressionLanguageAgnosticParameterParser();
+ final ParameterTokenList parameterTokenList = parameterParser.parseTokens(schedulingPeriod);
+ final String evaluatedValue = parameterTokenList.substitute(parameterContext);
+
+ if (evaluatedValue != null && !evaluatedValue.startsWith("#{")) {
+ return evaluatedValue;
+ } else {
+ return schedulingStrategy.getDefaultSchedulingPeriod();
+ }
Review Comment:
After more fully reviewing this PR, I would say this method doesn't really have anything to do with scheduling period. It has to do with evaluating parameters. So perhaps we should just name it `evaluateParameters` or `substituteParameters`.
##########
nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/controller/reporting/StatelessReportingTaskNode.java:
##########
@@ -46,6 +50,11 @@ public StatelessReportingTaskNode(final LoggableComponent<ReportingTask> reporti
this.statelessEngine = statelessEngine;
}
+ @Override
+ protected List<ValidationResult> validateConfig() {
+ return Collections.EMPTY_LIST;
Review Comment:
`Collections.emptyList()`
--
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