diff options
| -rw-r--r-- | ChangeLog | 6 | ||||
| -rw-r--r-- | plugins/imfile/imfile.c | 1 | ||||
| -rw-r--r-- | runtime/stream.c | 103 | ||||
| -rw-r--r-- | runtime/stream.h | 5 |
4 files changed, 90 insertions, 25 deletions
@@ -1,5 +1,11 @@ --------------------------------------------------------------------------- Version 7.3.9 [devel] 2013-03-?? +- imfile: now detects file change when rsyslog was inactive + Previosly, this case could not be detected, so if a file was overwritten + or rotated away while rsyslog was stopped, some data was missing. This + is now detected and the new file being forwarded right from the + beginning. + closes: http://bugzilla.adiscon.com/show_bug.cgi?id=228 - bugfix: build problems on non-Linux platforms - bugfix: stdout/stderr were not closed on forking but were closed when running in the forground - this was just reversed diff --git a/plugins/imfile/imfile.c b/plugins/imfile/imfile.c index 0f155c10..6ce08fd4 100644 --- a/plugins/imfile/imfile.c +++ b/plugins/imfile/imfile.c @@ -235,6 +235,7 @@ openFile(fileInfo_t *pThis) /* read back in the object */ CHKiRet(obj.Deserialize(&pThis->pStrm, (uchar*) "strm", psSF, NULL, pThis)); + strm.CheckFileChange(pThis->pStrm); CHKiRet(strm.SeekCurrOffs(pThis->pStrm)); /* note: we do not delete the state file, so that the last position remains diff --git a/runtime/stream.c b/runtime/stream.c index 3e890c71..00afcdaa 100644 --- a/runtime/stream.c +++ b/runtime/stream.c @@ -81,6 +81,7 @@ static void *asyncWriterThread(void *pPtr); static rsRetVal doZipWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf, int bFlush); static rsRetVal doZipFinish(strm_t *pThis); static rsRetVal strmPhysWrite(strm_t *pThis, uchar *pBuf, size_t lenBuf); +static rsRetVal strmSeekCurrOffs(strm_t *pThis); /* methods */ @@ -197,6 +198,7 @@ static rsRetVal doPhysOpen(strm_t *pThis) { int iFlags = 0; + struct stat statOpen; DEFiRet; ISOBJ_TYPE_assert(pThis, strm); @@ -234,15 +236,71 @@ doPhysOpen(strm_t *pThis) ABORT_FINALIZE(RS_RET_FILE_NOT_FOUND); else ABORT_FINALIZE(RS_RET_IO_ERROR); + } + + if(pThis->tOperationsMode == STREAMMODE_READ) { + if(fstat(pThis->fd, &statOpen) == -1) { + DBGPRINTF("Error: cannot obtain inode# for file %s\n", pThis->pszCurrFName); + ABORT_FINALIZE(RS_RET_IO_ERROR); + } + pThis->inode = statOpen.st_ino; + } + + if(!ustrcmp(pThis->pszCurrFName, UCHAR_CONSTANT(_PATH_CONSOLE)) || isatty(pThis->fd)) { + DBGPRINTF("file %d is a tty-type file\n", pThis->fd); + pThis->bIsTTY = 1; } else { - if(!ustrcmp(pThis->pszCurrFName, UCHAR_CONSTANT(_PATH_CONSOLE)) || isatty(pThis->fd)) { - DBGPRINTF("file %d is a tty-type file\n", pThis->fd); - pThis->bIsTTY = 1; + pThis->bIsTTY = 0; + } + +finalize_it: + RETiRet; +} + + +static rsRetVal +strmSetCurrFName(strm_t *pThis) +{ + DEFiRet; + + if(pThis->sType == STREAMTYPE_FILE_CIRCULAR) { + CHKiRet(genFileName(&pThis->pszCurrFName, pThis->pszDir, pThis->lenDir, + pThis->pszFName, pThis->lenFName, pThis->iCurrFNum, pThis->iFileNumDigits)); + } else { + if(pThis->pszDir == NULL) { + if((pThis->pszCurrFName = ustrdup(pThis->pszFName)) == NULL) + ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); } else { - pThis->bIsTTY = 0; + CHKiRet(genFileName(&pThis->pszCurrFName, pThis->pszDir, pThis->lenDir, + pThis->pszFName, pThis->lenFName, -1, 0)); } } +finalize_it: + RETiRet; +} + +/* This function checks if the actual file has changed and, if so, resets the + * offset. This is support for monitoring files. It should be called after + * deserializing the strm object and before doing any other operation on it + * (most importantly not an open or seek!). + */ +static rsRetVal +CheckFileChange(strm_t *pThis) +{ + struct stat statName; + DEFiRet; + CHKiRet(strmSetCurrFName(pThis)); + if(stat((char*) pThis->pszCurrFName, &statName) == -1) + ABORT_FINALIZE(RS_RET_IO_ERROR); + DBGPRINTF("stream/after deserialize checking for file change on '%s', " + "inode %u/%u, size/currOffs %llu/%llu\n", + pThis->pszCurrFName, (unsigned) pThis->inode, + (unsigned) statName.st_ino, statName.st_size, pThis->iCurrOffs); + if(pThis->inode != statName.st_ino || statName.st_size < pThis->iCurrOffs) { + DBGPRINTF("stream: file %s has changed\n", pThis->pszCurrFName); + pThis->iCurrOffs = 0; + } finalize_it: RETiRet; } @@ -265,19 +323,8 @@ static rsRetVal strmOpenFile(strm_t *pThis) if(pThis->pszFName == NULL) ABORT_FINALIZE(RS_RET_FILE_PREFIX_MISSING); - if(pThis->sType == STREAMTYPE_FILE_CIRCULAR) { - CHKiRet(genFileName(&pThis->pszCurrFName, pThis->pszDir, pThis->lenDir, - pThis->pszFName, pThis->lenFName, pThis->iCurrFNum, pThis->iFileNumDigits)); - } else { - if(pThis->pszDir == NULL) { - if((pThis->pszCurrFName = ustrdup(pThis->pszFName)) == NULL) - ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); - } else { - CHKiRet(genFileName(&pThis->pszCurrFName, pThis->pszDir, pThis->lenDir, - pThis->pszFName, pThis->lenFName, -1, 0)); - } - } - + CHKiRet(strmSetCurrFName(pThis)); + CHKiRet(doPhysOpen(pThis)); pThis->iCurrOffs = 0; @@ -357,6 +404,7 @@ static rsRetVal strmCloseFile(strm_t *pThis) if(pThis->fd != -1) { close(pThis->fd); pThis->fd = -1; + pThis->inode = 0; } if(pThis->fdDir != -1) { @@ -432,18 +480,15 @@ static rsRetVal strmHandleEOFMonitor(strm_t *pThis) { DEFiRet; - struct stat statOpen; struct stat statName; ISOBJ_TYPE_assert(pThis, strm); - if(fstat(pThis->fd, &statOpen) == -1) - ABORT_FINALIZE(RS_RET_IO_ERROR); if(stat((char*) pThis->pszCurrFName, &statName) == -1) ABORT_FINALIZE(RS_RET_IO_ERROR); - DBGPRINTF("stream checking for file change on '%s', inode %u/%u", - pThis->pszCurrFName, (unsigned) statOpen.st_ino, + DBGPRINTF("stream checking for file change on '%s', inode %u/%u\n", + pThis->pszCurrFName, (unsigned) pThis->inode, (unsigned) statName.st_ino); - if(statOpen.st_ino == statName.st_ino) { + if(pThis->inode == statName.st_ino) { ABORT_FINALIZE(RS_RET_EOF); } else { /* we had a file change! */ @@ -1343,7 +1388,11 @@ static rsRetVal strmSeek(strm_t *pThis, off64_t offs) } long long i; DBGOPRINT((obj_t*) pThis, "file %d seek, pos %llu\n", pThis->fd, (long long unsigned) offs); - i = lseek64(pThis->fd, offs, SEEK_SET); // TODO: check error! + i = lseek64(pThis->fd, offs, SEEK_SET); + if(i != offs) { + DBGPRINTF("strmSeek: error %lld seeking to offset %lld\n", i, offs); + ABORT_FINALIZE(RS_RET_IO_ERROR); + } pThis->iCurrOffs = offs; /* we are now at *this* offset */ pThis->iBufPtr = 0; /* buffer invalidated */ @@ -1697,6 +1746,9 @@ static rsRetVal strmSerialize(strm_t *pThis, strm_t *pStrm) l = pThis->iCurrOffs; objSerializeSCALAR_VAR(pStrm, iCurrOffs, INT64, l); + l = pThis->inode; + objSerializeSCALAR_VAR(pStrm, inode, INT64, l); + objSerializePTR(pStrm, prevLineSegment, PSZ); CHKiRet(obj.EndSerialize(pStrm)); @@ -1796,6 +1848,8 @@ static rsRetVal strmSetProperty(strm_t *pThis, var_t *pProp) CHKiRet(strmSettOpenMode(pThis, pProp->val.num)); } else if(isProp("iCurrOffs")) { pThis->iCurrOffs = pProp->val.num; + } else if(isProp("inode")) { + pThis->inode = (ino_t) pProp->val.num; } else if(isProp("iMaxFileSize")) { CHKiRet(strmSetiMaxFileSize(pThis, pProp->val.num)); } else if(isProp("iMaxFiles")) { @@ -1865,6 +1919,7 @@ CODESTARTobjQueryInterface(strm) pIf->GetCurrOffset = strmGetCurrOffset; pIf->Dup = strmDup; pIf->SetWCntr = strmSetWCntr; + pIf->CheckFileChange = CheckFileChange; /* set methods */ pIf->SetbDeleteOnClose = strmSetbDeleteOnClose; pIf->SetiMaxFileSize = strmSetiMaxFileSize; diff --git a/runtime/stream.h b/runtime/stream.h index b7e74074..b7cc6d36 100644 --- a/runtime/stream.h +++ b/runtime/stream.h @@ -112,6 +112,7 @@ typedef struct strm_s { int lenDir; int fd; /* the file descriptor, -1 if closed */ int fdDir; /* the directory's descriptor, in case bSync is requested (-1 if closed) */ + ino_t inode; /* current inode for files being monitored (undefined else) */ uchar *pszCurrFName; /* name of current file (if open) */ uchar *pIOBuf; /* the iobuffer currently in use to gather data */ size_t iBufPtrMax; /* current max Ptr in Buffer (if partial read!) */ @@ -187,8 +188,10 @@ BEGINinterface(strm) /* name must also be changed in ENDinterface macro! */ rsRetVal (*ReadLine)(strm_t *pThis, cstr_t **ppCStr, int mode); /* v7 added 2012-09-14 */ INTERFACEpropSetMeth(strm, bVeryReliableZip, int); + /* v8 added 2013-03-21 */ + rsRetVal (*CheckFileChange)(strm_t *pThis); ENDinterface(strm) -#define strmCURR_IF_VERSION 7 /* increment whenever you change the interface structure! */ +#define strmCURR_IF_VERSION 8 /* increment whenever you change the interface structure! */ static inline int strmGetCurrFileNum(strm_t *pStrm) { |