updating NarUnpacker to use multiple library directories

This commit is contained in:
Brandon DeVries 2015-04-07 16:18:48 -04:00
parent b1970ecc10
commit 275735648a
6 changed files with 327 additions and 8 deletions

View File

@ -25,6 +25,7 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.nio.file.Files;
import java.nio.file.Path;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayList;
@ -41,7 +42,6 @@ import java.util.jar.Manifest;
import org.apache.nifi.util.FileUtils;
import org.apache.nifi.util.NiFiProperties;
import org.apache.nifi.util.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -61,22 +61,35 @@ public final class NarUnpacker {
};
public static ExtensionMapping unpackNars(final NiFiProperties props) {
final File narLibraryDir = props.getNarLibraryDirectory();
final List<Path> narLibraryDirs = props.getNarLibraryDirectories();
final File frameworkWorkingDir = props.getFrameworkWorkingDirectory();
final File extensionsWorkingDir = props.getExtensionsWorkingDirectory();
final File docsWorkingDir = props.getComponentDocumentationWorkingDirectory();
try {
// make sure the nar directories are there and accessible
FileUtils.ensureDirectoryExistAndCanAccess(narLibraryDir);
File unpackedFramework = null;
final Set<File> unpackedExtensions = new HashSet<>();
final List<File> narFiles = new ArrayList<>();
// make sure the nar directories are there and accessible
FileUtils.ensureDirectoryExistAndCanAccess(frameworkWorkingDir);
FileUtils.ensureDirectoryExistAndCanAccess(extensionsWorkingDir);
FileUtils.ensureDirectoryExistAndCanAccess(docsWorkingDir);
File unpackedFramework = null;
final Set<File> unpackedExtensions = new HashSet<>();
final File[] narFiles = narLibraryDir.listFiles(NAR_FILTER);
if (narFiles != null) {
for (Path narLibraryDir : narLibraryDirs) {
File narDir = narLibraryDir.toFile();
FileUtils.ensureDirectoryExistAndCanAccess(narDir);
File[] dirFiles = narDir.listFiles(NAR_FILTER);
if (dirFiles != null){
List<File> fileList = Arrays.asList(dirFiles);
narFiles.addAll(fileList);
}
}
if (!narFiles.isEmpty()) {
for (File narFile : narFiles) {
logger.debug("Expanding NAR file: " + narFile.getAbsolutePath());

View File

@ -0,0 +1,177 @@
package org.apache.nifi.nar;
import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.FileVisitResult;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.SimpleFileVisitor;
import java.nio.file.attribute.BasicFileAttributes;
import org.apache.nifi.util.NiFiProperties;
import org.junit.BeforeClass;
import org.junit.Test;
public class NarUnpackerTest {
@BeforeClass
public static void copyResources() throws IOException {
final Path sourcePath = Paths.get("./src/test/resources");
final Path targetPath = Paths.get("./target");
Files.walkFileTree(sourcePath, new SimpleFileVisitor<Path>() {
@Override
public FileVisitResult preVisitDirectory(Path dir, BasicFileAttributes attrs) throws IOException {
Path relativeSource = sourcePath.relativize(dir);
Path target = targetPath.resolve(relativeSource);
Files.createDirectories(target);
return FileVisitResult.CONTINUE;
}
@Override
public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException {
Path relativeSource = sourcePath.relativize(file);
Path target = targetPath.resolve(relativeSource);
Files.copy(file, target, REPLACE_EXISTING);
return FileVisitResult.CONTINUE;
}
});
}
@Test
public void testUnpackNars() {
NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties");
assertEquals("./target/NarUnpacker/lib/", properties.getProperty("nifi.nar.library.directory"));
assertEquals("./target/NarUnpacker/lib2/", properties.getProperty("nifi.nar.library.directory.alt"));
final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties);
assertEquals(2,extensionMapping.getAllExtensionNames().size());
assertTrue(extensionMapping.getAllExtensionNames().contains("org.apache.nifi.processors.dummy.one"));
assertTrue(extensionMapping.getAllExtensionNames().contains("org.apache.nifi.processors.dummy.two"));
final File extensionsWorkingDir = properties.getExtensionsWorkingDirectory();
File[] extensionFiles = extensionsWorkingDir.listFiles();
assertEquals(2,extensionFiles.length);
assertEquals("dummy-one.nar-unpacked", extensionFiles[0].getName());
assertEquals("dummy-two.nar-unpacked", extensionFiles[1].getName());
}
@Test
public void testUnpackNarsFromEmptyDir() throws IOException {
NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties");
final File emptyDir = new File ("./target/empty/dir");
emptyDir.delete();
emptyDir.deleteOnExit();
assertTrue(emptyDir.mkdirs());
properties.setProperty("nifi.nar.library.directory.alt", emptyDir.toString());
final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties);
assertEquals(1,extensionMapping.getAllExtensionNames().size());
assertTrue(extensionMapping.getAllExtensionNames().contains("org.apache.nifi.processors.dummy.one"));
final File extensionsWorkingDir = properties.getExtensionsWorkingDirectory();
File[] extensionFiles = extensionsWorkingDir.listFiles();
assertEquals(1,extensionFiles.length);
assertEquals("dummy-one.nar-unpacked", extensionFiles[0].getName());
}
@Test
public void testUnpackNarsFromNonExistantDir() {
final File nonExistantDir = new File ("./target/this/dir/should/not/exist/");
nonExistantDir.delete();
nonExistantDir.deleteOnExit();
NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties");
properties.setProperty("nifi.nar.library.directory.alt", nonExistantDir.toString());
final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties);
assertTrue(extensionMapping.getAllExtensionNames().contains("org.apache.nifi.processors.dummy.one"));
assertEquals(1,extensionMapping.getAllExtensionNames().size());
final File extensionsWorkingDir = properties.getExtensionsWorkingDirectory();
File[] extensionFiles = extensionsWorkingDir.listFiles();
assertEquals(1,extensionFiles.length);
assertEquals("dummy-one.nar-unpacked", extensionFiles[0].getName());
}
@Test
public void testUnpackNarsFromNonDir() throws IOException {
final File nonDir = new File ("./target/file.txt");
nonDir.createNewFile();
nonDir.deleteOnExit();
NiFiProperties properties = loadSpecifiedProperties("/NarUnpacker/conf/nifi.properties");
properties.setProperty("nifi.nar.library.directory.alt", nonDir.toString());
final ExtensionMapping extensionMapping = NarUnpacker.unpackNars(properties);
assertNull(extensionMapping);
}
private NiFiProperties loadSpecifiedProperties(String propertiesFile) {
String file = NarUnpackerTest.class.getResource(propertiesFile).getFile();
System.setProperty(NiFiProperties.PROPERTIES_FILE_PATH, file);
NiFiProperties properties = NiFiProperties.getInstance();
// clear out existing properties
for (String prop : properties.stringPropertyNames()) {
properties.remove(prop);
}
InputStream inStream = null;
try {
inStream = new BufferedInputStream(new FileInputStream(file));
properties.load(inStream);
} catch (final Exception ex) {
throw new RuntimeException("Cannot load properties file due to " + ex.getLocalizedMessage(), ex);
} finally {
if (null != inStream) {
try {
inStream.close();
} catch (final Exception ex) {
/**
* do nothing *
*/
}
}
}
return properties;
}
}

View File

@ -0,0 +1,129 @@
# 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.
# Core Properties #
nifi.version=nifi-test 3.0.0
nifi.flow.configuration.file=./target/flow.xml.gz
nifi.flow.configuration.archive.dir=./target/archive/
nifi.flowcontroller.autoResumeState=true
nifi.flowcontroller.graceful.shutdown.period=10 sec
nifi.flowservice.writedelay.interval=2 sec
nifi.administrative.yield.duration=30 sec
nifi.reporting.task.configuration.file=./target/reporting-tasks.xml
nifi.controller.service.configuration.file=./target/controller-services.xml
nifi.templates.directory=./target/templates
nifi.ui.banner.text=UI Banner Text
nifi.ui.autorefresh.interval=30 sec
nifi.nar.library.directory=./target/NarUnpacker/lib/
nifi.nar.library.directory.alt=./target/NarUnpacker/lib2/
nifi.nar.working.directory=./target/work/nar/
# H2 Settings
nifi.database.directory=./target/database_repository
nifi.h2.url.append=;LOCK_TIMEOUT=25000;WRITE_DELAY=0;AUTO_SERVER=FALSE
# FlowFile Repository
nifi.flowfile.repository.directory=./target/test-repo
nifi.flowfile.repository.partitions=1
nifi.flowfile.repository.checkpoint.interval=2 mins
nifi.queue.swap.threshold=20000
nifi.swap.storage.directory=./target/test-repo/swap
nifi.swap.in.period=5 sec
nifi.swap.in.threads=1
nifi.swap.out.period=5 sec
nifi.swap.out.threads=4
# Content Repository
nifi.content.claim.max.appendable.size=10 MB
nifi.content.claim.max.flow.files=100
nifi.content.repository.directory.default=./target/content_repository
# Provenance Repository Properties
nifi.provenance.repository.storage.directory=./target/provenance_repository
nifi.provenance.repository.max.storage.time=24 hours
nifi.provenance.repository.max.storage.size=1 GB
nifi.provenance.repository.rollover.time=5 mins
nifi.provenance.repository.rollover.size=100 MB
# Site to Site properties
nifi.remote.input.socket.port=9990
nifi.remote.input.secure=true
# web properties #
nifi.web.war.directory=./target/lib
nifi.web.http.host=
nifi.web.http.port=8080
nifi.web.https.host=
nifi.web.https.port=
nifi.web.jetty.working.directory=./target/work/jetty
# security properties #
nifi.sensitive.props.key=key
nifi.sensitive.props.algorithm=PBEWITHMD5AND256BITAES-CBC-OPENSSL
nifi.sensitive.props.provider=BC
nifi.security.keystore=
nifi.security.keystoreType=
nifi.security.keystorePasswd=
nifi.security.keyPasswd=
nifi.security.truststore=
nifi.security.truststoreType=
nifi.security.truststorePasswd=
nifi.security.needClientAuth=
nifi.security.authorizedUsers.file=./target/conf/authorized-users.xml
nifi.security.user.credential.cache.duration=24 hours
nifi.security.user.authority.provider=nifi.authorization.FileAuthorizationProvider
nifi.security.support.new.account.requests=
nifi.security.default.user.roles=
# cluster common properties (cluster manager and nodes must have same values) #
nifi.cluster.protocol.heartbeat.interval=5 sec
nifi.cluster.protocol.is.secure=false
nifi.cluster.protocol.socket.timeout=30 sec
nifi.cluster.protocol.connection.handshake.timeout=45 sec
# if multicast is used, then nifi.cluster.protocol.multicast.xxx properties must be configured #
nifi.cluster.protocol.use.multicast=false
nifi.cluster.protocol.multicast.address=
nifi.cluster.protocol.multicast.port=
nifi.cluster.protocol.multicast.service.broadcast.delay=500 ms
nifi.cluster.protocol.multicast.service.locator.attempts=3
nifi.cluster.protocol.multicast.service.locator.attempts.delay=1 sec
# cluster node properties (only configure for cluster nodes) #
nifi.cluster.is.node=false
nifi.cluster.node.address=
nifi.cluster.node.protocol.port=
nifi.cluster.node.protocol.threads=2
# if multicast is not used, nifi.cluster.node.unicast.xxx must have same values as nifi.cluster.manager.xxx #
nifi.cluster.node.unicast.manager.address=
nifi.cluster.node.unicast.manager.protocol.port=
nifi.cluster.node.unicast.manager.authority.provider.port=
# cluster manager properties (only configure for cluster manager) #
nifi.cluster.is.manager=false
nifi.cluster.manager.address=
nifi.cluster.manager.protocol.port=
nifi.cluster.manager.authority.provider.port=
nifi.cluster.manager.authority.provider.threads=10
nifi.cluster.manager.node.firewall.file=
nifi.cluster.manager.node.event.history.size=10
nifi.cluster.manager.node.api.connection.timeout=30 sec
nifi.cluster.manager.node.api.read.timeout=30 sec
nifi.cluster.manager.node.api.request.threads=10
nifi.cluster.manager.flow.retrieval.delay=5 sec
nifi.cluster.manager.protocol.threads=10
nifi.cluster.manager.safemode.duration=0 sec