Skip to content

NIFI-16351 - Prevent Flow Registry synchronization failures during startup validation #11684

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
pvillard31 wants to merge 1 commit into
base: main
Choose a base branch
from
+519 −9
Open

NIFI-16351 - Prevent Flow Registry synchronization failures during startup validation #11684

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 42 additions & 7 deletions .../nifi-framework-components/src/main/java/org/apache/nifi/groups/StandardProcessGroup.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
import org.apache.nifi.authorization.resource.ResourceFactory;
import org.apache.nifi.authorization.resource.ResourceType;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.components.connector.ConnectorNode;
import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.components.state.StateManagerProvider;
Expand Down Expand Up @@ -3880,25 +3881,35 @@ public void synchronizeWithFlowRegistry(final FlowManager flowManager) {
return;
}

final ValidationStatus validationStatus = flowRegistry.getValidationStatus(10, TimeUnit.SECONDS);

if (validationStatus == ValidationStatus.VALIDATING) {
return;
}

if (validationStatus == ValidationStatus.INVALID) {
final String message = buildValidationFailureExplanation(flowRegistry);
versionControlFields.setSyncFailureExplanation(message);

LOG.error("{} for {}", message, this);
return;
}

final VersionedProcessGroup snapshot = vci.getFlowSnapshot();
if (snapshot == null && vci.getVersion() != null) {
// We have not yet obtained the snapshot from the Flow Registry, so we need to request the snapshot of our local version of the flow from the Flow Registry.
// This allows us to know whether or not the flow has been modified since it was last synced with the Flow Registry.
try {
final ValidationStatus validationStatus = flowRegistry.getValidationStatus(10, TimeUnit.SECONDS);
if (validationStatus == ValidationStatus.VALIDATING) {
throw new FlowRegistryException(flowRegistry + " cannot currently be used to synchronize with Flow Registry because it is currently validating");
}

final FlowVersionLocation flowVersionLocation = new FlowVersionLocation(vci.getBranch(), vci.getBucketIdentifier(), vci.getFlowIdentifier(), vci.getVersion());
final FlowSnapshotContainer registrySnapshotContainer = flowRegistry.getFlowContents(
FlowRegistryClientContextFactory.getAnonymousContext(), flowVersionLocation, false);
final RegisteredFlowSnapshot registrySnapshot = registrySnapshotContainer.getFlowSnapshot();
final VersionedProcessGroup registryFlow = registrySnapshot.getFlowContents();
vci.setFlowSnapshot(registryFlow);
} catch (final IOException | FlowRegistryException e) {
final String message = String.format("Failed to synchronize Process Group with Flow Registry because could not retrieve version %s of flow with identifier %s in bucket %s",
vci.getVersion(), vci.getFlowIdentifier(), vci.getBucketIdentifier());
final String message = appendExceptionMessage(String.format(
"Failed to synchronize Process Group with Flow Registry because could not retrieve version %s of flow with identifier %s in bucket %s",
vci.getVersion(), vci.getFlowIdentifier(), vci.getBucketIdentifier()), e);
versionControlFields.setSyncFailureExplanation(message);

final String logErrorMessage = "Failed to synchronize {} with Flow Registry because could not retrieve version {} of flow with identifier {} in bucket {}";
Expand Down Expand Up @@ -3945,6 +3956,30 @@ public void synchronizeWithFlowRegistry(final FlowManager flowManager) {
}
}

private String buildValidationFailureExplanation(final FlowRegistryClientNode flowRegistry) {
final Collection<ValidationResult> validationResults = flowRegistry.getValidationErrors();
final String validationErrors = validationResults == null ? null : validationResults.stream()
.map(validationResult -> StringUtils.isNotBlank(validationResult.getExplanation()) ? validationResult.getExplanation() : validationResult.toString())
.filter(StringUtils::isNotBlank)
.collect(Collectors.joining("; "));

final String message = "Failed to synchronize Process Group with Flow Registry because the Flow Registry failed validation";
if (StringUtils.isBlank(validationErrors)) {
return message;
}

return message + ": " + validationErrors;
}

private String appendExceptionMessage(final String message, final Exception exception) {
final String exceptionMessage = exception.getMessage();
if (StringUtils.isBlank(exceptionMessage)) {
return message;
}

return message + ": " + exceptionMessage;
}

@Override
public ComponentAdditions addVersionedComponents(final VersionedComponentAdditions additions, final String componentIdSeed) {
final ComponentIdGenerator idGenerator = (proposedId, instanceId, destinationGroupId) -> generateUuid(proposedId, destinationGroupId, componentIdSeed);
Expand Down
238 changes: 238 additions & 0 deletions ...i-framework-components/src/test/java/org/apache/nifi/groups/StandardProcessGroupTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,10 +17,12 @@
package org.apache.nifi.groups;

import org.apache.nifi.asset.AssetManager;
import org.apache.nifi.components.ValidationResult;
import org.apache.nifi.components.state.Scope;
import org.apache.nifi.components.state.StateManager;
import org.apache.nifi.components.state.StateManagerProvider;
import org.apache.nifi.components.state.StateMap;
import org.apache.nifi.components.validation.ValidationStatus;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.ConnectableType;
import org.apache.nifi.connectable.Connection;
Expand All @@ -36,8 +38,16 @@
import org.apache.nifi.controller.queue.QueueSize;
import org.apache.nifi.controller.service.ControllerServiceProvider;
import org.apache.nifi.flow.ExecutionEngine;
import org.apache.nifi.flow.VersionedProcessGroup;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.registry.flow.FlowRegistryClientNode;
import org.apache.nifi.registry.flow.FlowRegistryException;
import org.apache.nifi.registry.flow.FlowSnapshotContainer;
import org.apache.nifi.registry.flow.RegisteredFlow;
import org.apache.nifi.registry.flow.RegisteredFlowSnapshot;
import org.apache.nifi.registry.flow.StandardVersionControlInformation;
import org.apache.nifi.registry.flow.VersionControlInformation;
import org.apache.nifi.registry.flow.VersionedFlowState;
import org.apache.nifi.registry.flow.VersionedFlowStatus;
import org.apache.nifi.security.encryption.PropertyEncryptionProvider;
import org.apache.nifi.util.NiFiProperties;
Expand All @@ -48,16 +58,21 @@
import org.mockito.junit.jupiter.MockitoExtension;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
Expand All @@ -83,6 +98,9 @@ class StandardProcessGroupTest {
private static final String REGISTERED_FLOW_IDENTIFIER = "87654321-4321";
private static final String REGISTERED_FLOW_VERSION = "1.0.0";
private static final String REGISTERED_FLOW_IDENTIFIER_PATH = "/87654321-4321:1.0.0";
private static final String REGISTRY_ID = "registry-id";
private static final String BUCKET_ID = "bucket-id";
private static final String FLOW_ID = "flow-id";

@Mock
private ControllerServiceProvider controllerServiceProvider;
Expand Down Expand Up @@ -476,10 +494,230 @@ void testDropAllFlowFilesCompletionWaitsForEveryConnection() {
assertTrue(completionFuture.isDone());
}

@Test
void testSynchronizeWithFlowRegistryDoesNotLatchRetrievalFailureWhileRegistryIsValidating() throws Exception {
final FlowRegistryClientNode flowRegistry = mock(FlowRegistryClientNode.class);
when(flowManager.getFlowRegistryClient(REGISTRY_ID)).thenReturn(flowRegistry);
when(flowRegistry.getValidationStatus(anyLong(), eq(TimeUnit.SECONDS))).thenReturn(ValidationStatus.VALIDATING);

processGroup.setVersionControlInformation(createVersionControlInformation(null, VersionedFlowState.UP_TO_DATE, null), Map.of());

assertDoesNotThrow(() -> processGroup.synchronizeWithFlowRegistry(flowManager));

assertNull(getSyncFailureExplanation(processGroup));
verify(flowRegistry, never()).getFlowContents(any(), any(), eq(false));
verify(flowRegistry, never()).getFlow(any(), any());
verify(flowRegistry, never()).getLatestVersion(any(), any());
}

@Test
void testSynchronizeWithFlowRegistryPreservesPreviousFailureWhileRegistryIsValidating() throws Exception {
final FlowRegistryClientNode flowRegistry = mock(FlowRegistryClientNode.class);
final String previousFailure = "Previous registry failure";

when(flowManager.getFlowRegistryClient(REGISTRY_ID)).thenReturn(flowRegistry);
when(flowRegistry.getValidationStatus(anyLong(), eq(TimeUnit.SECONDS))).thenReturn(ValidationStatus.VALIDATING);

processGroup.setVersionControlInformation(
createVersionControlInformation(null, VersionedFlowState.SYNC_FAILURE, previousFailure),
Map.of()
);

processGroup.synchronizeWithFlowRegistry(flowManager);

assertEquals(previousFailure, getSyncFailureExplanation(processGroup));
verify(flowRegistry, never()).getFlowContents(any(), any(), eq(false));
verify(flowRegistry, never()).getFlow(any(), any());
verify(flowRegistry, never()).getLatestVersion(any(), any());
}

@Test
void testSynchronizeWithFlowRegistryDoesNotReadRegistryMetadataWhileRegistryIsValidatingWithExistingSnapshot() throws Exception {
final FlowRegistryClientNode flowRegistry = mock(FlowRegistryClientNode.class);

when(flowManager.getFlowRegistryClient(REGISTRY_ID)).thenReturn(flowRegistry);
when(flowRegistry.getValidationStatus(anyLong(), eq(TimeUnit.SECONDS))).thenReturn(ValidationStatus.VALIDATING);

processGroup.setVersionControlInformation(
createVersionControlInformation(mock(VersionedProcessGroup.class), VersionedFlowState.UP_TO_DATE, null),
Map.of()
);

assertDoesNotThrow(() -> processGroup.synchronizeWithFlowRegistry(flowManager));

assertNull(getSyncFailureExplanation(processGroup));
verify(flowRegistry, never()).getFlow(any(), any());
verify(flowRegistry, never()).getLatestVersion(any(), any());
}

@Test
void testSynchronizeWithFlowRegistryReadsRegistryMetadataWhenRegistryIsValidWithExistingSnapshot() throws Exception {
final FlowRegistryClientNode flowRegistry = mock(FlowRegistryClientNode.class);
final RegisteredFlow versionedFlow = mock(RegisteredFlow.class);

when(flowManager.getFlowRegistryClient(REGISTRY_ID)).thenReturn(flowRegistry);
when(flowRegistry.getValidationStatus(anyLong(), eq(TimeUnit.SECONDS))).thenReturn(ValidationStatus.VALID);
when(flowRegistry.getFlow(any(), any())).thenReturn(versionedFlow);
when(flowRegistry.getLatestVersion(any(), any())).thenReturn(Optional.of(REGISTERED_FLOW_VERSION));
when(versionedFlow.getBucketName()).thenReturn("Bucket");
when(versionedFlow.getName()).thenReturn("Flow");
when(versionedFlow.getDescription()).thenReturn("Description");
when(flowRegistry.getName()).thenReturn("Registry");

processGroup.setVersionControlInformation(
createVersionControlInformation(mock(VersionedProcessGroup.class), VersionedFlowState.UP_TO_DATE, null),
Map.of()
);

processGroup.synchronizeWithFlowRegistry(flowManager);

verify(flowRegistry).getFlow(any(), any());
verify(flowRegistry).getLatestVersion(any(), any());
assertNull(getSyncFailureExplanation(processGroup));
}

@Test
void testSynchronizeWithFlowRegistryMarksValidationFailureWithoutRegistryDataCallsWhenRegistryIsInvalid() throws Exception {
final FlowRegistryClientNode flowRegistry = mock(FlowRegistryClientNode.class);

when(flowManager.getFlowRegistryClient(REGISTRY_ID)).thenReturn(flowRegistry);
when(flowRegistry.getValidationStatus(anyLong(), eq(TimeUnit.SECONDS))).thenReturn(ValidationStatus.INVALID);
when(flowRegistry.getValidationErrors()).thenReturn(List.of(new ValidationResult.Builder()
.subject("Registry URL")
.input("")
.valid(false)
.explanation("Registry URL is required")
.build()));

processGroup.setVersionControlInformation(createVersionControlInformation(null, VersionedFlowState.UP_TO_DATE, null), Map.of());

processGroup.synchronizeWithFlowRegistry(flowManager);

final String syncFailureExplanation = getSyncFailureExplanation(processGroup);
assertNotNull(syncFailureExplanation);
assertTrue(syncFailureExplanation.toLowerCase().contains("validation"));
assertTrue(syncFailureExplanation.contains("Registry URL is required"));
verify(flowRegistry, never()).getFlowContents(any(), any(), eq(false));
verify(flowRegistry, never()).getFlow(any(), any());
verify(flowRegistry, never()).getLatestVersion(any(), any());
}

@Test
void testSynchronizeWithFlowRegistryIncludesUnderlyingRegistryErrorWhenSnapshotFetchFails() throws Exception {
final FlowRegistryClientNode flowRegistry = mock(FlowRegistryClientNode.class);

when(flowManager.getFlowRegistryClient(REGISTRY_ID)).thenReturn(flowRegistry);
when(flowRegistry.getValidationStatus(anyLong(), eq(TimeUnit.SECONDS))).thenReturn(ValidationStatus.VALID);
when(flowRegistry.getFlowContents(any(), any(), eq(false))).thenThrow(new FlowRegistryException("Registry offline"));

processGroup.setVersionControlInformation(createVersionControlInformation(null, VersionedFlowState.UP_TO_DATE, null), Map.of());

processGroup.synchronizeWithFlowRegistry(flowManager);

final String syncFailureExplanation = getSyncFailureExplanation(processGroup);
assertNotNull(syncFailureExplanation);
assertTrue(syncFailureExplanation.contains("could not retrieve version"));
assertTrue(syncFailureExplanation.contains("Registry offline"));
}

@Test
void testSynchronizeWithFlowRegistryClearsPreviousFailureAfterSuccessfulRetry() throws Exception {
final FlowRegistryClientNode flowRegistry = mock(FlowRegistryClientNode.class);
final FlowSnapshotContainer snapshotContainer = mock(FlowSnapshotContainer.class);
final RegisteredFlowSnapshot registeredFlowSnapshot = mock(RegisteredFlowSnapshot.class);
final RegisteredFlow versionedFlow = mock(RegisteredFlow.class);

when(flowManager.getFlowRegistryClient(REGISTRY_ID)).thenReturn(flowRegistry);
when(flowRegistry.getValidationStatus(anyLong(), eq(TimeUnit.SECONDS))).thenReturn(ValidationStatus.VALID);
when(flowRegistry.getFlowContents(any(), any(), eq(false)))
.thenThrow(new FlowRegistryException("Registry offline"))
.thenReturn(snapshotContainer);
when(snapshotContainer.getFlowSnapshot()).thenReturn(registeredFlowSnapshot);
when(registeredFlowSnapshot.getFlowContents()).thenReturn(mock(VersionedProcessGroup.class));
when(flowRegistry.getFlow(any(), any())).thenReturn(versionedFlow);
when(flowRegistry.getLatestVersion(any(), any())).thenReturn(Optional.of(REGISTERED_FLOW_VERSION));
when(versionedFlow.getBucketName()).thenReturn("Bucket");
when(versionedFlow.getName()).thenReturn("Flow");
when(versionedFlow.getDescription()).thenReturn("Description");
when(flowRegistry.getName()).thenReturn("Registry");

processGroup.setVersionControlInformation(createVersionControlInformation(null, VersionedFlowState.UP_TO_DATE, null), Map.of());

processGroup.synchronizeWithFlowRegistry(flowManager);
assertNotNull(getSyncFailureExplanation(processGroup));

processGroup.synchronizeWithFlowRegistry(flowManager);

assertNull(getSyncFailureExplanation(processGroup));
}

@Test
void testSynchronizeWithFlowRegistryClearsValidationFailureAfterSuccessfulRetry() throws Exception {
final FlowRegistryClientNode flowRegistry = mock(FlowRegistryClientNode.class);
final RegisteredFlow versionedFlow = mock(RegisteredFlow.class);

when(flowManager.getFlowRegistryClient(REGISTRY_ID)).thenReturn(flowRegistry);
when(flowRegistry.getValidationStatus(anyLong(), eq(TimeUnit.SECONDS)))
.thenReturn(ValidationStatus.INVALID)
.thenReturn(ValidationStatus.VALID);
when(flowRegistry.getValidationErrors()).thenReturn(List.of(new ValidationResult.Builder()
.subject("Registry URL")
.input("")
.valid(false)
.explanation("Registry URL is required")
.build()));
when(flowRegistry.getFlow(any(), any())).thenReturn(versionedFlow);
when(flowRegistry.getLatestVersion(any(), any())).thenReturn(Optional.of(REGISTERED_FLOW_VERSION));
when(versionedFlow.getBucketName()).thenReturn("Bucket");
when(versionedFlow.getName()).thenReturn("Flow");
when(versionedFlow.getDescription()).thenReturn("Description");
when(flowRegistry.getName()).thenReturn("Registry");

processGroup.setVersionControlInformation(
createVersionControlInformation(mock(VersionedProcessGroup.class), VersionedFlowState.UP_TO_DATE, null),
Map.of()
);

processGroup.synchronizeWithFlowRegistry(flowManager);
assertNotNull(getSyncFailureExplanation(processGroup));

processGroup.synchronizeWithFlowRegistry(flowManager);

assertNull(getSyncFailureExplanation(processGroup));
}

private StandardProcessGroup createStandardProcessGroup(final String id) {
return createStandardProcessGroup(id, null);
}

private StandardVersionControlInformation createVersionControlInformation(
final VersionedProcessGroup flowSnapshot,
final VersionedFlowState state,
final String explanation) {
return new StandardVersionControlInformation(
REGISTRY_ID,
"Registry",
"main",
BUCKET_ID,
FLOW_ID,
REGISTERED_FLOW_VERSION,
null,
flowSnapshot,
new StandardVersionedFlowStatus(state, explanation)
);
}

private String getSyncFailureExplanation(final StandardProcessGroup group) {
try {
final Field versionControlFieldsField = StandardProcessGroup.class.getDeclaredField("versionControlFields");
versionControlFieldsField.setAccessible(true);
final VersionControlFields versionControlFields = (VersionControlFields) versionControlFieldsField.get(group);
return versionControlFields.getSyncFailureExplanation();
} catch (final ReflectiveOperationException e) {
throw new AssertionError(e);
}
}

private StandardProcessGroup createStandardProcessGroup(final String id, final String connectorId) {
return new StandardProcessGroup(
id,
Expand Down
Loading
Loading