From 11617d8ab5192c12e3b33cd9c08ac32f1d334a85 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Thu, 30 Sep 2010 14:17:34 +0200 Subject: omhdfs: first shot at this new module (very rough PoC code) --- plugins/omhdfs/omhdfs.c | 296 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 296 insertions(+) create mode 100644 plugins/omhdfs/omhdfs.c (limited to 'plugins/omhdfs/omhdfs.c') diff --git a/plugins/omhdfs/omhdfs.c b/plugins/omhdfs/omhdfs.c new file mode 100644 index 00000000..42e5dc8a --- /dev/null +++ b/plugins/omhdfs/omhdfs.c @@ -0,0 +1,296 @@ +/* omhdfs.c + * This is the implementation of the build-in file output module. + * + * NOTE: read comments in module-template.h to understand how this file + * works! + * + * Copyright 2010 Rainer Gerhards and Adiscon GmbH. + * + * This program is free software; you can redistribute it and/or + * modify it under the terms of the GNU General Public License + * as published by the Free Software Foundation; either version 2 + * of the License, or (at your option) any later version. + * + * This program is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the + * GNU General Public License for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. + * + * A copy of the GPL can be found in the file "COPYING" in this distribution. + */ + +/* this is kind of a hack, if defined, it instructs omhdfs to use + * the regular (non-hdfs) file system calls. This eases development + * (and hopefully troubleshooting) especially in cases when no + * hdfs environment is available. + */ +#define USE_REGULAR_FS 1 + +#include "config.h" +#include "rsyslog.h" +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "syslogd-types.h" +#include "srUtils.h" +#include "template.h" +#include "conf.h" +#include "cfsysline.h" +#include "module-template.h" +#include "unicode-helper.h" +#ifndef USE_REGULAR_FS +# include "hdfs.h" +#endif + +MODULE_TYPE_OUTPUT + +/* internal structures + */ +DEF_OMOD_STATIC_DATA + +/* globals for default values */ +static uchar *fileName = NULL; +/* end globals for default values */ + +typedef struct _instanceData { + uchar *fileName; +# ifdef USE_REGULAR_FS + short fd; /* file descriptor for (current) file */ +# else + hdfsFS fs; + hdfsFile fd; +# endif +} instanceData; + + +BEGINisCompatibleWithFeature +CODESTARTisCompatibleWithFeature + if(eFeat == sFEATURERepeatedMsgReduction) + iRet = RS_RET_OK; +ENDisCompatibleWithFeature + + +BEGINdbgPrintInstInfo +CODESTARTdbgPrintInstInfo + printf("omhdfs: file:%s", pData->fileName); + if (pData->fd == -1) + printf(" (unused)"); +ENDdbgPrintInstInfo + + + +#if 0 +static void prepareFile(instanceData *pData, uchar *newFileName) +{ + if(access((char*)newFileName, F_OK) == 0) { + /* file already exists */ + pData->fd = open((char*) newFileName, O_WRONLY|O_APPEND|O_CREAT|O_NOCTTY, + pData->fCreateMode); + } else { + pData->fd = -1; + /* file does not exist, create it (and eventually parent directories */ + if(pData->bCreateDirs) { + /* we fist need to create parent dirs if they are missing + * We do not report any errors here ourselfs but let the code + * fall through to error handler below. + */ + if(makeFileParentDirs(newFileName, strlen((char*)newFileName), + pData->fDirCreateMode, pData->dirUID, + pData->dirGID, pData->bFailOnChown) != 0) { + return; /* we give up */ + } + } + /* no matter if we needed to create directories or not, we now try to create + * the file. -- rgerhards, 2008-12-18 (based on patch from William Tisater) + */ + pData->fd = open((char*) newFileName, O_WRONLY|O_APPEND|O_CREAT|O_NOCTTY, + pData->fCreateMode); + if(pData->fd != -1) { + /* check and set uid/gid */ + if(pData->fileUID != (uid_t)-1 || pData->fileGID != (gid_t) -1) { + /* we need to set owner/group */ + if(fchown(pData->fd, pData->fileUID, + pData->fileGID) != 0) { + if(pData->bFailOnChown) { + int eSave = errno; + close(pData->fd); + pData->fd = -1; + errno = eSave; + } + /* we will silently ignore the chown() failure + * if configured to do so. + */ + } + } + } + } +} +#endif + +static void prepareFile(instanceData *pData, uchar *newFileName) +{ +# if USE_REGULAR_FS + pData->fd = open((char*) newFileName, O_WRONLY|O_APPEND|O_CREAT|O_NOCTTY, 0666); +# else + pData->fs = hdfsConnect("default", 0); + pData->fd = hdfsOpenFile(fs, newFileName, O_WRONLY|O_CREAT, 0, 0, 0); + if(!pData->fd) { + dbgprintf(stderr, "Failed to open %s for writing!\n", newFileName); + // TODO: suspend/error report + } + +# endif +} + +static rsRetVal writeFile(uchar **ppString, unsigned iMsgOpts, instanceData *pData) +{ + size_t lenWrite; + DEFiRet; + +# if USE_REGULAR_FS + if (write(pData->fd, ppString[0], strlen((char*)ppString[0])) < 0) { + int e = errno; + dbgprintf("omhdfs write error!\n"); + + /* If the filesystem is filled up, just ignore + * it for now and continue writing when possible + */ + //if(pData->fileType == eTypeFILE && e == ENOSPC) + //return RS_RET_OK; + + //(void) close(pData->fd); + iRet = RS_RET_DISABLE_ACTION; + errno = e; + //logerror((char*) pData->f_fname); + } +# else + lenWrite = strlen(char*ppString[0]); + tSize num_written_bytes = hdfsWrite(pData->fs, pData->fd, ppString[0], lenWrite); + if(num_written_bytes != lenWrite) { + dbgprintf("Failed to write %s, expected %lu bytes, written %lu\n", pData->fileName, + lenWrite, num_written_bytes); + // TODO: suspend/error report + } +# endif +else { dbgprintf("omhdfs has successfully written to file\n"); } + + RETiRet; +} + + +BEGINcreateInstance +CODESTARTcreateInstance + pData->fd = -1; +ENDcreateInstance + + +BEGINfreeInstance +CODESTARTfreeInstance +# ifdef USE_REGULAR_FS + if(pData->fd != -1) + close(pData->fd); +# else + hdfsCloseFile(pData->fs, pData->fd); +# endif +ENDfreeInstance + + +BEGINtryResume +CODESTARTtryResume +ENDtryResume + +BEGINdoAction +CODESTARTdoAction + dbgprintf(" (%s)\n", pData->fileName); + iRet = writeFile(ppString, iMsgOpts, pData); +ENDdoAction + + +BEGINparseSelectorAct +CODESTARTparseSelectorAct + + /* first check if this config line is actually for us */ + if(strncmp((char*) p, ":omhdfs:", sizeof(":omhdfs:") - 1)) { + ABORT_FINALIZE(RS_RET_CONFLINE_UNPROCESSED); + } + + /* ok, if we reach this point, we have something for us */ + p += sizeof(":omhdfs:") - 1; /* eat indicator sequence (-1 because of '\0'!) */ + CHKiRet(createInstance(&pData)); + + CODE_STD_STRING_REQUESTparseSelectorAct(1) + /* rgerhards 2004-11-17: from now, we need to have different + * processing, because after the first comma, the template name + * to use is specified. So we need to scan for the first coma first + * and then look at the rest of the line. + */ + CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, 0, (uchar*) "RSYSLOG_FileFormat")); + //(pszFileDfltTplName == NULL) ? (uchar*)"RSYSLOG_FileFormat" : pszFileDfltTplName)); + + // TODO: check for NULL filename + CHKmalloc(pData->fileName = ustrdup(fileName)); + + prepareFile(pData, pData->fileName); + +#if 0 + if ( pData->fd < 0 ){ + pData->fd = -1; + dbgprintf("Error opening log file: %s\n", pData->f_fname); + logerror((char*) pData->f_fname); + } +#endif +CODE_STD_FINALIZERparseSelectorAct +ENDparseSelectorAct + + +/* Reset config variables for this module to default values. + * rgerhards, 2007-07-17 + */ +static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal) +{ +/* + fileUID = -1; + fileGID = -1; + dirUID = -1; + dirGID = -1; + bFailOnChown = 1; + iDynaFileCacheSize = 10; + fCreateMode = 0644; + fDirCreateMode = 0700; + bCreateDirs = 1; +*/ + return RS_RET_OK; +} + + +BEGINmodExit +CODESTARTmodExit +ENDmodExit + + +BEGINqueryEtryPt +CODESTARTqueryEtryPt +CODEqueryEtryPt_STD_OMOD_QUERIES +ENDqueryEtryPt + + +BEGINmodInit() +CODESTARTmodInit + *ipIFVersProvided = CURR_MOD_IF_VERSION; +CODEmodInit_QueryRegCFSLineHdlr + CHKiRet(omsdRegCFSLineHdlr((uchar *)"omhdfsfilename", 0, eCmdHdlrGetWord, NULL, &fileName, NULL)); + CHKiRet(omsdRegCFSLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler, resetConfigVariables, NULL, STD_LOADABLE_MODULE_ID)); +CODEmodInit_QueryRegCFSLineHdlr +ENDmodInit -- cgit v1.2.3 From 7ac7ad166b82034eea4a37c1937ca5ddd618ec45 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Thu, 30 Sep 2010 13:55:26 +0000 Subject: omhdfs: added "real" libhdfs code, now actually works on hdfs very crude implementation, but probably good enough to gather some early performance data and experience with the module. No real error handling done, if something breaks, the whole thing will be blown up ;) --- plugins/omhdfs/omhdfs.c | 62 +++++++++++++++++++++++++++++++++++++------------ 1 file changed, 47 insertions(+), 15 deletions(-) (limited to 'plugins/omhdfs/omhdfs.c') diff --git a/plugins/omhdfs/omhdfs.c b/plugins/omhdfs/omhdfs.c index 42e5dc8a..fcdff6e5 100644 --- a/plugins/omhdfs/omhdfs.c +++ b/plugins/omhdfs/omhdfs.c @@ -28,7 +28,7 @@ * (and hopefully troubleshooting) especially in cases when no * hdfs environment is available. */ -#define USE_REGULAR_FS 1 +//#define USE_REGULAR_FS 1 #include "config.h" #include "rsyslog.h" @@ -62,6 +62,8 @@ DEF_OMOD_STATIC_DATA /* globals for default values */ static uchar *fileName = NULL; +static uchar *hdfsHost = NULL; +int hdfsPort = 0; /* end globals for default values */ typedef struct _instanceData { @@ -71,6 +73,8 @@ typedef struct _instanceData { # else hdfsFS fs; hdfsFile fd; + const char *hdfsHost; + tPort hdfsPort; # endif } instanceData; @@ -85,8 +89,8 @@ ENDisCompatibleWithFeature BEGINdbgPrintInstInfo CODESTARTdbgPrintInstInfo printf("omhdfs: file:%s", pData->fileName); - if (pData->fd == -1) - printf(" (unused)"); + //if (pData->fd == -1) + //printf(" (unused)"); ENDdbgPrintInstInfo @@ -139,22 +143,41 @@ static void prepareFile(instanceData *pData, uchar *newFileName) } #endif -static void prepareFile(instanceData *pData, uchar *newFileName) +static void +prepareFile(instanceData *pData, uchar *newFileName) { # if USE_REGULAR_FS pData->fd = open((char*) newFileName, O_WRONLY|O_APPEND|O_CREAT|O_NOCTTY, 0666); # else - pData->fs = hdfsConnect("default", 0); - pData->fd = hdfsOpenFile(fs, newFileName, O_WRONLY|O_CREAT, 0, 0, 0); + dbgprintf("omhdfs: try to connect to host '%s' at port %d\n", + pData->hdfsHost, pData->hdfsPort); + pData->fs = hdfsConnect(pData->hdfsHost, pData->hdfsPort); + if(pData->fs == NULL) { + dbgprintf("omhdfs: error can not connect to hdfs\n"); + } + pData->fd = hdfsOpenFile(pData->fs, (char*)newFileName, O_WRONLY|O_APPEND, 0, 0, 0); + if(pData->fd == NULL) { + /* maybe the file does not exist, so we try to create it now. + * Note that we can not use hdfsExists() because of a deficit in + * it: https://issues.apache.org/jira/browse/HDFS-1154 + * As of my testing, libhdfs at least seems to return ENOENT if + * the file does not exist. + */ + if(errno == ENOENT) { + dbgprintf("omhdfs: ENOENT trying to append to '%s', now trying create\n", + newFileName); + pData->fd = hdfsOpenFile(pData->fs, (char*)newFileName, O_WRONLY|O_CREAT, 0, 0, 0); + } + } if(!pData->fd) { - dbgprintf(stderr, "Failed to open %s for writing!\n", newFileName); + dbgprintf("omhdfs: failed to open %s for writing!\n", newFileName); // TODO: suspend/error report } # endif } -static rsRetVal writeFile(uchar **ppString, unsigned iMsgOpts, instanceData *pData) +static rsRetVal writeFile(uchar **ppString, instanceData *pData) { size_t lenWrite; DEFiRet; @@ -176,15 +199,14 @@ static rsRetVal writeFile(uchar **ppString, unsigned iMsgOpts, instanceData *pDa //logerror((char*) pData->f_fname); } # else - lenWrite = strlen(char*ppString[0]); + lenWrite = strlen((char*) ppString[0]); tSize num_written_bytes = hdfsWrite(pData->fs, pData->fd, ppString[0], lenWrite); - if(num_written_bytes != lenWrite) { - dbgprintf("Failed to write %s, expected %lu bytes, written %lu\n", pData->fileName, - lenWrite, num_written_bytes); + if((unsigned) num_written_bytes != lenWrite) { + dbgprintf("omhdfs: failed to write %s, expected %lu bytes, written %lu\n", pData->fileName, + lenWrite, (unsigned long) num_written_bytes); // TODO: suspend/error report } # endif -else { dbgprintf("omhdfs has successfully written to file\n"); } RETiRet; } @@ -192,7 +214,7 @@ else { dbgprintf("omhdfs has successfully written to file\n"); } BEGINcreateInstance CODESTARTcreateInstance - pData->fd = -1; + //pData->fd = -1; ENDcreateInstance @@ -214,7 +236,7 @@ ENDtryResume BEGINdoAction CODESTARTdoAction dbgprintf(" (%s)\n", pData->fileName); - iRet = writeFile(ppString, iMsgOpts, pData); + iRet = writeFile(ppString, pData); ENDdoAction @@ -241,6 +263,12 @@ CODESTARTparseSelectorAct // TODO: check for NULL filename CHKmalloc(pData->fileName = ustrdup(fileName)); + if(hdfsHost == NULL) { + pData->hdfsHost = "default"; + } else { + CHKmalloc(pData->hdfsHost = strdup((char*)hdfsHost)); + } + pData->hdfsPort = hdfsPort; prepareFile(pData, pData->fileName); @@ -271,6 +299,8 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a fDirCreateMode = 0700; bCreateDirs = 1; */ + hdfsHost = NULL; + hdfsPort = 0; return RS_RET_OK; } @@ -291,6 +321,8 @@ CODESTARTmodInit *ipIFVersProvided = CURR_MOD_IF_VERSION; CODEmodInit_QueryRegCFSLineHdlr CHKiRet(omsdRegCFSLineHdlr((uchar *)"omhdfsfilename", 0, eCmdHdlrGetWord, NULL, &fileName, NULL)); + CHKiRet(omsdRegCFSLineHdlr((uchar *)"omhdfshost", 0, eCmdHdlrGetWord, NULL, &hdfsHost, NULL)); + CHKiRet(omsdRegCFSLineHdlr((uchar *)"omhdfsport", 0, eCmdHdlrInt, NULL, &hdfsPort, NULL)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"resetconfigvariables", 1, eCmdHdlrCustomHandler, resetConfigVariables, NULL, STD_LOADABLE_MODULE_ID)); CODEmodInit_QueryRegCFSLineHdlr ENDmodInit -- cgit v1.2.3 From d8a1489f545179591abced679ba24831d85ca224 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Fri, 1 Oct 2010 11:29:50 +0200 Subject: omhdfs: cleanup and lots of improvement now things look much better, also done some prep in order to support a file cache (we need this for multiple selectors writing to the same file). --- plugins/omhdfs/omhdfs.c | 247 +++++++++++++++++++++++++++++------------------- 1 file changed, 150 insertions(+), 97 deletions(-) (limited to 'plugins/omhdfs/omhdfs.c') diff --git a/plugins/omhdfs/omhdfs.c b/plugins/omhdfs/omhdfs.c index fcdff6e5..71080585 100644 --- a/plugins/omhdfs/omhdfs.c +++ b/plugins/omhdfs/omhdfs.c @@ -23,13 +23,6 @@ * A copy of the GPL can be found in the file "COPYING" in this distribution. */ -/* this is kind of a hack, if defined, it instructs omhdfs to use - * the regular (non-hdfs) file system calls. This eases development - * (and hopefully troubleshooting) especially in cases when no - * hdfs environment is available. - */ -//#define USE_REGULAR_FS 1 - #include "config.h" #include "rsyslog.h" #include @@ -42,6 +35,8 @@ #include #include #include +#include +#include #include "syslogd-types.h" #include "srUtils.h" @@ -50,15 +45,14 @@ #include "cfsysline.h" #include "module-template.h" #include "unicode-helper.h" -#ifndef USE_REGULAR_FS -# include "hdfs.h" -#endif +#include "errmsg.h" MODULE_TYPE_OUTPUT /* internal structures */ DEF_OMOD_STATIC_DATA +DEFobjCurrIf(errmsg) /* globals for default values */ static uchar *fileName = NULL; @@ -66,18 +60,23 @@ static uchar *hdfsHost = NULL; int hdfsPort = 0; /* end globals for default values */ -typedef struct _instanceData { - uchar *fileName; -# ifdef USE_REGULAR_FS - short fd; /* file descriptor for (current) file */ -# else +typedef struct { + uchar *name; hdfsFS fs; - hdfsFile fd; + hdfsFile fh; const char *hdfsHost; tPort hdfsPort; -# endif + int nUsers; + pthread_mutex_t mut; +} file_t; + + +typedef struct _instanceData { + file_t *pFile; } instanceData; +/* forward definitions (down here, need data types) */ +static inline rsRetVal fileClose(file_t *pFile); BEGINisCompatibleWithFeature CODESTARTisCompatibleWithFeature @@ -88,9 +87,7 @@ ENDisCompatibleWithFeature BEGINdbgPrintInstInfo CODESTARTdbgPrintInstInfo - printf("omhdfs: file:%s", pData->fileName); - //if (pData->fd == -1) - //printf(" (unused)"); + printf("omhdfs: file:%s", pData->pFile->name); ENDdbgPrintInstInfo @@ -100,10 +97,10 @@ static void prepareFile(instanceData *pData, uchar *newFileName) { if(access((char*)newFileName, F_OK) == 0) { /* file already exists */ - pData->fd = open((char*) newFileName, O_WRONLY|O_APPEND|O_CREAT|O_NOCTTY, + pData->fh = open((char*) newFileName, O_WRONLY|O_APPEND|O_CREAT|O_NOCTTY, pData->fCreateMode); } else { - pData->fd = -1; + pData->fh = -1; /* file does not exist, create it (and eventually parent directories */ if(pData->bCreateDirs) { /* we fist need to create parent dirs if they are missing @@ -119,18 +116,18 @@ static void prepareFile(instanceData *pData, uchar *newFileName) /* no matter if we needed to create directories or not, we now try to create * the file. -- rgerhards, 2008-12-18 (based on patch from William Tisater) */ - pData->fd = open((char*) newFileName, O_WRONLY|O_APPEND|O_CREAT|O_NOCTTY, + pData->fh = open((char*) newFileName, O_WRONLY|O_APPEND|O_CREAT|O_NOCTTY, pData->fCreateMode); - if(pData->fd != -1) { + if(pData->fh != -1) { /* check and set uid/gid */ if(pData->fileUID != (uid_t)-1 || pData->fileGID != (gid_t) -1) { /* we need to set owner/group */ - if(fchown(pData->fd, pData->fileUID, + if(fchown(pData->fh, pData->fileUID, pData->fileGID) != 0) { if(pData->bFailOnChown) { int eSave = errno; - close(pData->fd); - pData->fd = -1; + close(pData->fh); + pData->fh = -1; errno = eSave; } /* we will silently ignore the chown() failure @@ -143,20 +140,71 @@ static void prepareFile(instanceData *pData, uchar *newFileName) } #endif -static void -prepareFile(instanceData *pData, uchar *newFileName) +/* ---BEGIN FILE OBJECT---------------------------------------------------- */ +/* This code handles the "file object". This is split from the actual + * instance data, because several instances may write into the same file. + * If so, we need to use a single object, and also synchronize their writes. + * So we keep the file object separately, and just stick a reference into + * the instance data. + */ + +static inline rsRetVal +fileObjConstruct(file_t **ppFile) +{ + file_t *pFile; + DEFiRet; + + CHKmalloc(pFile = malloc(sizeof(file_t))); + pFile->name = NULL; + pFile->hdfsHost = NULL; + pFile->fh = NULL; + pFile->nUsers = 0; + + *ppFile = pFile; +finalize_it: + RETiRet; +} + +static inline void +fileObjAddUser(file_t *pFile) +{ + /* init mutex only when second user is added */ + ++pFile->nUsers; + if(pFile->nUsers == 2) + pthread_mutex_init(&pFile->mut, NULL); +} + +static inline rsRetVal +fileObjDestruct(file_t **ppFile) +{ + file_t *pFile = *ppFile; + if(pFile->nUsers > 1) + pthread_mutex_destroy(&pFile->mut); + fileClose(pFile); + free(pFile->name); + free((char*)pFile->hdfsHost); + free(pFile->fh); + + return RS_RET_OK; +} + +static inline rsRetVal +fileOpen(file_t *pFile) { -# if USE_REGULAR_FS - pData->fd = open((char*) newFileName, O_WRONLY|O_APPEND|O_CREAT|O_NOCTTY, 0666); -# else - dbgprintf("omhdfs: try to connect to host '%s' at port %d\n", - pData->hdfsHost, pData->hdfsPort); - pData->fs = hdfsConnect(pData->hdfsHost, pData->hdfsPort); - if(pData->fs == NULL) { - dbgprintf("omhdfs: error can not connect to hdfs\n"); + DEFiRet; + + assert(pFile->fh == NULL); + if(pFile->nUsers > 1) + d_pthread_mutex_lock(&pFile->mut); + DBGPRINTF("omhdfs: try to connect to HDFS at host '%s', port %d\n", + pFile->hdfsHost, pFile->hdfsPort); + pFile->fs = hdfsConnect(pFile->hdfsHost, pFile->hdfsPort); + if(pFile->fs == NULL) { + DBGPRINTF("omhdfs: error can not connect to hdfs\n"); + ABORT_FINALIZE(RS_RET_SUSPENDED); } - pData->fd = hdfsOpenFile(pData->fs, (char*)newFileName, O_WRONLY|O_APPEND, 0, 0, 0); - if(pData->fd == NULL) { + pFile->fh = hdfsOpenFile(pFile->fs, (char*)pFile->name, O_WRONLY|O_APPEND, 0, 0, 0); + if(pFile->fh == NULL) { /* maybe the file does not exist, so we try to create it now. * Note that we can not use hdfsExists() because of a deficit in * it: https://issues.apache.org/jira/browse/HDFS-1154 @@ -164,68 +212,75 @@ prepareFile(instanceData *pData, uchar *newFileName) * the file does not exist. */ if(errno == ENOENT) { - dbgprintf("omhdfs: ENOENT trying to append to '%s', now trying create\n", - newFileName); - pData->fd = hdfsOpenFile(pData->fs, (char*)newFileName, O_WRONLY|O_CREAT, 0, 0, 0); + 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); } } - if(!pData->fd) { - dbgprintf("omhdfs: failed to open %s for writing!\n", newFileName); - // TODO: suspend/error report + if(pFile->fh == NULL) { + DBGPRINTF("omhdfs: failed to open %s for writing!\n", pFile->name); + ABORT_FINALIZE(RS_RET_SUSPENDED); } -# endif +finalize_it: + if(pFile->nUsers > 1) + d_pthread_mutex_unlock(&pFile->mut); + RETiRet; } -static rsRetVal writeFile(uchar **ppString, instanceData *pData) + +static inline rsRetVal +fileWrite(file_t *pFile, uchar *buf) { size_t lenWrite; DEFiRet; -# if USE_REGULAR_FS - if (write(pData->fd, ppString[0], strlen((char*)ppString[0])) < 0) { - int e = errno; - dbgprintf("omhdfs write error!\n"); - - /* If the filesystem is filled up, just ignore - * it for now and continue writing when possible - */ - //if(pData->fileType == eTypeFILE && e == ENOSPC) - //return RS_RET_OK; - - //(void) close(pData->fd); - iRet = RS_RET_DISABLE_ACTION; - errno = e; - //logerror((char*) pData->f_fname); - } -# else - lenWrite = strlen((char*) ppString[0]); - tSize num_written_bytes = hdfsWrite(pData->fs, pData->fd, ppString[0], lenWrite); + assert(pFile->fh != NULL); + if(pFile->nUsers > 1) + d_pthread_mutex_lock(&pFile->mut); + lenWrite = strlen((char*) buf); + tSize num_written_bytes = hdfsWrite(pFile->fs, pFile->fh, buf, lenWrite); if((unsigned) num_written_bytes != lenWrite) { - dbgprintf("omhdfs: failed to write %s, expected %lu bytes, written %lu\n", pData->fileName, - lenWrite, (unsigned long) num_written_bytes); - // TODO: suspend/error report + errmsg.LogError(errno, RS_RET_ERR_HDFS_WRITE, "omhdfs: failed to write %s, expected %lu bytes, " + "written %lu\n", pFile->name, lenWrite, (unsigned long) num_written_bytes); + ABORT_FINALIZE(RS_RET_SUSPENDED); } -# endif +finalize_it: + if(pFile->nUsers > 1) + d_pthread_mutex_unlock(&pFile->mut); RETiRet; } +static inline rsRetVal +fileClose(file_t *pFile) +{ + if(pFile->nUsers > 1) + d_pthread_mutex_lock(&pFile->mut); + if(pFile->fh != NULL) { + hdfsCloseFile(pFile->fs, pFile->fh); + pFile->fh = NULL; + } + + if(pFile->nUsers > 1) + d_pthread_mutex_unlock(&pFile->mut); + + return RS_RET_OK; +} + +/* ---END FILE OBJECT---------------------------------------------------- */ + + BEGINcreateInstance CODESTARTcreateInstance - //pData->fd = -1; + pData->pFile = NULL; ENDcreateInstance BEGINfreeInstance CODESTARTfreeInstance -# ifdef USE_REGULAR_FS - if(pData->fd != -1) - close(pData->fd); -# else - hdfsCloseFile(pData->fs, pData->fd); -# endif + fileObjDestruct(&pData->pFile); ENDfreeInstance @@ -235,8 +290,8 @@ ENDtryResume BEGINdoAction CODESTARTdoAction - dbgprintf(" (%s)\n", pData->fileName); - iRet = writeFile(ppString, pData); + DBGPRINTF(" (%s)\n", pData->pFile->name); + iRet = fileWrite(pData->pFile, ppString[0]); ENDdoAction @@ -251,34 +306,29 @@ CODESTARTparseSelectorAct /* ok, if we reach this point, we have something for us */ p += sizeof(":omhdfs:") - 1; /* eat indicator sequence (-1 because of '\0'!) */ CHKiRet(createInstance(&pData)); - CODE_STD_STRING_REQUESTparseSelectorAct(1) - /* rgerhards 2004-11-17: from now, we need to have different - * processing, because after the first comma, the template name - * to use is specified. So we need to scan for the first coma first - * and then look at the rest of the line. - */ CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, 0, (uchar*) "RSYSLOG_FileFormat")); //(pszFileDfltTplName == NULL) ? (uchar*)"RSYSLOG_FileFormat" : pszFileDfltTplName)); - // TODO: check for NULL filename - CHKmalloc(pData->fileName = ustrdup(fileName)); + if(fileName == NULL) { + errmsg.LogError(0, RS_RET_ERR_HDFS_OPEN, "omhdfs: no file name specified, can not continue"); + ABORT_FINALIZE(RS_RET_FILE_NOT_SPECIFIED); + } + + CHKiRet(fileObjConstruct(&pData->pFile)); if(hdfsHost == NULL) { - pData->hdfsHost = "default"; + CHKmalloc(pData->pFile->hdfsHost = strdup("default")); } else { - CHKmalloc(pData->hdfsHost = strdup((char*)hdfsHost)); + CHKmalloc(pData->pFile->hdfsHost = strdup((char*)hdfsHost)); } - pData->hdfsPort = hdfsPort; + pData->pFile->hdfsPort = hdfsPort; - prepareFile(pData, pData->fileName); + fileOpen(pData->pFile); -#if 0 - if ( pData->fd < 0 ){ - pData->fd = -1; - dbgprintf("Error opening log file: %s\n", pData->f_fname); - logerror((char*) pData->f_fname); + if(pData->pFile->fh == NULL){ + errmsg.LogError(0, RS_RET_ERR_HDFS_OPEN, "omhdfs: failed to open %s - retrying later", pData->pFile->name); + iRet = RS_RET_SUSPENDED; } -#endif CODE_STD_FINALIZERparseSelectorAct ENDparseSelectorAct @@ -307,6 +357,7 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a BEGINmodExit CODESTARTmodExit + objRelease(errmsg, CORE_COMPONENT); ENDmodExit @@ -320,6 +371,8 @@ BEGINmodInit() CODESTARTmodInit *ipIFVersProvided = CURR_MOD_IF_VERSION; CODEmodInit_QueryRegCFSLineHdlr + CHKiRet(objUse(errmsg, CORE_COMPONENT)); + CHKiRet(omsdRegCFSLineHdlr((uchar *)"omhdfsfilename", 0, eCmdHdlrGetWord, NULL, &fileName, NULL)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"omhdfshost", 0, eCmdHdlrGetWord, NULL, &hdfsHost, NULL)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"omhdfsport", 0, eCmdHdlrInt, NULL, &hdfsPort, NULL)); -- cgit v1.2.3 From 0ee524c391d017225049542fffe572d7de7d1512 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Fri, 1 Oct 2010 09:55:21 +0000 Subject: omhdfs: fixed small bug ... that I could only see on my execution environment (I right now have two envs, a local one where I can compile, but not run and a remote one where I can do both, but this is a bit less convenient to use). --- plugins/omhdfs/omhdfs.c | 1 + 1 file changed, 1 insertion(+) (limited to 'plugins/omhdfs/omhdfs.c') diff --git a/plugins/omhdfs/omhdfs.c b/plugins/omhdfs/omhdfs.c index 71080585..5d28d5dd 100644 --- a/plugins/omhdfs/omhdfs.c +++ b/plugins/omhdfs/omhdfs.c @@ -316,6 +316,7 @@ CODESTARTparseSelectorAct } CHKiRet(fileObjConstruct(&pData->pFile)); + CHKmalloc(pData->pFile->name = (uchar*)strdup((char*)fileName)); if(hdfsHost == NULL) { CHKmalloc(pData->pFile->hdfsHost = strdup("default")); } else { -- cgit v1.2.3 From 9696cdef34f5d033564138fb9d4afb87daa6b1be Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Fri, 1 Oct 2010 12:32:01 +0200 Subject: omhdfs: files now kept inside a hashtable for use by multiple actions Note:compiles, but not yet tested --- plugins/omhdfs/omhdfs.c | 40 +++++++++++++++++++++++++++++----------- 1 file changed, 29 insertions(+), 11 deletions(-) (limited to 'plugins/omhdfs/omhdfs.c') diff --git a/plugins/omhdfs/omhdfs.c b/plugins/omhdfs/omhdfs.c index 5d28d5dd..eaca90e4 100644 --- a/plugins/omhdfs/omhdfs.c +++ b/plugins/omhdfs/omhdfs.c @@ -46,6 +46,7 @@ #include "module-template.h" #include "unicode-helper.h" #include "errmsg.h" +#include "hashtable.h" MODULE_TYPE_OUTPUT @@ -54,6 +55,9 @@ MODULE_TYPE_OUTPUT DEF_OMOD_STATIC_DATA DEFobjCurrIf(errmsg) +/* global data */ +static struct hashtable *files; /* holds all file objects that we know */ + /* globals for default values */ static uchar *fileName = NULL; static uchar *hdfsHost = NULL; @@ -172,6 +176,7 @@ fileObjAddUser(file_t *pFile) ++pFile->nUsers; if(pFile->nUsers == 2) pthread_mutex_init(&pFile->mut, NULL); + dbgprintf("omhdfs: file %s now being used by %d actions\n", pFile->name, pFile->nUsers); } static inline rsRetVal @@ -296,6 +301,9 @@ ENDdoAction BEGINparseSelectorAct + file_t *pFile; + int r; + uchar *keybuf; CODESTARTparseSelectorAct /* first check if this config line is actually for us */ @@ -315,21 +323,28 @@ CODESTARTparseSelectorAct ABORT_FINALIZE(RS_RET_FILE_NOT_SPECIFIED); } - CHKiRet(fileObjConstruct(&pData->pFile)); - CHKmalloc(pData->pFile->name = (uchar*)strdup((char*)fileName)); - if(hdfsHost == NULL) { - CHKmalloc(pData->pFile->hdfsHost = strdup("default")); - } else { - CHKmalloc(pData->pFile->hdfsHost = strdup((char*)hdfsHost)); + pFile = hashtable_search(files, fileName); + if(pFile == NULL) { + /* we need a new file object, this one not seen before */ + CHKiRet(fileObjConstruct(&pFile)); + CHKmalloc(pFile->name = (uchar*)strdup((char*)fileName)); + CHKmalloc(keybuf = ustrdup(fileName)); + r = hashtable_insert(files, keybuf, pFile); + if(r == 0) + ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); } - pData->pFile->hdfsPort = hdfsPort; + fileObjAddUser(pFile); - fileOpen(pData->pFile); - - if(pData->pFile->fh == NULL){ - errmsg.LogError(0, RS_RET_ERR_HDFS_OPEN, "omhdfs: failed to open %s - retrying later", pData->pFile->name); + CHKmalloc(pFile->hdfsHost = strdup((hdfsHost == NULL) ? "default" : (char*) hdfsHost)); + pFile->hdfsPort = hdfsPort; + fileOpen(pFile); + if(pFile->fh == NULL){ + errmsg.LogError(0, RS_RET_ERR_HDFS_OPEN, "omhdfs: failed to open %s - retrying later", pFile->name); iRet = RS_RET_SUSPENDED; } + + pData->pFile = pFile; + CODE_STD_FINALIZERparseSelectorAct ENDparseSelectorAct @@ -359,6 +374,8 @@ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __a BEGINmodExit CODESTARTmodExit objRelease(errmsg, CORE_COMPONENT); + if(files != NULL) + hashtable_destroy(files, 1); /* 1 => free all values automatically */ ENDmodExit @@ -373,6 +390,7 @@ CODESTARTmodInit *ipIFVersProvided = CURR_MOD_IF_VERSION; CODEmodInit_QueryRegCFSLineHdlr CHKiRet(objUse(errmsg, CORE_COMPONENT)); + CHKmalloc(files = create_hashtable(20, hash_from_string, key_equals_string)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"omhdfsfilename", 0, eCmdHdlrGetWord, NULL, &fileName, NULL)); CHKiRet(omsdRegCFSLineHdlr((uchar *)"omhdfshost", 0, eCmdHdlrGetWord, NULL, &hdfsHost, NULL)); -- cgit v1.2.3 From 255895a58b3f2a54fecf971da700caf265b4e1f0 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Fri, 1 Oct 2010 13:59:48 +0000 Subject: omhdfs: more improvements finally this looks almost production ready for files where no directory path needs to be created --- plugins/omhdfs/omhdfs.c | 47 +++++++++++++++++++++++++++++++++-------------- 1 file changed, 33 insertions(+), 14 deletions(-) (limited to 'plugins/omhdfs/omhdfs.c') diff --git a/plugins/omhdfs/omhdfs.c b/plugins/omhdfs/omhdfs.c index eaca90e4..734c28cd 100644 --- a/plugins/omhdfs/omhdfs.c +++ b/plugins/omhdfs/omhdfs.c @@ -61,6 +61,7 @@ static struct hashtable *files; /* holds all file objects that we know */ /* globals for default values */ static uchar *fileName = NULL; static uchar *hdfsHost = NULL; +static uchar *dfltTplName = NULL; /* default template name to use */ int hdfsPort = 0; /* end globals for default values */ @@ -176,10 +177,10 @@ fileObjAddUser(file_t *pFile) ++pFile->nUsers; if(pFile->nUsers == 2) pthread_mutex_init(&pFile->mut, NULL); - dbgprintf("omhdfs: file %s now being used by %d actions\n", pFile->name, pFile->nUsers); + DBGPRINTF("omhdfs: file %s now being used by %d actions\n", pFile->name, pFile->nUsers); } -static inline rsRetVal +static rsRetVal fileObjDestruct(file_t **ppFile) { file_t *pFile = *ppFile; @@ -193,6 +194,18 @@ fileObjDestruct(file_t **ppFile) return RS_RET_OK; } +/* this function is to be used as destructor for the + * hash table code. + */ +static void +fileObjDestruct4Hashtable(void *ptr) +{ + dbgprintf("omfile: fileObjDestruct4Hashtable called\n"); + file_t *pFile = (file_t*) ptr; + fileObjDestruct(&pFile); +} + + static inline rsRetVal fileOpen(file_t *pFile) { @@ -261,6 +274,8 @@ finalize_it: static inline rsRetVal fileClose(file_t *pFile) { + DEFiRet; + if(pFile->nUsers > 1) d_pthread_mutex_lock(&pFile->mut); if(pFile->fh != NULL) { @@ -271,7 +286,7 @@ fileClose(file_t *pFile) if(pFile->nUsers > 1) d_pthread_mutex_unlock(&pFile->mut); - return RS_RET_OK; + RETiRet; } /* ---END FILE OBJECT---------------------------------------------------- */ @@ -285,7 +300,8 @@ ENDcreateInstance BEGINfreeInstance CODESTARTfreeInstance - fileObjDestruct(&pData->pFile); + if(pData->pFile != NULL) + fileObjDestruct(&pData->pFile); ENDfreeInstance @@ -315,8 +331,8 @@ CODESTARTparseSelectorAct p += sizeof(":omhdfs:") - 1; /* eat indicator sequence (-1 because of '\0'!) */ CHKiRet(createInstance(&pData)); CODE_STD_STRING_REQUESTparseSelectorAct(1) - CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, 0, (uchar*) "RSYSLOG_FileFormat")); - //(pszFileDfltTplName == NULL) ? (uchar*)"RSYSLOG_FileFormat" : pszFileDfltTplName)); + CHKiRet(cflineParseTemplateName(&p, *ppOMSR, 0, 0, + (dfltTplName == NULL) ? (uchar*)"RSYSLOG_FileFormat" : dfltTplName)); if(fileName == NULL) { errmsg.LogError(0, RS_RET_ERR_HDFS_OPEN, "omhdfs: no file name specified, can not continue"); @@ -327,17 +343,18 @@ CODESTARTparseSelectorAct if(pFile == NULL) { /* we need a new file object, this one not seen before */ CHKiRet(fileObjConstruct(&pFile)); - CHKmalloc(pFile->name = (uchar*)strdup((char*)fileName)); + CHKmalloc(pFile->name = fileName); CHKmalloc(keybuf = ustrdup(fileName)); + fileName = NULL; /* re-set, data passed to file object */ + CHKmalloc(pFile->hdfsHost = strdup((hdfsHost == NULL) ? "default" : (char*) hdfsHost)); + pFile->hdfsPort = hdfsPort; + fileOpen(pFile); r = hashtable_insert(files, keybuf, pFile); if(r == 0) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); } fileObjAddUser(pFile); - CHKmalloc(pFile->hdfsHost = strdup((hdfsHost == NULL) ? "default" : (char*) hdfsHost)); - pFile->hdfsPort = hdfsPort; - fileOpen(pFile); if(pFile->fh == NULL){ errmsg.LogError(0, RS_RET_ERR_HDFS_OPEN, "omhdfs: failed to open %s - retrying later", pFile->name); iRet = RS_RET_SUSPENDED; @@ -390,11 +407,13 @@ CODESTARTmodInit *ipIFVersProvided = CURR_MOD_IF_VERSION; CODEmodInit_QueryRegCFSLineHdlr CHKiRet(objUse(errmsg, CORE_COMPONENT)); - CHKmalloc(files = create_hashtable(20, hash_from_string, key_equals_string)); + CHKmalloc(files = create_hashtable(20, hash_from_string, key_equals_string, + fileObjDestruct4Hashtable)); - CHKiRet(omsdRegCFSLineHdlr((uchar *)"omhdfsfilename", 0, eCmdHdlrGetWord, NULL, &fileName, NULL)); - CHKiRet(omsdRegCFSLineHdlr((uchar *)"omhdfshost", 0, eCmdHdlrGetWord, NULL, &hdfsHost, NULL)); - CHKiRet(omsdRegCFSLineHdlr((uchar *)"omhdfsport", 0, eCmdHdlrInt, NULL, &hdfsPort, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"omhdfsfilename", 0, eCmdHdlrGetWord, NULL, &fileName, NULL)); + CHKiRet(regCfSysLineHdlr((uchar *)"omhdfshost", 0, eCmdHdlrGetWord, NULL, &hdfsHost, NULL)); + 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)); CODEmodInit_QueryRegCFSLineHdlr ENDmodInit -- cgit v1.2.3 From 670e81c9a8275a5509efd71dae66d9a267ec1574 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Fri, 1 Oct 2010 14:36:39 +0000 Subject: omhdfs: made action suspend/resume working --- plugins/omhdfs/omhdfs.c | 36 ++++++++++++++++++++++++++---------- 1 file changed, 26 insertions(+), 10 deletions(-) (limited to 'plugins/omhdfs/omhdfs.c') diff --git a/plugins/omhdfs/omhdfs.c b/plugins/omhdfs/omhdfs.c index 734c28cd..4fbf2ef4 100644 --- a/plugins/omhdfs/omhdfs.c +++ b/plugins/omhdfs/omhdfs.c @@ -276,16 +276,19 @@ fileClose(file_t *pFile) { DEFiRet; + if(pFile->fh == NULL) + FINALIZE; + if(pFile->nUsers > 1) d_pthread_mutex_lock(&pFile->mut); - if(pFile->fh != NULL) { - hdfsCloseFile(pFile->fs, pFile->fh); - pFile->fh = NULL; - } + + hdfsCloseFile(pFile->fs, pFile->fh); + pFile->fh = NULL; if(pFile->nUsers > 1) d_pthread_mutex_unlock(&pFile->mut); +finalize_it: RETiRet; } @@ -307,12 +310,26 @@ ENDfreeInstance BEGINtryResume CODESTARTtryResume + fileClose(pData->pFile); + fileOpen(pData->pFile); + if(pData->pFile->fh == NULL){ + dbgprintf("omhdfs: tried to resume file %s, but still no luck...\n", + pData->pFile->name); + iRet = RS_RET_SUSPENDED; + } ENDtryResume BEGINdoAction CODESTARTdoAction DBGPRINTF(" (%s)\n", pData->pFile->name); + if(pData->pFile->fh == NULL) { + fileOpen(pData->pFile); + if(pData->pFile->fh == NULL) { + ABORT_FINALIZE(RS_RET_SUSPENDED); + } + } iRet = fileWrite(pData->pFile, ppString[0]); +finalize_it: ENDdoAction @@ -349,17 +366,16 @@ CODESTARTparseSelectorAct CHKmalloc(pFile->hdfsHost = strdup((hdfsHost == NULL) ? "default" : (char*) hdfsHost)); pFile->hdfsPort = hdfsPort; fileOpen(pFile); + if(pFile->fh == NULL){ + errmsg.LogError(0, RS_RET_ERR_HDFS_OPEN, "omhdfs: failed to open %s - " + "retrying later", pFile->name); + iRet = RS_RET_SUSPENDED; + } r = hashtable_insert(files, keybuf, pFile); if(r == 0) ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY); } fileObjAddUser(pFile); - - if(pFile->fh == NULL){ - errmsg.LogError(0, RS_RET_ERR_HDFS_OPEN, "omhdfs: failed to open %s - retrying later", pFile->name); - iRet = RS_RET_SUSPENDED; - } - pData->pFile = pFile; CODE_STD_FINALIZERparseSelectorAct -- cgit v1.2.3 From d5f16404f93d54afddebb9fb683469fc712d2335 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Fri, 1 Oct 2010 17:26:14 +0200 Subject: omhdfs: added doc --- plugins/omhdfs/omhdfs.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'plugins/omhdfs/omhdfs.c') diff --git a/plugins/omhdfs/omhdfs.c b/plugins/omhdfs/omhdfs.c index 4fbf2ef4..42ed834f 100644 --- a/plugins/omhdfs/omhdfs.c +++ b/plugins/omhdfs/omhdfs.c @@ -1,5 +1,5 @@ /* omhdfs.c - * This is the implementation of the build-in file output module. + * This is an output module to support Hadoop's HDFS. * * NOTE: read comments in module-template.h to understand how this file * works! -- cgit v1.2.3 From e40cb595a2da943483124fae8f215a397add9fca Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Mon, 4 Oct 2010 13:58:41 +0200 Subject: omhdfs: support for HUP added --- plugins/omhdfs/omhdfs.c | 42 +++++++++++++++++++++++++++++++++--------- 1 file changed, 33 insertions(+), 9 deletions(-) (limited to 'plugins/omhdfs/omhdfs.c') diff --git a/plugins/omhdfs/omhdfs.c b/plugins/omhdfs/omhdfs.c index 42ed834f..eefea722 100644 --- a/plugins/omhdfs/omhdfs.c +++ b/plugins/omhdfs/omhdfs.c @@ -47,6 +47,7 @@ #include "unicode-helper.h" #include "errmsg.h" #include "hashtable.h" +#include "hashtable_itr.h" MODULE_TYPE_OUTPUT @@ -200,7 +201,6 @@ fileObjDestruct(file_t **ppFile) static void fileObjDestruct4Hashtable(void *ptr) { - dbgprintf("omfile: fileObjDestruct4Hashtable called\n"); file_t *pFile = (file_t*) ptr; fileObjDestruct(&pFile); } @@ -214,6 +214,7 @@ fileOpen(file_t *pFile) assert(pFile->fh == NULL); if(pFile->nUsers > 1) d_pthread_mutex_lock(&pFile->mut); + DBGPRINTF("omhdfs: try to connect to HDFS at host '%s', port %d\n", pFile->hdfsHost, pFile->hdfsPort); pFile->fs = hdfsConnect(pFile->hdfsHost, pFile->hdfsPort); @@ -256,6 +257,17 @@ fileWrite(file_t *pFile, uchar *buf) assert(pFile->fh != NULL); if(pFile->nUsers > 1) d_pthread_mutex_lock(&pFile->mut); + + /* open file if not open. This must be done *here* and while mutex-protected + * because of HUP handling (which is async to normal processing!). + */ + if(pFile->fh == NULL) { + fileOpen(pFile); + if(pFile->fh == NULL) { + ABORT_FINALIZE(RS_RET_SUSPENDED); + } + } + lenWrite = strlen((char*) buf); tSize num_written_bytes = hdfsWrite(pFile->fs, pFile->fh, buf, lenWrite); if((unsigned) num_written_bytes != lenWrite) { @@ -321,15 +333,8 @@ ENDtryResume BEGINdoAction CODESTARTdoAction - DBGPRINTF(" (%s)\n", pData->pFile->name); - if(pData->pFile->fh == NULL) { - fileOpen(pData->pFile); - if(pData->pFile->fh == NULL) { - ABORT_FINALIZE(RS_RET_SUSPENDED); - } - } + DBGPRINTF("omuxsock: action to to write to %s\n", pData->pFile->name); iRet = fileWrite(pData->pFile, ppString[0]); -finalize_it: ENDdoAction @@ -382,6 +387,24 @@ CODE_STD_FINALIZERparseSelectorAct ENDparseSelectorAct +BEGINdoHUP + file_t *pFile; + struct hashtable_itr *itr; +CODESTARTdoHUP + /* Iterator constructor only returns a valid iterator if + * the hashtable is not empty */ + itr = hashtable_iterator(files); + if(hashtable_count(files) > 0) + { + do { + pFile = (file_t *) hashtable_iterator_value(itr); + fileClose(pFile); + DBGPRINTF("imuxsock: HUP, closing file %s\n", pFile->name); + } while (hashtable_iterator_advance(itr)); + } +ENDdoHUP + + /* Reset config variables for this module to default values. * rgerhards, 2007-07-17 */ @@ -415,6 +438,7 @@ ENDmodExit BEGINqueryEtryPt CODESTARTqueryEtryPt CODEqueryEtryPt_STD_OMOD_QUERIES +CODEqueryEtryPt_doHUP ENDqueryEtryPt -- cgit v1.2.3 From 4b9a92bc725f7436b7958e673a9665a90b548e86 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Mon, 4 Oct 2010 12:46:39 +0000 Subject: omhdfs: fixed some issues with previous commit --- plugins/omhdfs/omhdfs.c | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) (limited to 'plugins/omhdfs/omhdfs.c') diff --git a/plugins/omhdfs/omhdfs.c b/plugins/omhdfs/omhdfs.c index eefea722..b075432d 100644 --- a/plugins/omhdfs/omhdfs.c +++ b/plugins/omhdfs/omhdfs.c @@ -254,7 +254,6 @@ fileWrite(file_t *pFile, uchar *buf) size_t lenWrite; DEFiRet; - assert(pFile->fh != NULL); if(pFile->nUsers > 1) d_pthread_mutex_lock(&pFile->mut); @@ -391,17 +390,18 @@ BEGINdoHUP file_t *pFile; struct hashtable_itr *itr; CODESTARTdoHUP - /* Iterator constructor only returns a valid iterator if - * the hashtable is not empty */ - itr = hashtable_iterator(files); - if(hashtable_count(files) > 0) - { - do { - pFile = (file_t *) hashtable_iterator_value(itr); - fileClose(pFile); - DBGPRINTF("imuxsock: HUP, closing file %s\n", pFile->name); - } while (hashtable_iterator_advance(itr)); - } + DBGPRINTF("omhdfs: HUP received (file count %d)\n", hashtable_count(files)); + /* Iterator constructor only returns a valid iterator if + * the hashtable is not empty */ + itr = hashtable_iterator(files); + if(hashtable_count(files) > 0) + { + do { + pFile = (file_t *) hashtable_iterator_value(itr); + fileClose(pFile); + DBGPRINTF("omhdfs: HUP, closing file %s\n", pFile->name); + } while (hashtable_iterator_advance(itr)); + } ENDdoHUP -- cgit v1.2.3 From a627ac99ba2c3404ca926a19fb06cbd6f43b53c8 Mon Sep 17 00:00:00 2001 From: Rainer Gerhards Date: Mon, 4 Oct 2010 16:25:49 +0200 Subject: omhdfs: added ability to create non-existing directories in name path --- plugins/omhdfs/omhdfs.c | 128 +++++++++++++++++++++++++++--------------------- 1 file changed, 72 insertions(+), 56 deletions(-) (limited to 'plugins/omhdfs/omhdfs.c') diff --git a/plugins/omhdfs/omhdfs.c b/plugins/omhdfs/omhdfs.c index b075432d..9705b7fd 100644 --- a/plugins/omhdfs/omhdfs.c +++ b/plugins/omhdfs/omhdfs.c @@ -97,54 +97,41 @@ CODESTARTdbgPrintInstInfo ENDdbgPrintInstInfo - -#if 0 -static void prepareFile(instanceData *pData, uchar *newFileName) +/* note that hdfsFileExists() does not work, so we did our + * own function to see if a pathname exists. Returns 0 if the + * file does not exists, something else otherwise. Note that + * we can also check a directroy (if that matters...) + */ +static int +HDFSFileExists(hdfsFS fs, uchar *name) { - if(access((char*)newFileName, F_OK) == 0) { - /* file already exists */ - pData->fh = open((char*) newFileName, O_WRONLY|O_APPEND|O_CREAT|O_NOCTTY, - pData->fCreateMode); + int r; + hdfsFileInfo *info; + + info = hdfsGetPathInfo(fs, (char*)name); + /* if things go wrong, we assume it is because the file + * does not exist. We do not get too much information... + */ + if(info == NULL) { + r = 0; } else { - pData->fh = -1; - /* file does not exist, create it (and eventually parent directories */ - if(pData->bCreateDirs) { - /* we fist need to create parent dirs if they are missing - * We do not report any errors here ourselfs but let the code - * fall through to error handler below. - */ - if(makeFileParentDirs(newFileName, strlen((char*)newFileName), - pData->fDirCreateMode, pData->dirUID, - pData->dirGID, pData->bFailOnChown) != 0) { - return; /* we give up */ - } - } - /* no matter if we needed to create directories or not, we now try to create - * the file. -- rgerhards, 2008-12-18 (based on patch from William Tisater) - */ - pData->fh = open((char*) newFileName, O_WRONLY|O_APPEND|O_CREAT|O_NOCTTY, - pData->fCreateMode); - if(pData->fh != -1) { - /* check and set uid/gid */ - if(pData->fileUID != (uid_t)-1 || pData->fileGID != (gid_t) -1) { - /* we need to set owner/group */ - if(fchown(pData->fh, pData->fileUID, - pData->fileGID) != 0) { - if(pData->bFailOnChown) { - int eSave = errno; - close(pData->fh); - pData->fh = -1; - errno = eSave; - } - /* we will silently ignore the chown() failure - * if configured to do so. - */ - } - } - } + r = 1; + hdfsFreeFileInfo(info, 1); } + return r; +} + +static inline rsRetVal +HDFSmkdir(hdfsFS fs, uchar *name) +{ + DEFiRet; + if(hdfsCreateDirectory(fs, (char*)name) == -1) + ABORT_FINALIZE(RS_RET_ERR); + +finalize_it: + RETiRet; } -#endif + /* ---BEGIN FILE OBJECT---------------------------------------------------- */ /* This code handles the "file object". This is split from the actual @@ -195,6 +182,42 @@ fileObjDestruct(file_t **ppFile) return RS_RET_OK; } + +/* check, and potentially create, all names inside a path */ +static rsRetVal +filePrepare(file_t *pFile) +{ + uchar *p; + uchar *pszWork; + size_t len; + DEFiRet; + + if(HDFSFileExists(pFile->fs, pFile->name)) + FINALIZE; + + /* file does not exist, create it (and eventually parent directories */ + if(1) { // check if bCreateDirs + len = ustrlen(pFile->name) + 1; + CHKmalloc(pszWork = MALLOC(sizeof(uchar) * len)); + memcpy(pszWork, pFile->name, len); + for(p = pszWork+1 ; *p ; p++) + if(*p == '/') { + /* temporarily terminate string, create dir and go on */ + *p = '\0'; + if(!HDFSFileExists(pFile->fs, pszWork)) { + CHKiRet(HDFSmkdir(pFile->fs, pszWork)); + } + *p = '/'; + } + free(pszWork); + return 0; + } + +finalize_it: + RETiRet; +} + + /* this function is to be used as destructor for the * hash table code. */ @@ -222,6 +245,9 @@ fileOpen(file_t *pFile) DBGPRINTF("omhdfs: error can not connect to hdfs\n"); ABORT_FINALIZE(RS_RET_SUSPENDED); } + + CHKiRet(filePrepare(pFile)); + pFile->fh = hdfsOpenFile(pFile->fs, (char*)pFile->name, O_WRONLY|O_APPEND, 0, 0, 0); if(pFile->fh == NULL) { /* maybe the file does not exist, so we try to create it now. @@ -271,7 +297,8 @@ fileWrite(file_t *pFile, uchar *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, lenWrite, (unsigned long) num_written_bytes); + "written %lu\n", pFile->name, (unsigned long) lenWrite, + (unsigned long) num_written_bytes); ABORT_FINALIZE(RS_RET_SUSPENDED); } @@ -410,17 +437,6 @@ ENDdoHUP */ static rsRetVal resetConfigVariables(uchar __attribute__((unused)) *pp, void __attribute__((unused)) *pVal) { -/* - fileUID = -1; - fileGID = -1; - dirUID = -1; - dirGID = -1; - bFailOnChown = 1; - iDynaFileCacheSize = 10; - fCreateMode = 0644; - fDirCreateMode = 0700; - bCreateDirs = 1; -*/ hdfsHost = NULL; hdfsPort = 0; return RS_RET_OK; -- cgit v1.2.3