@Test public void testArchiveStorageCountLimit() throws Exception { final long intervalMillis = 60_000; File[] oldArchives = new File[5]; createSimulatedOldArchives(oldArchives, intervalMillis); final FlowConfigurationArchiveManager archiveManager = createArchiveManager(2,null, null); final File archive = archiveManager.archive(); assertTrue(!oldArchives[0].exists()); assertTrue(!oldArchives[1].exists()); assertTrue(!oldArchives[2].exists()); assertTrue(!oldArchives[3].exists()); assertTrue(oldArchives[4].exists()); assertTrue(archive.exists()); assertTrue("Original file should remain intact", flowFile.isFile()); }
public File archive() throws IOException { final String originalFlowConfigFileName = flowConfigFile.getFileName().toString(); final File archiveFile = setupArchiveFile(); final long now = System.currentTimeMillis(); final AtomicLong totalArchiveSize = new AtomicLong(0); final List<Path> archives = Files.walk(archiveDir, 1).filter(p -> { final String filename = p.getFileName().toString(); if (Files.isRegularFile(p) && filename.endsWith("_" + originalFlowConfigFileName)) { final Matcher matcher = archiveFilenamePattern.matcher(filename); if (matcher.matches() && filename.equals(matcher.group(1) + "_" + originalFlowConfigFileName)) { try { totalArchiveSize.getAndAdd(Files.size(p)); } catch (IOException e) { logger.warn("Failed to get file size of {} due to {}", p, e); } return true; } } return false; }).collect(Collectors.toList()); archives.sort(Comparator.comparingLong(path -> path.toFile().lastModified())); logger.debug("archives={}", archives); final int archiveCount = archives.size(); final long flowConfigFileSize = Files.size(flowConfigFile); IntStream.range(0, archiveCount).filter(i -> { boolean old = maxCount != null && maxCount > 0 && (archiveCount - i) > maxCount - 1; final File archive = archives.get(i).toFile(); old = old || (maxTimeMillis != null && maxTimeMillis > 0 && (now - archive.lastModified()) > maxTimeMillis); old = old || (maxStorageBytes != null && maxStorageBytes > 0 && (totalArchiveSize.get() + flowConfigFileSize > maxStorageBytes)); if (old) { totalArchiveSize.getAndAdd(archive.length() * -1); logger.info("Removing old archive file {} to reduce storage usage. currentSize={}", archive, totalArchiveSize); } return old; }).forEach(i -> { try { Files.delete(archives.get(i)); } catch (IOException e) { logger.warn("Failed to delete {} to reduce storage usage, due to {}", archives.get(i), e); } }); Files.copy(flowConfigFile, archiveFile.toPath(), StandardCopyOption.REPLACE_EXISTING); if (maxStorageBytes != null && maxStorageBytes > 0 && flowConfigFileSize > maxStorageBytes) { logger.warn("Size of {} ({}) exceeds configured maxStorage size ({}). Archive won't be able to keep old files.", flowConfigFile, flowConfigFileSize, maxStorageBytes); } return archiveFile; }
FlowConfigurationArchiveManager { public File archive() throws IOException { final String originalFlowConfigFileName = flowConfigFile.getFileName().toString(); final File archiveFile = setupArchiveFile(); final long now = System.currentTimeMillis(); final AtomicLong totalArchiveSize = new AtomicLong(0); final List<Path> archives = Files.walk(archiveDir, 1).filter(p -> { final String filename = p.getFileName().toString(); if (Files.isRegularFile(p) && filename.endsWith("_" + originalFlowConfigFileName)) { final Matcher matcher = archiveFilenamePattern.matcher(filename); if (matcher.matches() && filename.equals(matcher.group(1) + "_" + originalFlowConfigFileName)) { try { totalArchiveSize.getAndAdd(Files.size(p)); } catch (IOException e) { logger.warn("Failed to get file size of {} due to {}", p, e); } return true; } } return false; }).collect(Collectors.toList()); archives.sort(Comparator.comparingLong(path -> path.toFile().lastModified())); logger.debug("archives={}", archives); final int archiveCount = archives.size(); final long flowConfigFileSize = Files.size(flowConfigFile); IntStream.range(0, archiveCount).filter(i -> { boolean old = maxCount != null && maxCount > 0 && (archiveCount - i) > maxCount - 1; final File archive = archives.get(i).toFile(); old = old || (maxTimeMillis != null && maxTimeMillis > 0 && (now - archive.lastModified()) > maxTimeMillis); old = old || (maxStorageBytes != null && maxStorageBytes > 0 && (totalArchiveSize.get() + flowConfigFileSize > maxStorageBytes)); if (old) { totalArchiveSize.getAndAdd(archive.length() * -1); logger.info("Removing old archive file {} to reduce storage usage. currentSize={}", archive, totalArchiveSize); } return old; }).forEach(i -> { try { Files.delete(archives.get(i)); } catch (IOException e) { logger.warn("Failed to delete {} to reduce storage usage, due to {}", archives.get(i), e); } }); Files.copy(flowConfigFile, archiveFile.toPath(), StandardCopyOption.REPLACE_EXISTING); if (maxStorageBytes != null && maxStorageBytes > 0 && flowConfigFileSize > maxStorageBytes) { logger.warn("Size of {} ({}) exceeds configured maxStorage size ({}). Archive won't be able to keep old files.", flowConfigFile, flowConfigFileSize, maxStorageBytes); } return archiveFile; } }
FlowConfigurationArchiveManager { public File archive() throws IOException { final String originalFlowConfigFileName = flowConfigFile.getFileName().toString(); final File archiveFile = setupArchiveFile(); final long now = System.currentTimeMillis(); final AtomicLong totalArchiveSize = new AtomicLong(0); final List<Path> archives = Files.walk(archiveDir, 1).filter(p -> { final String filename = p.getFileName().toString(); if (Files.isRegularFile(p) && filename.endsWith("_" + originalFlowConfigFileName)) { final Matcher matcher = archiveFilenamePattern.matcher(filename); if (matcher.matches() && filename.equals(matcher.group(1) + "_" + originalFlowConfigFileName)) { try { totalArchiveSize.getAndAdd(Files.size(p)); } catch (IOException e) { logger.warn("Failed to get file size of {} due to {}", p, e); } return true; } } return false; }).collect(Collectors.toList()); archives.sort(Comparator.comparingLong(path -> path.toFile().lastModified())); logger.debug("archives={}", archives); final int archiveCount = archives.size(); final long flowConfigFileSize = Files.size(flowConfigFile); IntStream.range(0, archiveCount).filter(i -> { boolean old = maxCount != null && maxCount > 0 && (archiveCount - i) > maxCount - 1; final File archive = archives.get(i).toFile(); old = old || (maxTimeMillis != null && maxTimeMillis > 0 && (now - archive.lastModified()) > maxTimeMillis); old = old || (maxStorageBytes != null && maxStorageBytes > 0 && (totalArchiveSize.get() + flowConfigFileSize > maxStorageBytes)); if (old) { totalArchiveSize.getAndAdd(archive.length() * -1); logger.info("Removing old archive file {} to reduce storage usage. currentSize={}", archive, totalArchiveSize); } return old; }).forEach(i -> { try { Files.delete(archives.get(i)); } catch (IOException e) { logger.warn("Failed to delete {} to reduce storage usage, due to {}", archives.get(i), e); } }); Files.copy(flowConfigFile, archiveFile.toPath(), StandardCopyOption.REPLACE_EXISTING); if (maxStorageBytes != null && maxStorageBytes > 0 && flowConfigFileSize > maxStorageBytes) { logger.warn("Size of {} ({}) exceeds configured maxStorage size ({}). Archive won't be able to keep old files.", flowConfigFile, flowConfigFileSize, maxStorageBytes); } return archiveFile; } FlowConfigurationArchiveManager(final Path flowConfigFile, final NiFiProperties properties); }
FlowConfigurationArchiveManager { public File archive() throws IOException { final String originalFlowConfigFileName = flowConfigFile.getFileName().toString(); final File archiveFile = setupArchiveFile(); final long now = System.currentTimeMillis(); final AtomicLong totalArchiveSize = new AtomicLong(0); final List<Path> archives = Files.walk(archiveDir, 1).filter(p -> { final String filename = p.getFileName().toString(); if (Files.isRegularFile(p) && filename.endsWith("_" + originalFlowConfigFileName)) { final Matcher matcher = archiveFilenamePattern.matcher(filename); if (matcher.matches() && filename.equals(matcher.group(1) + "_" + originalFlowConfigFileName)) { try { totalArchiveSize.getAndAdd(Files.size(p)); } catch (IOException e) { logger.warn("Failed to get file size of {} due to {}", p, e); } return true; } } return false; }).collect(Collectors.toList()); archives.sort(Comparator.comparingLong(path -> path.toFile().lastModified())); logger.debug("archives={}", archives); final int archiveCount = archives.size(); final long flowConfigFileSize = Files.size(flowConfigFile); IntStream.range(0, archiveCount).filter(i -> { boolean old = maxCount != null && maxCount > 0 && (archiveCount - i) > maxCount - 1; final File archive = archives.get(i).toFile(); old = old || (maxTimeMillis != null && maxTimeMillis > 0 && (now - archive.lastModified()) > maxTimeMillis); old = old || (maxStorageBytes != null && maxStorageBytes > 0 && (totalArchiveSize.get() + flowConfigFileSize > maxStorageBytes)); if (old) { totalArchiveSize.getAndAdd(archive.length() * -1); logger.info("Removing old archive file {} to reduce storage usage. currentSize={}", archive, totalArchiveSize); } return old; }).forEach(i -> { try { Files.delete(archives.get(i)); } catch (IOException e) { logger.warn("Failed to delete {} to reduce storage usage, due to {}", archives.get(i), e); } }); Files.copy(flowConfigFile, archiveFile.toPath(), StandardCopyOption.REPLACE_EXISTING); if (maxStorageBytes != null && maxStorageBytes > 0 && flowConfigFileSize > maxStorageBytes) { logger.warn("Size of {} ({}) exceeds configured maxStorage size ({}). Archive won't be able to keep old files.", flowConfigFile, flowConfigFileSize, maxStorageBytes); } return archiveFile; } FlowConfigurationArchiveManager(final Path flowConfigFile, final NiFiProperties properties); File setupArchiveFile(); File archive(); }
FlowConfigurationArchiveManager { public File archive() throws IOException { final String originalFlowConfigFileName = flowConfigFile.getFileName().toString(); final File archiveFile = setupArchiveFile(); final long now = System.currentTimeMillis(); final AtomicLong totalArchiveSize = new AtomicLong(0); final List<Path> archives = Files.walk(archiveDir, 1).filter(p -> { final String filename = p.getFileName().toString(); if (Files.isRegularFile(p) && filename.endsWith("_" + originalFlowConfigFileName)) { final Matcher matcher = archiveFilenamePattern.matcher(filename); if (matcher.matches() && filename.equals(matcher.group(1) + "_" + originalFlowConfigFileName)) { try { totalArchiveSize.getAndAdd(Files.size(p)); } catch (IOException e) { logger.warn("Failed to get file size of {} due to {}", p, e); } return true; } } return false; }).collect(Collectors.toList()); archives.sort(Comparator.comparingLong(path -> path.toFile().lastModified())); logger.debug("archives={}", archives); final int archiveCount = archives.size(); final long flowConfigFileSize = Files.size(flowConfigFile); IntStream.range(0, archiveCount).filter(i -> { boolean old = maxCount != null && maxCount > 0 && (archiveCount - i) > maxCount - 1; final File archive = archives.get(i).toFile(); old = old || (maxTimeMillis != null && maxTimeMillis > 0 && (now - archive.lastModified()) > maxTimeMillis); old = old || (maxStorageBytes != null && maxStorageBytes > 0 && (totalArchiveSize.get() + flowConfigFileSize > maxStorageBytes)); if (old) { totalArchiveSize.getAndAdd(archive.length() * -1); logger.info("Removing old archive file {} to reduce storage usage. currentSize={}", archive, totalArchiveSize); } return old; }).forEach(i -> { try { Files.delete(archives.get(i)); } catch (IOException e) { logger.warn("Failed to delete {} to reduce storage usage, due to {}", archives.get(i), e); } }); Files.copy(flowConfigFile, archiveFile.toPath(), StandardCopyOption.REPLACE_EXISTING); if (maxStorageBytes != null && maxStorageBytes > 0 && flowConfigFileSize > maxStorageBytes) { logger.warn("Size of {} ({}) exceeds configured maxStorage size ({}). Archive won't be able to keep old files.", flowConfigFile, flowConfigFileSize, maxStorageBytes); } return archiveFile; } FlowConfigurationArchiveManager(final Path flowConfigFile, final NiFiProperties properties); File setupArchiveFile(); File archive();  }