summaryrefslogtreecommitdiffstats
path: root/plugins/omhdfs/omhdfs.c
diff options
context:
space:
mode:
authorRainer Gerhards <rgerhards@adiscon.com>2011-03-09 17:24:03 +0100
committerRainer Gerhards <rgerhards@adiscon.com>2011-03-09 17:24:03 +0100
commitfe52af4d409a30d50e66f870b9eb503bee47cbfb (patch)
tree4c87afc2c9d43542b874334520f33ccc5968c2a1 /plugins/omhdfs/omhdfs.c
parent53afa44de8d7796df46a12d66fd9866165779d94 (diff)
downloadrsyslog-fe52af4d409a30d50e66f870b9eb503bee47cbfb.tar.gz
rsyslog-fe52af4d409a30d50e66f870b9eb503bee47cbfb.tar.xz
rsyslog-fe52af4d409a30d50e66f870b9eb503bee47cbfb.zip
enhanced omhdfs to support batching mode. This permits to increase performance,
as we now call the HDFS API with much larger message sizes and far more infrequently
Diffstat (limited to 'plugins/omhdfs/omhdfs.c')
-rw-r--r--plugins/omhdfs/omhdfs.c85
1 files changed, 73 insertions, 12 deletions
diff --git a/plugins/omhdfs/omhdfs.c b/plugins/omhdfs/omhdfs.c
index 8b72747f..76128a4e 100644
--- a/plugins/omhdfs/omhdfs.c
+++ b/plugins/omhdfs/omhdfs.c
@@ -80,6 +80,8 @@ typedef struct {
typedef struct _instanceData {
file_t *pFile;
+ uchar ioBuf[64*1024];
+ unsigned offsBuf;
} instanceData;
/* forward definitions (down here, need data types) */
@@ -260,7 +262,8 @@ fileOpen(file_t *pFile)
if(errno == ENOENT) {
DBGPRINTF("omhdfs: ENOENT trying to append to '%s', now trying create\n",
pFile->name);
- pFile->fh = hdfsOpenFile(pFile->fs, (char*)pFile->name, O_WRONLY|O_CREAT, 0, 0, 0);
+ pFile->fh = hdfsOpenFile(pFile->fs,
+ (char*)pFile->name, O_WRONLY|O_CREAT, 0, 0, 0);
}
}
if(pFile->fh == NULL) {
@@ -275,12 +278,15 @@ finalize_it:
}
+/* Note: lenWrite is reset to zero on successful write! */
static inline rsRetVal
-fileWrite(file_t *pFile, uchar *buf)
+fileWrite(file_t *pFile, uchar *buf, size_t *lenWrite)
{
- size_t lenWrite;
DEFiRet;
+ if(*lenWrite == 0)
+ FINALIZE;
+
if(pFile->nUsers > 1)
d_pthread_mutex_lock(&pFile->mut);
@@ -294,18 +300,18 @@ fileWrite(file_t *pFile, uchar *buf)
}
}
- lenWrite = strlen((char*) buf);
- tSize num_written_bytes = hdfsWrite(pFile->fs, pFile->fh, buf, lenWrite);
- if((unsigned) num_written_bytes != lenWrite) {
- errmsg.LogError(errno, RS_RET_ERR_HDFS_WRITE, "omhdfs: failed to write %s, expected %lu bytes, "
- "written %lu\n", pFile->name, (unsigned long) lenWrite,
+dbgprintf("XXXXX: omhdfs writing %u bytes\n", *lenWrite);
+ tSize num_written_bytes = hdfsWrite(pFile->fs, pFile->fh, buf, *lenWrite);
+ if((unsigned) num_written_bytes != *lenWrite) {
+ errmsg.LogError(errno, RS_RET_ERR_HDFS_WRITE,
+ "omhdfs: failed to write %s, expected %lu bytes, "
+ "written %lu\n", pFile->name, (unsigned long) *lenWrite,
(unsigned long) num_written_bytes);
ABORT_FINALIZE(RS_RET_SUSPENDED);
}
+ *lenWrite = 0;
finalize_it:
- if(pFile->nUsers > 1)
- d_pthread_mutex_unlock(&pFile->mut);
RETiRet;
}
@@ -333,6 +339,40 @@ finalize_it:
/* ---END FILE OBJECT---------------------------------------------------- */
+/* This adds data to the output buffer and performs an actual write
+ * if the new data does not fit into the buffer. Note that we never write
+ * partial data records. Other actions may write into the same file, and if
+ * we would write partial records, data could become severely mixed up.
+ * Note that we must check of some new data arrived is large than our
+ * buffer. In that case, the new data will written with its own
+ * write operation.
+ */
+static inline rsRetVal
+addData(instanceData *pData, uchar *buf)
+{
+ unsigned len;
+ DEFiRet;
+
+ len = strlen((char*)buf);
+ if(pData->offsBuf + len < sizeof(pData->ioBuf)) {
+ /* new data fits into remaining buffer */
+ memcpy((char*) pData->ioBuf + pData->offsBuf, buf, len);
+ pData->offsBuf += len;
+ } else {
+dbgprintf("XXXXX: not enough room, need to flush\n");
+ CHKiRet(fileWrite(pData->pFile, pData->ioBuf, &pData->offsBuf));
+ if(len >= sizeof(pData->ioBuf)) {
+ CHKiRet(fileWrite(pData->pFile, buf, &len));
+ } else {
+ memcpy((char*) pData->ioBuf + pData->offsBuf, buf, len);
+ pData->offsBuf += len;
+ }
+ }
+
+ iRet = RS_RET_DEFER_COMMIT;
+finalize_it:
+ RETiRet;
+}
BEGINcreateInstance
CODESTARTcreateInstance
@@ -358,13 +398,31 @@ CODESTARTtryResume
}
ENDtryResume
+
+BEGINbeginTransaction
+CODESTARTbeginTransaction
+dbgprintf("omhdfs: beginTransaction\n");
+ENDbeginTransaction
+
+
BEGINdoAction
CODESTARTdoAction
- DBGPRINTF("omuxsock: action to to write to %s\n", pData->pFile->name);
- iRet = fileWrite(pData->pFile, ppString[0]);
+ DBGPRINTF("omhdfs: action to to write to %s\n", pData->pFile->name);
+ iRet = addData(pData, ppString[0]);
+dbgprintf("omhdfs: done doAction\n");
ENDdoAction
+BEGINendTransaction
+CODESTARTendTransaction
+dbgprintf("omhdfs: endTransaction\n");
+ if(pData->offsBuf != 0) {
+ DBGPRINTF("omhdfs: data unwritten at end of transaction, persisting...\n");
+ iRet = fileWrite(pData->pFile, pData->ioBuf, &pData->offsBuf);
+ }
+ENDendTransaction
+
+
BEGINparseSelectorAct
file_t *pFile;
int r;
@@ -409,6 +467,7 @@ CODESTARTparseSelectorAct
}
fileObjAddUser(pFile);
pData->pFile = pFile;
+ pData->offsBuf = 0;
CODE_STD_FINALIZERparseSelectorAct
ENDparseSelectorAct
@@ -455,6 +514,7 @@ ENDmodExit
BEGINqueryEtryPt
CODESTARTqueryEtryPt
CODEqueryEtryPt_STD_OMOD_QUERIES
+CODEqueryEtryPt_TXIF_OMOD_QUERIES /* we support the transactional interface! */
CODEqueryEtryPt_doHUP
ENDqueryEtryPt
@@ -472,5 +532,6 @@ CODEmodInit_QueryRegCFSLineHdlr
CHKiRet(regCfSysLineHdlr((uchar *)"omhdfsport", 0, eCmdHdlrInt, NULL, &hdfsPort, NULL));
CHKiRet(regCfSysLineHdlr((uchar *)"omhdfsdefaulttemplate", 0, eCmdHdlrGetWord, NULL, &dfltTplName, NULL));
CHKiRet(omsdRegCFSLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler, resetConfigVariables, NULL, STD_LOADABLE_MODULE_ID));
+ DBGPRINTF("omhdfs: module compiled with rsyslog version %s.\n", VERSION);
CODEmodInit_QueryRegCFSLineHdlr
ENDmodInit