NIFI-12735 Adding the option to execute flow analysis before committing snapshot to registry #8377
Conversation
| if (analyzeProcessGroupToRegister(snapshot)) { | ||
| return node.registerFlowSnapshot(context, flow, snapshot, externalControllerServices, parameterContexts, parameterProviderReferences, comments, expectedVersion); | ||
| } else { | ||
| throw new ViolatingFlowException("Cannot store flow version to registry due to rules violations"); |
There was a problem hiding this comment.
Currently on the UI we get this error message:
Failed to register flow with Flow Registry due to Cannot store flow version to registry due to rules violations
which is a bit convoluted, something like this sounds more natural:
Failed to register flow with Flow Registry due to There are unresolved rule violations
| throw new ViolatingFlowException("Cannot store flow version to registry due to rules violations"); | |
| throw new InvalidFlowException("There are unresolved rule violations"); |
| package org.apache.nifi.registry.flow; | ||
|
|
||
| public class ViolatingFlowException extends FlowRegistryException { | ||
| public ViolatingFlowException(final String message) { |
There was a problem hiding this comment.
'Violating' is not really an expression we use normally as an adjective so it's use like this feels unnatural.
I propose the name InvalidFlowException that has the added benefit of being more general, in case we later add other methods to validate flows.
| public ViolatingFlowException(final String message) { | |
| public InvalidFlowException(final String message) { |
| final FlowAnalyzer flowAnalyzer, | ||
| final RuleViolationsManager ruleViolationsManager, | ||
| final FlowManager flowManager, | ||
| final Supplier<NiFiRegistryFlowMapper> flowMapperSupplier |
There was a problem hiding this comment.
I don't see a reason for this to be a Supplier. The mapper the flow analysis requires an implementation that doesn't rely on any configuration that can change runtime.
So we can just create it and pass on.
| final InstantiatedVersionedProcessGroup nonVersionedProcessGroup = mapper.mapNonVersionedProcessGroup(flowManager.getGroup(snapshot.getInstanceIdentifier()), serviceProvider); | ||
|
|
||
| flowAnalyzer.analyzeProcessGroup(nonVersionedProcessGroup); | ||
| final Collection<RuleViolation> ruleViolations = ruleViolationsManager.getRuleViolationsForGroup(snapshot.getInstanceIdentifier()); |
There was a problem hiding this comment.
We probably want only violations with ENFORCE policy to block committing to Registry.
| final Collection<RuleViolation> ruleViolations = ruleViolationsManager.getRuleViolationsForGroup(snapshot.getInstanceIdentifier()); | |
| final List<RuleViolation> enforcingRuleViolations = ruleViolationsManager.getRuleViolationsForGroup(snapshot.getInstanceIdentifier()).stream() | |
| .filter(ruleViolation -> EnforcementPolicy.ENFORCE.equals(ruleViolation.getEnforcementPolicy())) | |
| .collect(Collectors.toList()); |
There was a problem hiding this comment.
Good catch. What do you think, from rules and violation perspective, would it make sense to allow users to configure the level or violations which can prevents commit?
There was a problem hiding this comment.
I don't have a strong opinion on this. We could change the nifi.property entry from a boolean to accommodate for this but even that might be a much.
| } | ||
|
|
||
| @Test | ||
| public void testWhenNoViolationFound() throws IOException, FlowRegistryException { |
There was a problem hiding this comment.
Test reports can help a lot understanding the various use cases if the method names describe the cases properly. The 'test' prefix on the other hand has no meaningful purpose and although doesn't hurt much, if the method name would become too long, it's better to omit it.
| public void testWhenNoViolationFound() throws IOException, FlowRegistryException { | |
| public void allowFlowRegistrationWhenNoEnforcingViolationFound() throws IOException, FlowRegistryException { |
| .flowAnalysisAtRegistryCommit(nifiProperties.shouldCheckForRulesViolationWhenRegistryCommit()); | ||
|
|
||
| if (getFlowAnalyzer().isPresent()) { | ||
| extensionBuilder.flowAnalyzer(getFlowAnalyzer().get()); | ||
| } | ||
|
|
||
| if (getRuleViolationsManager().isPresent()) { | ||
| extensionBuilder.ruleViolationsManager(getRuleViolationsManager().get()); | ||
| } | ||
|
|
||
| final FlowRegistryClientNode clientNode = | ||
| extensionBuilder.buildFlowRegistryClient(); |
There was a problem hiding this comment.
| .flowAnalysisAtRegistryCommit(nifiProperties.shouldCheckForRulesViolationWhenRegistryCommit()); | |
| if (getFlowAnalyzer().isPresent()) { | |
| extensionBuilder.flowAnalyzer(getFlowAnalyzer().get()); | |
| } | |
| if (getRuleViolationsManager().isPresent()) { | |
| extensionBuilder.ruleViolationsManager(getRuleViolationsManager().get()); | |
| } | |
| final FlowRegistryClientNode clientNode = | |
| extensionBuilder.buildFlowRegistryClient(); | |
| .flowAnalyzer(getFlowAnalyzer().orElse(null)) | |
| .ruleViolationsManager(getRuleViolationsManager().orElse(null)) |
| @@ -300,6 +300,9 @@ public class NiFiProperties extends ApplicationProperties { | |||
| // flow analysis properties | |||
| public static final String BACKGROUND_FLOW_ANALYSIS_SCHEDULE = "nifi.flow.analysis.background.task.schedule"; | |||
|
|
|||
| // registry client properties | |||
| public static final String CHECK_FOL_RULES_VIOLATION_WHEN_REGISTRY_COMMIT = "nifi.registry.check.for.rules.violation"; | |||
There was a problem hiding this comment.
This doesn't seem to grasp the concept properly. It indicates only the fact that violations can be important but doesn't tell why or what the consequences are.
| public static final String CHECK_FOL_RULES_VIOLATION_WHEN_REGISTRY_COMMIT = "nifi.registry.check.for.rules.violation"; | |
| public static final String NIFI_REGISTRY_CHECK_FOR_RULE_VIOLATIONS_BEFORE_COMMIT = "nifi.registry.check.for.rule.violations.before.commit"; |
| @@ -401,6 +404,7 @@ public class NiFiProperties extends ApplicationProperties { | |||
| private static final String DEFAULT_SECURITY_USER_JWS_KEY_ROTATION_PERIOD = "PT1H"; | |||
| public static final String DEFAULT_WEB_SHOULD_SEND_SERVER_VERSION = "true"; | |||
| public static final int DEFAULT_LISTENER_BOOTSTRAP_PORT = 0; | |||
| public static final Boolean DEFAULT_CHECK_FOL_RULES_VIOLATION_WHEN_REGISTRY_COMMIT = false; | |||
There was a problem hiding this comment.
| public static final Boolean DEFAULT_CHECK_FOL_RULES_VIOLATION_WHEN_REGISTRY_COMMIT = false; | |
| public static final Boolean DEFAULT_NIFI_REGISTRY_CHECK_FOR_RULE_VIOLATIONS_BEFORE_COMMIT = false; |
| @@ -1826,6 +1830,17 @@ public Set<String> getDirectSubsequentTokens(final String prefix) { | |||
| .collect(Collectors.toSet()); | |||
| } | |||
|
|
|||
| /** | |||
| * @return Returns true if NiFi should execute flow analysis check on the process group before committing it to a registry. Returns | |||
There was a problem hiding this comment.
| * @return Returns true if NiFi should execute flow analysis check on the process group before committing it to a registry. Returns | |
| * @return Returns true if NiFi should execute flow analysis on the process group before committing it to a registry. Returns |
| public boolean shouldCheckForRulesViolationWhenRegistryCommit() { | ||
| final String shouldCheckForRulesViolationWhenRegistryCommit = getProperty( | ||
| CHECK_FOL_RULES_VIOLATION_WHEN_REGISTRY_COMMIT, | ||
| DEFAULT_CHECK_FOL_RULES_VIOLATION_WHEN_REGISTRY_COMMIT.toString()); | ||
| return Boolean.parseBoolean(shouldCheckForRulesViolationWhenRegistryCommit); | ||
| } |
There was a problem hiding this comment.
| public boolean shouldCheckForRulesViolationWhenRegistryCommit() { | |
| final String shouldCheckForRulesViolationWhenRegistryCommit = getProperty( | |
| CHECK_FOL_RULES_VIOLATION_WHEN_REGISTRY_COMMIT, | |
| DEFAULT_CHECK_FOL_RULES_VIOLATION_WHEN_REGISTRY_COMMIT.toString()); | |
| return Boolean.parseBoolean(shouldCheckForRulesViolationWhenRegistryCommit); | |
| } | |
| public boolean nifiRegistryCheckForRuleViolationsBeforeCommit() { | |
| final String nifiRegistryCheckForRuleViolationsBeforeCommit = getProperty( | |
| NIFI_REGISTRY_CHECK_FOR_RULE_VIOLATIONS_BEFORE_COMMIT, | |
| DEFAULT_NIFI_REGISTRY_CHECK_FOR_RULE_VIOLATIONS_BEFORE_COMMIT.toString()); | |
| if (!"true".equalsIgnoreCase(nifiRegistryCheckForRuleViolationsBeforeCommit) && !"false".equalsIgnoreCase(nifiRegistryCheckForRuleViolationsBeforeCommit)) { | |
| throw new RuntimeException(String.format("%s was '%s', expected true or false", NiFiProperties.ZOOKEEPER_CLIENT_ENSEMBLE_TRACKER, nifiRegistryCheckForRuleViolationsBeforeCommit)); | |
| } | |
| return Boolean.parseBoolean(nifiRegistryCheckForRuleViolationsBeforeCommit); | |
| } |
| @@ -303,6 +303,9 @@ public class NiFiProperties extends ApplicationProperties { | |||
| // flow analysis properties | |||
| public static final String BACKGROUND_FLOW_ANALYSIS_SCHEDULE = "nifi.flow.analysis.background.task.schedule"; | |||
|
|
|||
| // registry client properties | |||
| public static final String FLOW_REGISTRY_CHECK_FOR_RULE_VIOLATIONS_BEFORE_COMMIT = "nifi.registry.check.for.rule.violation.before.commit"; | |||
There was a problem hiding this comment.
We should add this property in nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/src/main/resources/conf/nifi.properties and in nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-resources/pom.xml as well.
| public static final String FLOW_REGISTRY_CHECK_FOR_RULE_VIOLATIONS_BEFORE_COMMIT = "nifi.registry.check.for.rule.violation.before.commit"; | |
| public static final String FLOW_REGISTRY_CHECK_FOR_RULE_VIOLATIONS_BEFORE_COMMIT = "nifi.registry.check.for.rule.violations.before.commit"; |
…ng snapshot to registry
…ents based on code review comments
|
LGTM |
Summary
NIFI-12735
Tracking
Please complete the following tracking steps prior to pull request creation.
Issue Tracking
Pull Request Tracking
NIFI-00000NIFI-00000Pull Request Formatting
mainbranchVerification
Please indicate the verification steps performed prior to pull request creation.
Build
mvn clean install -P contrib-checkLicensing
LICENSEandNOTICEfilesDocumentation