| 38 | namespace repository { |
| 39 | |
| 40 | void FlowFileRepository::flush() { |
| 41 | auto opendb = db_->open(); |
| 42 | if (!opendb) { |
| 43 | return; |
| 44 | } |
| 45 | rocksdb::WriteBatch batch; |
| 46 | rocksdb::ReadOptions options; |
| 47 | |
| 48 | std::vector<std::shared_ptr<FlowFile>> purgeList; |
| 49 | |
| 50 | std::vector<rocksdb::Slice> keys; |
| 51 | std::list<std::string> keystrings; |
| 52 | std::vector<std::string> values; |
| 53 | |
| 54 | while (keys_to_delete.size_approx() > 0) { |
| 55 | std::string key; |
| 56 | if (keys_to_delete.try_dequeue(key)) { |
| 57 | keystrings.push_back(std::move(key)); // rocksdb::Slice doesn't copy the string, only grabs ptrs. Hacky, but have to ensure the required lifetime of the strings. |
| 58 | keys.push_back(keystrings.back()); |
| 59 | } |
| 60 | } |
| 61 | |
| 62 | auto multistatus = opendb->MultiGet(options, keys, &values); |
| 63 | |
| 64 | for(size_t i=0; i<keys.size() && i<values.size() && i<multistatus.size(); ++i) { |
| 65 | if(!multistatus[i].ok()) { |
| 66 | logger_->log_error("Failed to read key from rocksdb: %s! DB is most probably in an inconsistent state!", keys[i].data()); |
| 67 | keystrings.remove(keys[i].data()); |
| 68 | continue; |
| 69 | } |
| 70 | |
| 71 | utils::Identifier containerId; |
| 72 | auto eventRead = FlowFileRecord::DeSerialize(reinterpret_cast<const uint8_t *>(values[i].data()), gsl::narrow<int>(values[i].size()), content_repo_, containerId); |
| 73 | if (eventRead) { |
| 74 | purgeList.push_back(eventRead); |
| 75 | } |
| 76 | logger_->log_debug("Issuing batch delete, including %s, Content path %s", eventRead->getUUIDStr(), eventRead->getContentFullPath()); |
| 77 | batch.Delete(keys[i]); |
| 78 | } |
| 79 | |
| 80 | auto operation = [&batch, &opendb]() { return opendb->Write(rocksdb::WriteOptions(), &batch); }; |
| 81 | |
| 82 | if (!ExecuteWithRetry(operation)) { |
| 83 | for (const auto& key: keystrings) { |
| 84 | keys_to_delete.enqueue(key); // Push back the values that we could get but couldn't delete |
| 85 | } |
| 86 | return; // Stop here - don't delete from content repo while we have records in FF repo |
| 87 | } |
| 88 | |
| 89 | if (content_repo_) { |
| 90 | for (const auto &ffr : purgeList) { |
| 91 | auto claim = ffr->getResourceClaim(); |
| 92 | if (claim) claim->decreaseFlowFileRecordOwnedCount(); |
| 93 | } |
| 94 | } |
| 95 | } |
| 96 | |
| 97 | void FlowFileRepository::printStats() { |
no test coverage detected