Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-555: Create index name based off of event time of first event in index, not based on creation time of index

This commit is contained in:
Mark Payne 2015-04-29 09:41:40 -04:00
parent 20831c87fc
commit 4baf48ae95
3 changed files with 15 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/IndexConfiguration.java
View File

@ -121,13 +121,13 @@ public class IndexConfiguration {
} }
} }
public File getWritableIndexDirectory(final File provenanceLogFile) { public File getWritableIndexDirectory(final File provenanceLogFile, final long newIndexTimestamp) {
lock.lock(); lock.lock();
try { try {
final File storageDirectory = provenanceLogFile.getParentFile(); final File storageDirectory = provenanceLogFile.getParentFile();
List<File> indexDirectories = this.indexDirectoryMap.get(storageDirectory); List<File> indexDirectories = this.indexDirectoryMap.get(storageDirectory);
if (indexDirectories == null) { if (indexDirectories == null) {
final File newDir = addNewIndex(storageDirectory, provenanceLogFile); final File newDir = addNewIndex(storageDirectory, provenanceLogFile, newIndexTimestamp);
indexDirectories = new ArrayList<>(); indexDirectories = new ArrayList<>();
indexDirectories.add(newDir); indexDirectories.add(newDir);
indexDirectoryMap.put(storageDirectory, indexDirectories); indexDirectoryMap.put(storageDirectory, indexDirectories);
@ -135,7 +135,7 @@ public class IndexConfiguration {
} }
if (indexDirectories.isEmpty()) { if (indexDirectories.isEmpty()) {
final File newDir = addNewIndex(storageDirectory, provenanceLogFile); final File newDir = addNewIndex(storageDirectory, provenanceLogFile, newIndexTimestamp);
indexDirectories.add(newDir); indexDirectories.add(newDir);
return newDir; return newDir;
} }
@ -143,7 +143,7 @@ public class IndexConfiguration {
final File lastDir = indexDirectories.get(indexDirectories.size() - 1); final File lastDir = indexDirectories.get(indexDirectories.size() - 1);
final long size = getSize(lastDir); final long size = getSize(lastDir);
if (size > repoConfig.getDesiredIndexSize()) { if (size > repoConfig.getDesiredIndexSize()) {
final File newDir = addNewIndex(storageDirectory, provenanceLogFile); final File newDir = addNewIndex(storageDirectory, provenanceLogFile, newIndexTimestamp);
indexDirectories.add(newDir); indexDirectories.add(newDir);
return newDir; return newDir;
} else { } else {
@ -154,14 +154,14 @@ public class IndexConfiguration {
} }
} }
private File addNewIndex(final File storageDirectory, final File provenanceLogFile) { private File addNewIndex(final File storageDirectory, final File provenanceLogFile, final long newIndexTimestamp) {
// Build the event time of the first record into the index's filename so that we can determine // Build the event time of the first record into the index's filename so that we can determine
// which index files to look at when we perform a search. We use the timestamp of the first record // which index files to look at when we perform a search. We use the timestamp of the first record
// in the Provenance Log file, rather than the current time, because we may perform the Indexing // in the Provenance Log file, rather than the current time, because we may perform the Indexing
// retroactively. // retroactively.
Long firstEntryTime = getFirstEntryTime(provenanceLogFile); Long firstEntryTime = getFirstEntryTime(provenanceLogFile);
if (firstEntryTime == null) { if (firstEntryTime == null) {
firstEntryTime = System.currentTimeMillis(); firstEntryTime = newIndexTimestamp;
} }
return new File(storageDirectory, "index-" + firstEntryTime); return new File(storageDirectory, "index-" + firstEntryTime);
} }
@ -222,7 +222,7 @@ public class IndexConfiguration {
} }
}); });
for (File indexDir : sortedIndexDirectories) { for (final File indexDir : sortedIndexDirectories) {
// If the index was last modified before the start time, we know that it doesn't // If the index was last modified before the start time, we know that it doesn't
// contain any data for us to query. // contain any data for us to query.
if (startTime != null && indexDir.lastModified() < startTime) { if (startTime != null && indexDir.lastModified() < startTime) {
@ -282,7 +282,7 @@ public class IndexConfiguration {
} }
boolean foundIndexCreatedLater = false; boolean foundIndexCreatedLater = false;
for (File indexDir : sortedIndexDirectories) { for (final File indexDir : sortedIndexDirectories) {
// If the index was last modified before the log file was created, we know the index doesn't include // If the index was last modified before the log file was created, we know the index doesn't include
// any data for the provenance log. // any data for the provenance log.
if (indexDir.lastModified() < firstEntryTime) { if (indexDir.lastModified() < firstEntryTime) {

6
nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/PersistentProvenanceRepository.java
View File

@ -1231,6 +1231,7 @@ public class PersistentProvenanceRepository implements ProvenanceEventRepository
} }
}); });
long earliestTimestamp = System.currentTimeMillis();
for (final RecordReader reader : readers) { for (final RecordReader reader : readers) {
StandardProvenanceEventRecord record = null; StandardProvenanceEventRecord record = null;
@ -1252,6 +1253,9 @@ public class PersistentProvenanceRepository implements ProvenanceEventRepository
continue; continue;
} }
if ( record.getEventTime() < earliestTimestamp ) {
earliestTimestamp = record.getEventTime();
}
recordToReaderMap.put(record, reader); recordToReaderMap.put(record, reader);
} }
@ -1262,7 +1266,7 @@ public class PersistentProvenanceRepository implements ProvenanceEventRepository
final IndexingAction indexingAction = new IndexingAction(this); final IndexingAction indexingAction = new IndexingAction(this);
final File indexingDirectory = indexConfig.getWritableIndexDirectory(writerFile); final File indexingDirectory = indexConfig.getWritableIndexDirectory(writerFile, earliestTimestamp);
final IndexWriter indexWriter = indexManager.borrowIndexWriter(indexingDirectory); final IndexWriter indexWriter = indexManager.borrowIndexWriter(indexingDirectory);
try { try {
long maxId = 0L; long maxId = 0L;

4
nifi/nifi-nar-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/lucene/LineageQuery.java
View File

@ -96,8 +96,8 @@ public class LineageQuery {
final DocsReader docsReader = new DocsReader(repo.getConfiguration().getStorageDirectories()); final DocsReader docsReader = new DocsReader(repo.getConfiguration().getStorageDirectories());
final Set<ProvenanceEventRecord> recs = docsReader.read(uuidQueryTopDocs, searcher.getIndexReader(), repo.getAllLogFiles(), new AtomicInteger(0), Integer.MAX_VALUE); final Set<ProvenanceEventRecord> recs = docsReader.read(uuidQueryTopDocs, searcher.getIndexReader(), repo.getAllLogFiles(), new AtomicInteger(0), Integer.MAX_VALUE);
final long readDocsEnd = System.nanoTime(); final long readDocsEnd = System.nanoTime();
logger.debug("Finished Lineage Query; Lucene search took {} millis, reading records took {} millis", logger.debug("Finished Lineage Query against {}; Lucene search took {} millis, reading records took {} millis",
TimeUnit.NANOSECONDS.toMillis(searchEnd - searchStart), TimeUnit.NANOSECONDS.toMillis(readDocsEnd - searchEnd)); indexDirectory, TimeUnit.NANOSECONDS.toMillis(searchEnd - searchStart), TimeUnit.NANOSECONDS.toMillis(readDocsEnd - searchEnd));
return recs; return recs;
} finally { } finally {