- shuffled code from tcpsyslog.c to omfwd.c. It looks like it belongs more

    to that file. But we need to look at it some time later. The move was
    absolutely necessary so that no access to f->f_un happened in
    tcpsyslog.c (which was evil)
- milestone reache: no more access to f->f_un from non-output modules

1 files changed, 370 insertions, 0 deletions

diff --git a/omfwd.c b/omfwd.c
index 44e1861e..81aa5424 100644
--- a/omfwd.c
+++ b/omfwd.c
@@ -41,6 +41,7 @@
 #include <assert.h>
 #include <errno.h>
 #include <ctype.h>
+#include <unistd.h>
 #ifdef USE_PTHREADS
 #include <pthread.h>
 #endif
@@ -113,6 +114,375 @@ CODESTARTdbgPrintInstInfo
 	printf("%s", f->f_un.f_forw.f_hname);
 ENDdbgPrintInstInfo
 
+/* CODE FOR SENDING TCP MESSAGES */
+
+/* get send status
+ * rgerhards, 2005-10-24
+ */
+static void TCPSendSetStatus(selector_t *f, enum TCPSendStatus iNewState)
+{
+	assert(f != NULL);
+	assert(f->f_type == F_FORW);
+	assert(f->f_un.f_forw.protocol == FORW_TCP);
+	assert(   (iNewState == TCP_SEND_NOTCONNECTED)
+	       || (iNewState == TCP_SEND_CONNECTING)
+	       || (iNewState == TCP_SEND_READY));
+
+	/* there can potentially be a race condition, so guard by mutex */
+#	ifdef	USE_PTHREADS
+		pthread_mutex_lock(&f->f_un.f_forw.mtxTCPSend);
+#	endif
+	f->f_un.f_forw.status = iNewState;
+#	ifdef	USE_PTHREADS
+		pthread_mutex_unlock(&f->f_un.f_forw.mtxTCPSend);
+#	endif
+}
+
+
+/* set send status
+ * rgerhards, 2005-10-24
+ */
+static enum TCPSendStatus TCPSendGetStatus(selector_t *f)
+{
+	enum TCPSendStatus eState;
+	assert(f != NULL);
+	assert(f->f_type == F_FORW);
+	assert(f->f_un.f_forw.protocol == FORW_TCP);
+
+	/* there can potentially be a race condition, so guard by mutex */
+#	ifdef	USE_PTHREADS
+		pthread_mutex_lock(&f->f_un.f_forw.mtxTCPSend);
+#	endif
+	eState = f->f_un.f_forw.status;
+#	ifdef	USE_PTHREADS
+		pthread_mutex_unlock(&f->f_un.f_forw.mtxTCPSend);
+#	endif
+
+	return eState;
+}
+
+
+/* Initialize TCP sockets (for sender)
+ * This is done once per selector line, if not yet initialized.
+ */
+static int TCPSendCreateSocket(selector_t *f, struct addrinfo *addrDest)
+{
+	int fd;
+	struct addrinfo *r; 
+	
+	assert(f != NULL);
+	
+	r = addrDest;
+
+	while(r != NULL) {
+		fd = socket(r->ai_family, r->ai_socktype, r->ai_protocol);
+		if (fd != -1) {
+			/* We can not allow the TCP sender to block syslogd, at least
+			 * not in a single-threaded design. That would cause rsyslogd to
+			 * loose input messages - which obviously also would affect
+			 * other selector lines, too. So we do set it to non-blocking and 
+			 * handle the situation ourselfs (by discarding messages). IF we run
+			 * dual-threaded, however, the situation is different: in this case,
+			 * the receivers and the selector line processing are only loosely
+			 * coupled via a memory buffer. Now, I think, we can afford the extra
+			 * wait time. Thus, we enable blocking mode for TCP if we compile with
+			 * pthreads.
+			 * rgerhards, 2005-10-25
+			 */
+#	ifndef USE_PTHREADS
+			/* set to nonblocking - rgerhards 2005-07-20 */
+			fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) | O_NONBLOCK);
+#	endif		
+			if (connect (fd, r->ai_addr, r->ai_addrlen) != 0) {
+				if(errno == EINPROGRESS) {
+					/* this is normal - will complete during select */
+					TCPSendSetStatus(f, TCP_SEND_CONNECTING);
+					return fd;
+				} else {
+					dprintf("create tcp connection failed, reason %s",
+						strerror(errno));
+				}
+
+			}
+			else {
+				TCPSendSetStatus(f, TCP_SEND_READY);
+				return fd;
+			}
+			close(fd);
+		}
+		else {
+			dprintf("couldn't create send socket, reason %s", strerror(errno));
+		}		
+		r = r->ai_next;
+	}
+
+	dprintf("no working socket could be obtained");
+
+	return -1;
+}
+
+/* Sends a TCP message. It is first checked if the
+ * session is open and, if not, it is opened. Then the send
+ * is tried. If it fails, one silent re-try is made. If the send
+ * fails again, an error status (-1) is returned. If all goes well,
+ * 0 is returned. The TCP session is NOT torn down.
+ * For now, EAGAIN is ignored (causing message loss) - but it is
+ * hard to do something intelligent in this case. With this
+ * implementation here, we can not block and/or defer. Things are
+ * probably a bit better when we move to liblogging. The alternative
+ * would be to enhance the current select server with buffering and
+ * write descriptors. This seems not justified, given the expected
+ * short life span of this code (and the unlikeliness of this event).
+ * rgerhards 2005-07-06
+ * This function is now expected to stay. Libloging won't be used for
+ * that purpose. I have added the param "len", because it is known by the
+ * caller and so safes us some time. Also, it MUST be given because there
+ * may be NULs inside msg so that we can not rely on strlen(). Please note
+ * that the restrictions outlined above do not existin in multi-threaded
+ * mode, which we assume will now be most often used. So there is no
+ * real issue with the potential message loss in single-threaded builds.
+ * rgerhards, 2006-11-30
+ * 
+ * In order to support compressed messages via TCP, we must support an
+ * octet-counting based framing (LF may be part of the compressed message).
+ * We are now supporting the same mode that is available in IETF I-D
+ * syslog-transport-tls-05 (current at the time of this writing). This also
+ * eases things when we go ahead and implement that framing. I have now made
+ * available two cases where this framing is used: either by explitely
+ * specifying it in the config file or implicitely when sending a compressed
+ * message. In the later case, compressed and uncompressed messages within
+ * the same session have different framings. If it is explicitely set to
+ * octet-counting, only this framing mode is used within the session.
+ * rgerhards, 2006-12-07
+ */
+static int TCPSend(selector_t *f, char *msg, size_t len)
+{
+	int retry = 0;
+	int done = 0;
+	int bIsCompressed;
+	int lenSend;
+	char *buf = NULL;	/* if this is non-NULL, it MUST be freed before return! */
+	enum TCPSendStatus eState;
+	TCPFRAMINGMODE framingToUse;
+
+	assert(f != NULL);
+	assert(msg != NULL);
+	assert(len > 0);
+
+	bIsCompressed = *msg == 'z';	/* cache this, so that we can modify the message buffer */
+	/* select framing for this record. If we have a compressed record, we always need to
+	 * use octet counting because the data potentially contains all control characters
+	 * including LF.
+	 */
+	framingToUse = bIsCompressed ? TCP_FRAMING_OCTET_COUNTING : f->f_un.f_forw.tcp_framing;
+
+	do { /* try to send message */
+		if(f->f_file <= 0) {
+			/* we need to open the socket first */
+			if((f->f_file = TCPSendCreateSocket(f, f->f_un.f_forw.f_addr)) <= 0) {
+				return -1;
+			}
+		}
+
+		eState = TCPSendGetStatus(f); /* cache info */
+
+		if(eState == TCP_SEND_CONNECTING) {
+			/* In this case, we save the buffer. If we have a
+			 * system with few messages, that hopefully prevents
+			 * message loss at all. However, we make no further attempts,
+			 * just the first message is saved. So we only try this
+			 * if there is not yet a saved message present.
+			 * rgerhards 2005-07-20
+			 */
+			if(f->f_un.f_forw.savedMsg == NULL) {
+				f->f_un.f_forw.savedMsg = malloc(len * sizeof(char));
+				if(f->f_un.f_forw.savedMsg == NULL)
+					return 0; /* nothing we can do... */
+				memcpy(f->f_un.f_forw.savedMsg, msg, len);
+				f->f_un.f_forw.savedMsgLen = len;
+			}
+			return 0;
+		} else if(eState != TCP_SEND_READY)
+			/* This here is debatable. For the time being, we
+			 * accept the loss of a single message (e.g. during
+			 * connection setup in favour of not messing with
+			 * wait time and timeouts. The reason is that such
+			 * things might otherwise cost us considerable message
+			 * loss on the receiving side (even at a timeout set
+			 * to just 1 second).  - rgerhards 2005-07-20
+			 */
+			return 0;
+
+		/* now check if we need to add a line terminator. We need to
+		 * copy the string in memory in this case, this is probably
+		 * quicker than using writev and definitely quicker than doing
+		 * two socket calls.
+		 * rgerhards 2005-07-22
+		 *//*
+		 * Some messages already contain a \n character at the end
+		 * of the message. We append one only if we there is not
+		 * already one. This seems the best fit, though this also
+		 * means the message does not arrive unaltered at the final
+		 * destination. But in the spirit of legacy syslog, this is
+		 * probably the best to do...
+		 * rgerhards 2005-07-20
+		 */
+
+		/* Build frame based on selected framing */
+		if(framingToUse == TCP_FRAMING_OCTET_STUFFING) {
+			if((*(msg+len-1) != '\n')) {
+				if(buf != NULL)
+					free(buf);
+				/* in the malloc below, we need to add 2 to the length. The
+				 * reason is that we a) add one character and b) len does
+				 * not take care of the '\0' byte. Up until today, it was just
+				 * +1 , which caused rsyslogd to sometimes dump core.
+				 * I have added this comment so that the logic is not accidently
+				 * changed again. rgerhards, 2005-10-25
+				 */
+				if((buf = malloc((len + 2) * sizeof(char))) == NULL) {
+					/* extreme mem shortage, try to solve
+					 * as good as we can. No point in calling
+					 * any alarms, they might as well run out
+					 * of memory (the risk is very high, so we
+					 * do NOT risk that). If we have a message of
+					 * more than 1 byte (what I guess), we simply
+					 * overwrite the last character.
+					 * rgerhards 2005-07-22
+					 */
+					if(len > 1) {
+						*(msg+len-1) = '\n';
+					} else {
+						/* we simply can not do anything in
+						 * this case (its an error anyhow...).
+						 */
+					}
+				} else {
+					/* we got memory, so we can copy the message */
+					memcpy(buf, msg, len); /* do not copy '\0' */
+					*(buf+len) = '\n';
+					*(buf+len+1) = '\0';
+					msg = buf; /* use new one */
+					++len; /* care for the \n */
+				}
+			}
+		} else {
+			/* Octect-Counting
+			 * In this case, we need to always allocate a buffer. This is because
+			 * we need to put a header in front of the message text
+			 */
+			char szLenBuf[16];
+			int iLenBuf;
+
+			/* important: the printf-mask is "%d<sp>" because there must be a
+			 * space after the len!
+			 *//* The chairs of the IETF syslog-sec WG have announced that it is
+			 * consensus to do the octet count on the SYSLOG-MSG part only. I am
+			 * now changing the code to reflect this. Hopefully, it will not change
+			 * once again (there can no compatibility layer programmed for this).
+			 * To be on the save side, I just comment the code out. I mark these
+			 * comments with "IETF20061218".
+			 * rgerhards, 2006-12-19
+			 */
+			iLenBuf = snprintf(szLenBuf, sizeof(szLenBuf)/sizeof(char), "%d ", (int) len);
+			/* IETF20061218 iLenBuf =
+			  snprintf(szLenBuf, sizeof(szLenBuf)/sizeof(char), "%d ", len + iLenBuf);*/
+
+			if((buf = malloc((len + iLenBuf) * sizeof(char))) == NULL) {
+			 	/* we are out of memory. This is an extreme situation. We do not
+				 * call any alarm handlers because they most likely run out of mem,
+				 * too. We are brave enough to call debug output, though. Other than
+				 * that, there is nothing left to do. We can not sent the message (as
+				 * in case of the other framing, because the message is incomplete.
+				 * We could, however, send two chunks (header and text separate), but
+				 * that would cause a lot of complexity in the code. So we think it
+				 * is appropriate enough to just make sure we do not crash in this
+				 * very unlikely case. For this, it is justified just to loose
+				 * the message. Rgerhards, 2006-12-07
+				 */
+				 dprintf("Error: out of memory when building TCP octet-counted "
+				         "frame. Message is lost, trying to continue.\n");
+				return 0;
+			}
+
+			 memcpy(buf, szLenBuf, iLenBuf); /* header */
+			 memcpy(buf + iLenBuf, msg, len); /* message */
+			 len += iLenBuf;	/* new message size */
+			 msg = buf;	/* set message buffer */
+		}
+
+		/* frame building complete, on to actual sending */
+
+		lenSend = send(f->f_file, msg, len, 0);
+		dprintf("TCP sent %d bytes, requested %d, msg: '%s'\n", lenSend, len,
+			bIsCompressed ? "***compressed***" : msg);
+		if((unsigned)lenSend == len) {
+			/* all well */
+			if(buf != NULL) {
+				free(buf);
+			}
+			return 0;
+		} else if(lenSend != -1) {
+			/* no real error, could "just" not send everything... 
+			 * For the time being, we ignore this...
+			 * rgerhards, 2005-10-25
+			 */
+			dprintf("message not completely (tcp)send, ignoring %d\n", lenSend);
+#			if USE_PTHREADS
+			usleep(1000); /* experimental - might be benefitial in this situation */
+#			endif
+			if(buf != NULL)
+				free(buf);
+			return 0;
+		}
+
+		switch(errno) {
+		case EMSGSIZE:
+			dprintf("message not (tcp)send, too large\n");
+			/* This is not a real error, so it is not flagged as one */
+			if(buf != NULL)
+				free(buf);
+			return 0;
+			break;
+		case EINPROGRESS:
+		case EAGAIN:
+			dprintf("message not (tcp)send, would block\n");
+#			if USE_PTHREADS
+			usleep(1000); /* experimental - might be benefitial in this situation */
+#			endif
+			/* we loose this message, but that's better than loosing
+			 * all ;)
+			 */
+			/* This is not a real error, so it is not flagged as one */
+			if(buf != NULL)
+				free(buf);
+			return 0;
+			break;
+		default:
+			dprintf("message not (tcp)send");
+			break;
+		}
+	
+		if(retry == 0) {
+			++retry;
+			/* try to recover */
+			close(f->f_file);
+			TCPSendSetStatus(f, TCP_SEND_NOTCONNECTED);
+			f->f_file = -1;
+		} else {
+			if(buf != NULL)
+				free(buf);
+			return -1;
+		}
+	} while(!done); /* warning: do ... while() */
+	/*NOT REACHED*/
+
+	if(buf != NULL)
+		free(buf);
+	return -1; /* only to avoid compiler warning! */
+}
+
+
 /* get the syslog forward port from selector_t. The passed in
  * struct must be one that is setup for forwarding.
  * rgerhards, 2007-06-28