<commit_before>//---------------------------------------------------------- -*- Mode: C++ -*-
// $Id$
//
// Created 2010/03/02
// Author: <NAME>
//
// Copyright 2010-2012 Quantcast Corp.
//
// This file is part of Kosmos File System (KFS).
//
// 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.
//
// Double buffered message log writer implementation.
//
//----------------------------------------------------------------------------

#include "BufferedLogWriter.h"

#include <fcntl.h>
#include <unistd.h>
#include <poll.h>
#include <dirent.h>
#include <stdio.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <inttypes.h>
#include <stdarg.h>
#include <errno.h>
#include <string.h>

#include <vector>
#include <string>
#include <sstream>
#include <iomanip>
#include <iostream>

#include "qcdio/QCMutex.h"
#include "qcdio/QCUtils.h"
#include "qcdio/qcstutils.h"
#include "qcdio/qcdebug.h"
#include "qcdio/QCThread.h"
#include "Properties.h"

namespace KFS
{

using std::string;
using std::max;
using std::min;
using std::ostringstream;
using std::ostream;
using std::streambuf;
using std::setw;
using std::setfill;
using std::streamsize;
using std::vector;

const char* const kBufferedLogWriter_LogLevels[] = {
    "FATAL",
    "ALERT",
    "CRIT",
    "ERROR",
    "WARN",
    "NOTICE",
    "INFO",
    "DEBUG",
    "NOTSET"
};

const int64_t kLogWriterMinLogFileSize               = 16 << 10;
const int64_t kLogWriterMinOpenRetryIntervalMicroSec = 10000;
const int     kLogWriterMinLogBufferSize             = 16 << 10;
const int64_t kLogWirterDefaultTimeToKeepSecs        = 60 * 60 * 24 * 30;
const int     kLogWriterDefaulOpenFlags              =
    O_CREAT | O_APPEND | O_WRONLY /* | O_SYNC */;

class BufferedLogWriter::Impl : public QCRunnable
{
public:
    Impl(
        int         inFd,
        const char* inFileNamePtr,
        int         inBufSize,
        const char* inTrucatedSuffixPtr,
        int64_t     inOpenRetryIntervalMicroSec,
        int64_t     inFlushIntervalMicroSec,
        int64_t     inMaxLogFileSize,
        int         inMaxLogsFiles,
        int64_t     inMaxLogWaitTimeMicroSec,
        const char* inTimeStampFormatPtr,
        bool        inUseGMTFlag)
        : mMutex(),
          mWriteCond(),
          mWriteDoneCond(),
          mThread(0, "LogWriter"),
          mLogFileNamePrefixes(),
          mFileName(inFileNamePtr ? inFileNamePtr : ""),
          mTruncatedSuffix(inTrucatedSuffixPtr ? inTrucatedSuffixPtr : "..."),
          mTimeStampFormat(inTimeStampFormatPtr ?
            inTimeStampFormatPtr : "%m-%d-%Y %H:%M:%S"),
          mNewLogSuffix(),
          mDroppedCount(0),
          mTotalDroppedCount(0),
          mTruncatedCount(0),
          mBufWaitersCount(0),
          mWriteErrCount(0),
          mWriteBytesDiscardedCount(0),
          mCurWritten(0),
          mBufWaitedCount(0),
          mMaxLogFileSize(inMaxLogFileSize <= 0 ? inMaxLogFileSize :
            max(kLogWriterMinLogFileSize, inMaxLogFileSize)),
          mMaxLogFiles(inMaxLogFileSize > 0 ? inMaxLogsFiles : -1),
          mNextOpenRetryTime(Now() - Seconds(10)),
          mOpenRetryInterval(max(kLogWriterMinOpenRetryIntervalMicroSec,
            inOpenRetryIntervalMicroSec)),
          mFlushInterval(max((int64_t)0, inFlushIntervalMicroSec)),
          mNextFlushTime(-1),
          mTotalLogWaitedTime(0),
          mCurLogWatedTime(0),
          mMaxLogWaitTime(inMaxLogWaitTimeMicroSec),
          mNextDeleteOldLogsTimeSec(0),
          mMinModTimeToKeepSec(kLogWirterDefaultTimeToKeepSecs),
          mOpenFlags(kLogWriterDefaulOpenFlags),
          mOpenMode(0644),
          mBufSize(max(inBufSize > 0 ? inBufSize : (1 << 20),
            (int)mTruncatedSuffix.size() + kLogWriterMinLogBufferSize)),
          mLastError(0),
          mFd(inFd >= 0 ? dup(inFd) : inFd),
          mMaxAppendLength(mBufSize - mTruncatedSuffix.size() + 1),
          mRunFlag(false),
          mDeleteOldLogsFlag(false),
          mBuf0Ptr(new char [mBufSize * 2]),
          mBuf1Ptr(mBuf0Ptr + mBufSize),
          mBufPtr(mBuf0Ptr),
          mCurPtr(mBufPtr),
          mEndPtr(mBuf0Ptr + mBufSize),
          mWritePtr(0),
          mWriteEndPtr(0),
          mCloseFlag(false),
          mUseGMTFlag(inUseGMTFlag),
          mTimeSec(0),
          mLogTimeStampSec(0),
          mMsgAppendCount(0),
          mRotateLogsDailyFlag(true),
          mMaxMsgStreamCount(256),
          mMsgStreamCount(0),
          mMsgStreamHeadPtr(0)
    {
        if (! mFileName.empty()) {
            mLogFileNamePrefixes.push_back(mFileName);
        }
        mLogTimeStampPrefixStr[0] = 0;
        int64_t theSec      = 0;
        int64_t theMicroSec = 0;
        Now(theSec, theMicroSec);
        mTimeSec         = theSec - 1;
        mLogTimeStampSec = mTimeSec;
        UpdateTimeTm(theSec);
        mLastLogTm       = mTimeTm;
        GetLogTimeStampPrefixPtr(theSec);
        mNextFlushTime = Seconds(theSec) + theMicroSec + mFlushInterval;
        if (mFd >= 0) {
            ::fcntl(mFd, F_SETFL, O_NONBLOCK);
        }
    }
    static const char* GetLogLevelNamePtr(
        LogLevel inLogLevel)
    {
        const int theLogLevel = inLogLevel / 100;
        return (<commit_msg>Common lib: buffered log writer do not set non blocking io if file descriptor is a terminal.<commit_after>//---------------------------------------------------------- -*- Mode: C++ -*-
// $Id$
//
// Created 2010/03/02
// Author: <NAME>
//
// Copyright 2010-2012 Quantcast Corp.
//
// This file is part of Kosmos File System (KFS).
//
// 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.
//
// Double buffered message log writer implementation.
//
//----------------------------------------------------------------------------

#include "BufferedLogWriter.h"
#include "Properties.h"

#include "qcdio/QCMutex.h"
#include "qcdio/QCUtils.h"
#include "qcdio/qcstutils.h"
#include "qcdio/qcdebug.h"
#include "qcdio/QCThread.h"

#include <fcntl.h>
#include <unistd.h>
#include <poll.h>
#include <dirent.h>
#include <stdio.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/time.h>
#include <inttypes.h>
#include <stdarg.h>
#include <errno.h>
#include <string.h>

#include <vector>
#include <string>
#include <sstream>
#include <iomanip>
#include <iostream>

namespace KFS
{

using std::string;
using std::max;
using std::min;
using std::ostringstream;
using std::ostream;
using std::streambuf;
using std::setw;
using std::setfill;
using std::streamsize;
using std::vector;

const char* const kBufferedLogWriter_LogLevels[] = {
    "FATAL",
    "ALERT",
    "CRIT",
    "ERROR",
    "WARN",
    "NOTICE",
    "INFO",
    "DEBUG",
    "NOTSET"
};

const int64_t kLogWriterMinLogFileSize               = 16 << 10;
const int64_t kLogWriterMinOpenRetryIntervalMicroSec = 10000;
const int     kLogWriterMinLogBufferSize             = 16 << 10;
const int64_t kLogWirterDefaultTimeToKeepSecs        = 60 * 60 * 24 * 30;
const int     kLogWriterDefaulOpenFlags              =
    O_CREAT | O_APPEND | O_WRONLY /* | O_SYNC */;

class BufferedLogWriter::Impl : public QCRunnable
{
public:
    Impl(
        int         inFd,
        const char* inFileNamePtr,
        int         inBufSize,
        const char* inTrucatedSuffixPtr,
        int64_t     inOpenRetryIntervalMicroSec,
        int64_t     inFlushIntervalMicroSec,
        int64_t     inMaxLogFileSize,
        int         inMaxLogsFiles,
        int64_t     inMaxLogWaitTimeMicroSec,
        const char* inTimeStampFormatPtr,
        bool        inUseGMTFlag)
        : mMutex(),
          mWriteCond(),
          mWriteDoneCond(),
          mThread(0, "LogWriter"),
          mLogFileNamePrefixes(),
          mFileName(inFileNamePtr ? inFileNamePtr : ""),
          mTruncatedSuffix(inTrucatedSuffixPtr ? inTrucatedSuffixPtr : "..."),
          mTimeStampFormat(inTimeStampFormatPtr ?
            inTimeStampFormatPtr : "%m-%d-%Y %H:%M:%S"),
          mNewLogSuffix(),
          mDroppedCount(0),
          mTotalDroppedCount(0),
          mTruncatedCount(0),
          mBufWaitersCount(0),
          mWriteErrCount(0),
          mWriteBytesDiscardedCount(0),
          mCurWritten(0),
          mBufWaitedCount(0),
          mMaxLogFileSize(inMaxLogFileSize <= 0 ? inMaxLogFileSize :
            max(kLogWriterMinLogFileSize, inMaxLogFileSize)),
          mMaxLogFiles(inMaxLogFileSize > 0 ? inMaxLogsFiles : -1),
          mNextOpenRetryTime(Now() - Seconds(10)),
          mOpenRetryInterval(max(kLogWriterMinOpenRetryIntervalMicroSec,
            inOpenRetryIntervalMicroSec)),
          mFlushInterval(max((int64_t)0, inFlushIntervalMicroSec)),
          mNextFlushTime(-1),
          mTotalLogWaitedTime(0),
          mCurLogWatedTime(0),
          mMaxLogWaitTime(inMaxLogWaitTimeMicroSec),
          mNextDeleteOldLogsTimeSec(0),
          mMinModTimeToKeepSec(kLogWirterDefaultTimeToKeepSecs),
          mOpenFlags(kLogWriterDefaulOpenFlags),
          mOpenMode(0644),
          mBufSize(max(inBufSize > 0 ? inBufSize : (1 << 20),
            (int)mTruncatedSuffix.size() + kLogWriterMinLogBufferSize)),
          mLastError(0),
          mFd(inFd >= 0 ? dup(inFd) : inFd),
          mMaxAppendLength(mBufSize - mTruncatedSuffix.size() + 1),
          mRunFlag(false),
          mDeleteOldLogsFlag(false),
          mBuf0Ptr(new char [mBufSize * 2]),
          mBuf1Ptr(mBuf0Ptr + mBufSize),
          mBufPtr(mBuf0Ptr),
          mCurPtr(mBufPtr),
          mEndPtr(mBuf0Ptr + mBufSize),
          mWritePtr(0),
          mWriteEndPtr(0),
          mCloseFlag(false),
          mUseGMTFlag(inUseGMTFlag),
          mTimeSec(0),
          mLogTimeStampSec(0),
          mMsgAppendCount(0),
          mRotateLogsDailyFlag(true),
          mMaxMsgStreamCount(256),
          mMsgStreamCount(0),
          mMsgStreamHeadPtr(0)
    {
        if (! mFileName.empty()) {
            mLogFileNamePrefixes.push_back(mFileName);
        }
        mLogTimeStampPrefixStr[0] = 0;
        int64_t theSec      = 0;
        int64_t theMicroSec = 0;
        Now(theSec, theMicroSec);
        mTimeSec         = theSec - 1;
        mLogTimeStampSec = mTimeSec;
        UpdateTimeTm(theSec);
        mLastLogTm       = mTimeTm;
        GetLogTimeStampPrefixPtr(theSec);
        mNextFlushTime = Seconds(theSec) + theMicroSec + mFlushInterval;
        if (mFd >= 0 && ! isatty(mFd)) {
            ::fcntl(mFd, F_SETFL, O_NONBLOCK);
        }
    }
    static const char* GetLogLevelNamePtr(
        LogLevel inLogLevel)
    {
        const int theLogLevel = inLogLevel / 100;
        return (
{'fraction_non_alphanumeric': 0.12959765152823347, 'fraction_numerical': 0.020031082714557072, 'mean_word_length': 2.337078651685393, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 47, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3811953', 'n_tokens_mistral': 3898, 'n_tokens_neox': 3735, 'n_words': 767}