// Copyright (c) 2007-2008 Facebook

//

// Licensed 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.

//

// See accompanying file LICENSE or visit the Scribe site at:

// http://developers.facebook.com/scribe/

//

// @author Bobby Johnson

// @author James Wang

// @author Jason Sobel

// @author Alex Moskalyuk

// @author Avinash Lakshman

// @author Anthony Giardullo

// @author Jan Oravec

// @author John Song

#include <algorithm>

#include "common.h"

#include "scribe_server.h"

#include "network_dynamic_config.h"

#include <boost/algorithm/string.hpp>

using namespace std;

using namespace boost;

using namespace boost::filesystem;

using namespace apache::thrift;

using namespace apache::thrift::protocol;

using namespace apache::thrift::transport;

using namespace apache::thrift::server;

using namespace scribe::thrift;

#define DEFAULT_FILESTORE_MAX_SIZE 1000000000

#define DEFAULT_FILESTORE_MAX_WRITE_SIZE 1000000

#define DEFAULT_FILESTORE_ROLL_HOUR 1

#define DEFAULT_FILESTORE_ROLL_MINUTE 15

#define DEFAULT_BUFFERSTORE_SEND_RATE 1

#define DEFAULT_BUFFERSTORE_AVG_RETRY_INTERVAL 300

#define DEFAULT_BUFFERSTORE_RETRY_INTERVAL_RANGE 60

#define DEFAULT_BUCKETSTORE_DELIMITER ':'

#define DEFAULT_NETWORKSTORE_CACHE_TIMEOUT 300

#define DEFAULT_BUFFERSTORE_BYPASS_MAXQSIZE_RATIO 0.75

// magic threshold

#define DEFAULT_NETWORKSTORE_DUMMY_THRESHOLD 4096

// Parameters for adaptive_backoff

#define DEFAULT_MIN_RETRY 5

#define DEFAULT_MAX_RETRY 100

#define DEFAULT_RANDOM_OFFSET_RANGE 20

const double MULT_INC_FACTOR = 1.414; //sqrt(2)

#define ADD_DEC_FACTOR 2

#define CONT_SUCCESS_THRESHOLD 1

ConnPool g_connPool;

const string meta_logfile_prefix = "scribe_meta<new_logfile>: ";

// Checks if we should try sending a dummy Log in the n/w store

bool shouldSendDummy(boost::shared_ptr<logentry_vector_t> messages) {

size_t size = 0;

for (logentry_vector_t::iterator iter = messages->begin();

iter != messages->end(); ++iter) {

size += (**iter).message.size();

if (size > DEFAULT_NETWORKSTORE_DUMMY_THRESHOLD) {

return true;

}

}

return false;

}

boost::shared_ptr<Store>

Store::createStore(StoreQueue* storeq, const string& type,

const string& category, bool readable,

bool multi_category) {

if (0 == type.compare("file")) {

return shared_ptr<Store>(new FileStore(storeq, category, multi_category,

readable));

} else if (0 == type.compare("buffer")) {

return shared_ptr<Store>(new BufferStore(storeq,category, multi_category));

} else if (0 == type.compare("network")) {

return shared_ptr<Store>(new NetworkStore(storeq, category,

multi_category));

} else if (0 == type.compare("bucket")) {

return shared_ptr<Store>(new BucketStore(storeq, category,

multi_category));

} else if (0 == type.compare("thriftfile")) {

return shared_ptr<Store>(new ThriftFileStore(storeq, category,

multi_category));

} else if (0 == type.compare("null")) {

return shared_ptr<Store>(new NullStore(storeq, category, multi_category));

} else if (0 == type.compare("multi")) {

return shared_ptr<Store>(new MultiStore(storeq, category, multi_category));

} else if (0 == type.compare("category")) {

return shared_ptr<Store>(new CategoryStore(storeq, category,

multi_category));

} else if (0 == type.compare("multifile")) {

return shared_ptr<Store>(new MultiFileStore(storeq, category,

multi_category));

} else if (0 == type.compare("thriftmultifile")) {

return shared_ptr<Store>(new ThriftMultiFileStore(storeq, category,

multi_category));

} else {

return shared_ptr<Store>();

}

}

Store::Store(StoreQueue* storeq,

const string& category,

const string &type,

bool multi_category)

: categoryHandled(category),

multiCategory(multi_category),

storeType(type),

storeQueue(storeq) {

pthread_mutex_init(&statusMutex, NULL);

}

Store::~Store() {

pthread_mutex_destroy(&statusMutex);

}

void Store::configure(pStoreConf configuration, pStoreConf parent) {

storeConf = configuration;

storeConf->setParent(parent);

}

void Store::setStatus(const std::string& new_status) {

pthread_mutex_lock(&statusMutex);

status = new_status;

pthread_mutex_unlock(&statusMutex);

}

std::string Store::getStatus() {

pthread_mutex_lock(&statusMutex);

std::string return_status(status);

pthread_mutex_unlock(&statusMutex);

return return_status;

}

bool Store::readOldest(/*out*/ boost::shared_ptr<logentry_vector_t> messages,

struct tm* now) {

LOG_OPER("[%s] ERROR: attempting to read from a write-only store",

categoryHandled.c_str());

return false;

}

bool Store::replaceOldest(boost::shared_ptr<logentry_vector_t> messages,

struct tm* now) {

LOG_OPER("[%s] ERROR: attempting to read from a write-only store",

categoryHandled.c_str());

return false;

}

void Store::deleteOldest(struct tm* now) {

LOG_OPER("[%s] ERROR: attempting to read from a write-only store",

categoryHandled.c_str());

}

bool Store::empty(struct tm* now) {

LOG_OPER("[%s] ERROR: attempting to read from a write-only store",

categoryHandled.c_str());

return true;

}

const std::string& Store::getType() {

return storeType;

}

FileStoreBase::FileStoreBase(StoreQueue* storeq,

const string& category,

const string &type, bool multi_category)

: Store(storeq, category, type, multi_category),

baseFilePath("/tmp"),

subDirectory(""),

filePath("/tmp"),

baseFileName(category),

baseSymlinkName(""),

maxSize(DEFAULT_FILESTORE_MAX_SIZE),

maxWriteSize(DEFAULT_FILESTORE_MAX_WRITE_SIZE),

rollPeriod(ROLL_NEVER),

rollPeriodLength(0),

rollHour(DEFAULT_FILESTORE_ROLL_HOUR),

rollMinute(DEFAULT_FILESTORE_ROLL_MINUTE),

fsType("std"),

chunkSize(0),

writeMeta(false),

writeCategory(false),

createSymlink(true),

writeStats(false),

rotateOnReopen(false),

currentSize(0),

lastRollTime(0),

eventsWritten(0) {

}

FileStoreBase::~FileStoreBase() {

}

void FileStoreBase::configure(pStoreConf configuration, pStoreConf parent) {

Store::configure(configuration, parent);

// We can run using defaults for all of these, but there are

// a couple of suspicious things we warn about.

std::string tmp;

configuration->getString("file_path", baseFilePath);

configuration->getString("sub_directory", subDirectory);

configuration->getString("use_hostname_sub_directory", tmp);

if (0 == tmp.compare("yes")) {

setHostNameSubDir();

}

filePath = baseFilePath;

if (!subDirectory.empty()) {

filePath += "/" + subDirectory;

}

if (!configuration->getString("base_filename", baseFileName)) {

LOG_OPER(

"[%s] WARNING: Bad config - no base_filename specified for file store",

categoryHandled.c_str());

}

// check if symlink name is optionally specified

configuration->getString("base_symlink_name", baseSymlinkName);

if (configuration->getString("rotate_period", tmp)) {

if (0 == tmp.compare("hourly")) {

rollPeriod = ROLL_HOURLY;

} else if (0 == tmp.compare("daily")) {

rollPeriod = ROLL_DAILY;

} else if (0 == tmp.compare("never")) {

rollPeriod = ROLL_NEVER;

} else {

errno = 0;

char* endptr;

rollPeriod = ROLL_OTHER;

rollPeriodLength = strtol(tmp.c_str(), &endptr, 10);

bool ok = errno == 0 && rollPeriodLength > 0 && endptr != tmp.c_str() &&

(*endptr == '\0' || endptr[1] == '\0');

switch (*endptr) {

case 'w':

rollPeriodLength *= 60 * 60 * 24 * 7;

break;

case 'd':

rollPeriodLength *= 60 * 60 * 24;

break;

case 'h':

rollPeriodLength *= 60 * 60;

break;

case 'm':

rollPeriodLength *= 60;

break;

case 's':

case '\0':

break;

default:

ok = false;

break;

}

if (!ok) {

rollPeriod = ROLL_NEVER;

LOG_OPER("[%s] WARNING: Bad config - invalid format of rotate_period, rotations disabled",

categoryHandled.c_str());

}

}

}

if (configuration->getString("write_meta", tmp)) {

if (0 == tmp.compare("yes")) {

writeMeta = true;

}

}

if (configuration->getString("write_category", tmp)) {

if (0 == tmp.compare("yes")) {

writeCategory = true;

}

}

if (configuration->getString("create_symlink", tmp)) {

if (0 == tmp.compare("yes")) {

createSymlink = true;

} else {

createSymlink = false;

}

}

if (configuration->getString("write_stats", tmp)) {

if (0 == tmp.compare("yes")) {

writeStats = true;

} else {

writeStats = false;

}

}

configuration->getString("fs_type", fsType);

configuration->getUnsigned("max_size", maxSize);

if(0 == maxSize) {

maxSize = ULONG_MAX;

}

configuration->getUnsigned("max_write_size", maxWriteSize);

configuration->getUnsigned("rotate_hour", rollHour);

configuration->getUnsigned("rotate_minute", rollMinute);

configuration->getUnsigned("chunk_size", chunkSize);

if (configuration->getString("rotate_on_reopen", tmp)) {

if (0 == tmp.compare("yes")) {

rotateOnReopen = true;

} else {

rotateOnReopen = false;

}

}

}

void FileStoreBase::copyCommon(const FileStoreBase *base) {

subDirectory = base->subDirectory;

chunkSize = base->chunkSize;

maxSize = base->maxSize;

maxWriteSize = base->maxWriteSize;

rollPeriod = base->rollPeriod;

rollPeriodLength = base->rollPeriodLength;

rollHour = base->rollHour;

rollMinute = base->rollMinute;

fsType = base->fsType;

writeMeta = base->writeMeta;

writeCategory = base->writeCategory;

createSymlink = base->createSymlink;

baseSymlinkName = base->baseSymlinkName;

writeStats = base->writeStats;

rotateOnReopen = base->rotateOnReopen;

/*

* append the category name to the base file path and change the

* baseFileName to the category name. these are arbitrary, could be anything

* unique

*/

baseFilePath = base->baseFilePath + std::string("/") + categoryHandled;

filePath = baseFilePath;

if (!subDirectory.empty()) {

filePath += "/" + subDirectory;

}

baseFileName = categoryHandled;

}

bool FileStoreBase::open() {

return openInternal(rotateOnReopen, NULL);

}

// Decides whether conditions are sufficient for us to roll files

void FileStoreBase::periodicCheck() {

time_t rawtime = time(NULL);

struct tm timeinfo;

localtime_r(&rawtime, &timeinfo);

// Roll the file if we're over max size, or an hour or day has passed

bool rotate = ((currentSize > maxSize) && (maxSize != 0));

if (!rotate) {

switch (rollPeriod) {

case ROLL_DAILY:

rotate = timeinfo.tm_mday != lastRollTime &&

static_cast<uint>(timeinfo.tm_hour) >= rollHour &&

static_cast<uint>(timeinfo.tm_min) >= rollMinute;

break;

case ROLL_HOURLY:

rotate = timeinfo.tm_hour != lastRollTime &&

static_cast<uint>(timeinfo.tm_min) >= rollMinute;

break;

case ROLL_OTHER:

rotate = rawtime >= lastRollTime + rollPeriodLength;

break;

case ROLL_NEVER:

break;

}

}

if (rotate) {

rotateFile(rawtime);

}

}

void FileStoreBase::rotateFile(time_t currentTime) {

struct tm timeinfo;

currentTime = currentTime > 0 ? currentTime : time(NULL);

localtime_r(&currentTime, &timeinfo);

LOG_OPER("[%s] %d:%d rotating file <%s> old size <%lu> max size <%lu>",

categoryHandled.c_str(), timeinfo.tm_hour, timeinfo.tm_min,

makeBaseFilename(&timeinfo).c_str(), currentSize,

maxSize == ULONG_MAX ? 0 : maxSize);

printStats();

openInternal(true, &timeinfo);

}

string FileStoreBase::makeFullFilename(int suffix, struct tm* creation_time,

bool use_full_path) {

ostringstream filename;

if (use_full_path) {

filename << filePath << '/';

}

filename << makeBaseFilename(creation_time);

filename << '_' << setw(5) << setfill('0') << suffix;

return filename.str();

}

string FileStoreBase::makeBaseSymlink() {

ostringstream base;

if (!baseSymlinkName.empty()) {

base << baseSymlinkName << "_current";

} else {

base << baseFileName << "_current";

}

return base.str();

}

string FileStoreBase::makeFullSymlink() {

ostringstream filename;

filename << filePath << '/' << makeBaseSymlink();

return filename.str();

}

string FileStoreBase::makeBaseFilename(struct tm* creation_time) {

ostringstream filename;

filename << baseFileName;

if (rollPeriod != ROLL_NEVER) {

filename << '-' << creation_time->tm_year + 1900 << '-'

<< setw(2) << setfill('0') << creation_time->tm_mon + 1 << '-'

<< setw(2) << setfill('0') << creation_time->tm_mday;

}

return filename.str();

}

// returns the suffix of the newest file matching base_filename

int FileStoreBase::findNewestFile(const string& base_filename) {

std::vector<std::string> files = FileInterface::list(filePath, fsType);

int max_suffix = -1;

std::string retval;

for (std::vector<std::string>::iterator iter = files.begin();

iter != files.end();

++iter) {

int suffix = getFileSuffix(*iter, base_filename);

if (suffix > max_suffix) {

max_suffix = suffix;

}

}

return max_suffix;

}

int FileStoreBase::findOldestFile(const string& base_filename) {

std::vector<std::string> files = FileInterface::list(filePath, fsType);

int min_suffix = -1;

std::string retval;

for (std::vector<std::string>::iterator iter = files.begin();

iter != files.end();

++iter) {

int suffix = getFileSuffix(*iter, base_filename);

if (suffix >= 0 &&

(min_suffix == -1 || suffix < min_suffix)) {

min_suffix = suffix;

}

}

return min_suffix;

}

int FileStoreBase::getFileSuffix(const string& filename,

const string& base_filename) {

int suffix = -1;

string::size_type suffix_pos = filename.rfind('_');

bool retVal = (0 == filename.substr(0, suffix_pos).compare(base_filename));

if (string::npos != suffix_pos &&

filename.length() > suffix_pos &&

retVal) {

stringstream stream;

stream << filename.substr(suffix_pos + 1);

stream >> suffix;

}

return suffix;

}

void FileStoreBase::printStats() {

if (!writeStats) {

return;

}

string filename(filePath);

filename += "/scribe_stats";

boost::shared_ptr<FileInterface> stats_file =

FileInterface::createFileInterface(fsType, filename);

if (!stats_file ||

!stats_file->createDirectory(filePath) ||

!stats_file->openWrite()) {

LOG_OPER("[%s] Failed to open stats file <%s> of type <%s> for writing",

categoryHandled.c_str(), filename.c_str(), fsType.c_str());

// This isn't enough of a problem to change our status

return;

}

time_t rawtime = time(NULL);

struct tm timeinfo;

localtime_r(&rawtime, &timeinfo);

ostringstream msg;

msg << timeinfo.tm_year + 1900 << '-'

<< setw(2) << setfill('0') << timeinfo.tm_mon + 1 << '-'

<< setw(2) << setfill('0') << timeinfo.tm_mday << '-'

<< setw(2) << setfill('0') << timeinfo.tm_hour << ':'

<< setw(2) << setfill('0') << timeinfo.tm_min;

msg << " wrote <" << currentSize << "> bytes in <" << eventsWritten

<< "> events to file <" << currentFilename << ">" << endl;

stats_file->write(msg.str());

stats_file->close();

}

// Returns the number of bytes to pad to align to the specified chunk size

unsigned long FileStoreBase::bytesToPad(unsigned long next_message_length,

unsigned long current_file_size,

unsigned long chunk_size) {

if (chunk_size > 0) {

unsigned long space_left_in_chunk =

chunk_size - current_file_size % chunk_size;

if (next_message_length > space_left_in_chunk) {

return space_left_in_chunk;

} else {

return 0;

}

}

// chunk_size <= 0 means don't do any chunking

return 0;

}

// set subDirectory to the name of this machine

void FileStoreBase::setHostNameSubDir() {

if (!subDirectory.empty()) {

string error_msg = "WARNING: Bad config - ";

error_msg += "use_hostname_sub_directory will override sub_directory path";

LOG_OPER("[%s] %s", categoryHandled.c_str(), error_msg.c_str());

}

char hostname[255];

int error = gethostname(hostname, sizeof(hostname));

if (error) {

LOG_OPER("[%s] WARNING: gethostname returned error: %d ",

categoryHandled.c_str(), error);

}

string hoststring(hostname);

if (hoststring.empty()) {

LOG_OPER("[%s] WARNING: could not get host name",

categoryHandled.c_str());

} else {

subDirectory = hoststring;

}

}

FileStore::FileStore(StoreQueue* storeq,

const string& category,

bool multi_category, bool is_buffer_file)

: FileStoreBase(storeq, category, "file", multi_category),

isBufferFile(is_buffer_file),

addNewlines(false),

lostBytes_(0) {

}

FileStore::~FileStore() {

}

void FileStore::configure(pStoreConf configuration, pStoreConf parent) {

FileStoreBase::configure(configuration, parent);

// We can run using defaults for all of these, but there are

// a couple of suspicious things we warn about.

if (isBufferFile) {

// scheduled file rotations of buffer files lead to too many messy cases

rollPeriod = ROLL_NEVER;

// Chunks don't work with the buffer file. There's no good reason

// for this, it's just that the FileStore handles chunk padding and

// the FileInterface handles framing, and you need to look at both to

// read a file that's both chunked and framed. The buffer file has

// to be framed, so we don't allow it to be chunked.

// (framed means we write a message size to disk before the message

// data, which allows us to identify separate messages in binary data.

// Chunked means we pad with zeroes to ensure that every multiple

// of n bytes is the start of a message, which helps in recovering

// corrupted binary data and seeking into large files)

chunkSize = 0;

// Combine all categories in a single file for buffers

if (multiCategory) {

writeCategory = true;

}

}

unsigned long inttemp = 0;

configuration->getUnsigned("add_newlines", inttemp);

addNewlines = inttemp ? true : false;

}

bool FileStore::openInternal(bool incrementFilename, struct tm* current_time) {

bool success = false;

struct tm timeinfo;

if (!current_time) {

time_t rawtime = time(NULL);

localtime_r(&rawtime, &timeinfo);

current_time = &timeinfo;

}

try {

int suffix = findNewestFile(makeBaseFilename(current_time));

if (incrementFilename) {

++suffix;

}

// this is the case where there's no file there and we're not incrementing

if (suffix < 0) {

if (rollPeriod == ROLL_HOURLY) {

suffix = current_time->tm_hour;

}

else {

suffix = 0;

}

}

string file = makeFullFilename(suffix, current_time);

switch (rollPeriod) {

case ROLL_DAILY:

lastRollTime = current_time->tm_mday;

break;

case ROLL_HOURLY:

lastRollTime = current_time->tm_hour;

break;

case ROLL_OTHER:

lastRollTime = time(NULL);

break;

case ROLL_NEVER:

break;

}

if (writeFile) {

if (writeMeta) {

writeFile->write(meta_logfile_prefix + file);

if (addNewlines) {

writeFile->write("\n");

}

}

writeFile->close();

}

writeFile = FileInterface::createFileInterface(fsType, file, isBufferFile);

if (!writeFile) {

LOG_OPER("[%s] Failed to create file <%s> of type <%s> for writing",

categoryHandled.c_str(), file.c_str(), fsType.c_str());

setStatus("file open error");

return false;

}

success = writeFile->createDirectory(baseFilePath);

// If we created a subdirectory, we need to create two directories

if (success && !subDirectory.empty()) {

success = writeFile->createDirectory(filePath);

}

if (!success) {

LOG_OPER("[%s] Failed to create directory for file <%s>",

categoryHandled.c_str(), file.c_str());

setStatus("File open error");

return false;

}

success = writeFile->openWrite();

if (!success) {

LOG_OPER("[%s] Failed to open file <%s> for writing",

categoryHandled.c_str(),

file.c_str());

setStatus("File open error");

} else {

/* just make a best effort here, and don't error if it fails */

if (createSymlink && !isBufferFile) {

string symlinkName = makeFullSymlink();

boost::shared_ptr<FileInterface> tmp =

FileInterface::createFileInterface(fsType, symlinkName, isBufferFile);

tmp->deleteFile();

string symtarget = makeFullFilename(suffix, current_time, false);

writeFile->createSymlink(symtarget, symlinkName);

}

// else it confuses the filename code on reads

LOG_OPER("[%s] Opened file <%s> for writing", categoryHandled.c_str(),

file.c_str());

currentSize = writeFile->fileSize();

currentFilename = file;

eventsWritten = 0;

setStatus("");

}

} catch(const std::exception& e) {

LOG_OPER("[%s] Failed to create/open file of type <%s> for writing",

categoryHandled.c_str(), fsType.c_str());

LOG_OPER("Exception: %s", e.what());

setStatus("file create/open error");

return false;

}

return success;

}

bool FileStore::isOpen() {

return writeFile && writeFile->isOpen();

}

void FileStore::close() {

if (writeFile) {

writeFile->close();

}

}

void FileStore::flush() {

if (writeFile) {

writeFile->flush();

}

}

shared_ptr<Store> FileStore::copy(const std::string &category) {

FileStore *store = new FileStore(storeQueue, category, multiCategory,

isBufferFile);

shared_ptr<Store> copied = shared_ptr<Store>(store);

store->addNewlines = addNewlines;

store->copyCommon(this);

return copied;

}

bool FileStore::handleMessages(boost::shared_ptr<logentry_vector_t> messages) {

if (!isOpen()) {

if (!open()) {

LOG_OPER("[%s] File failed to open FileStore::handleMessages()",

categoryHandled.c_str());

return false;

}

}

// write messages to current file

return writeMessages(messages);

}

// writes messages to either the specified file or the the current writeFile

bool FileStore::writeMessages(boost::shared_ptr<logentry_vector_t> messages,

boost::shared_ptr<FileInterface> file) {

// Data is written to a buffer first, then sent to disk in one call to write.

// This costs an extra copy of the data, but dramatically improves latency with

// network based files. (nfs, etc)

string write_buffer;

bool success = true;

unsigned long current_size_buffered = 0; // size of data in write_buffer

unsigned long num_buffered = 0;

unsigned long num_written = 0;

boost::shared_ptr<FileInterface> write_file;

unsigned long max_write_size = min(maxSize, maxWriteSize);

// if no file given, use current writeFile

if (file) {

write_file = file;

} else {

write_file = writeFile;

}

try {

for (logentry_vector_t::iterator iter = messages->begin();

iter != messages->end();

++iter) {

// have to be careful with the length here. getFrame wants the length without

// the frame, then bytesToPad wants the length of the frame and the message.

unsigned long length = 0;

unsigned long message_length = (*iter)->message.length();

string frame, category_frame;

if (addNewlines) {

++message_length;

}

length += message_length;

if (writeCategory) {

//add space for category+newline and category frame

unsigned long category_length = (*iter)->category.length() + 1;

length += category_length;

category_frame = write_file->getFrame(category_length);

length += category_frame.length();

}

// frame is a header that the underlying file class can add to each message

frame = write_file->getFrame(message_length);

length += frame.length();

// padding to align messages on chunk boundaries

unsigned long padding = bytesToPad(length, current_size_buffered, chunkSize);

length += padding;

if (padding) {

write_buffer += string(padding, 0);

}

if (writeCategory) {

write_buffer += category_frame;

write_buffer += (*iter)->category + "\n";

}

write_buffer += frame;

write_buffer += (*iter)->message;

if (addNewlines) {

write_buffer += "\n";

}

current_size_buffered += length;

num_buffered++;

// Write buffer if processing last message or if larger than allowed

if ((current_size_buffered > max_write_size && maxSize != 0) ||

messages->end() == iter + 1 ) {

if (!write_file->write(write_buffer)) {

LOG_OPER("[%s] File store failed to write (%lu) messages to file",

categoryHandled.c_str(), messages->size());

setStatus("File write error");

success = false;

break;

}

num_written += num_buffered;

currentSize += current_size_buffered;

num_buffered = 0;

current_size_buffered = 0;

write_buffer = "";

}

// rotate file if large enough and not writing to a separate file

if ((currentSize > maxSize && maxSize != 0 )&& !file) {

rotateFile();

write_file = writeFile;

}

}

} catch (const std::exception& e) {

LOG_OPER("[%s] File store failed to write. Exception: %s",

categoryHandled.c_str(), e.what());

success = false;

}

eventsWritten += num_written;

if (!success) {

close();

// update messages to include only the messages that were not handled

if (num_written > 0) {

messages->erase(messages->begin(), messages->begin() + num_written);

}

}

return success;

}

// Deletes the oldest file

// currently gets invoked from within a bufferstore

void FileStore::deleteOldest(struct tm* now) {

int index = findOldestFile(makeBaseFilename(now));

if (index < 0) {

return;

}

shared_ptr<FileInterface> deletefile = FileInterface::createFileInterface(fsType,

makeFullFilename(index, now));

if (lostBytes_) {

g_Handler->incCounter(categoryHandled, "bytes lost", lostBytes_);

lostBytes_ = 0;

}

deletefile->deleteFile();

}

// Replace the messages in the oldest file at this timestamp with the input messages

bool FileStore::replaceOldest(boost::shared_ptr<logentry_vector_t> messages,

struct tm* now) {

string base_name = makeBaseFilename(now);

int index = findOldestFile(base_name);

if (index < 0) {

LOG_OPER("[%s] Could not find files <%s>", categoryHandled.c_str(), base_name.c_str());

return false;

}

string filename = makeFullFilename(index, now);

// Need to close and reopen store in case we already have this file open

close();

shared_ptr<FileInterface> infile = FileInterface::createFileInterface(fsType,

filename, isBufferFile);

// overwrite the old contents of the file

bool success;

if (infile->openTruncate()) {

success = writeMessages(messages, infile);

} else {

LOG_OPER("[%s] Failed to open file <%s> for writing and truncate",

categoryHandled.c_str(), filename.c_str());

success = false;

}

// close this file and re-open store

infile->close();

open();

return success;

}

bool FileStore::readOldest(/*out*/ boost::shared_ptr<logentry_vector_t> messages,

struct tm* now) {

long loss;

int index = findOldestFile(makeBaseFilename(now));

if (index < 0) {

// This isn't an error. It's legit to call readOldest when there aren't any

// files left, in which case the call succeeds but returns messages empty.

return true;

}

std::string filename = makeFullFilename(index, now);

shared_ptr<FileInterface> infile = FileInterface::createFileInterface(fsType,

filename, isBufferFile);

if (!infile->openRead()) {

LOG_OPER("[%s] Failed to open file <%s> for reading",

categoryHandled.c_str(), filename.c_str());

return false;

}

uint32_t bsize = 0;

std::string message;

while ((loss = infile->readNext(message)) > 0) {

if (!message.empty()) {

logentry_ptr_t entry = logentry_ptr_t(new LogEntry);

// check whether a category is stored with the message

if (writeCategory) {

// get category without trailing \n

entry->category = message.substr(0, message.length() - 1);

if ((loss = infile->readNext(message)) <= 0) {

LOG_OPER("[%s] category not stored with message <%s> "

"corruption?, incompatible config change?",

categoryHandled.c_str(), entry->category.c_str());