Re: [runit/svlogd] RFE: periodic log-rotation; filter-malfunction

[Gerrit Pape <pape@smarden.org>]

[-- Attachment #1: Type: text/plain, Size: 938 bytes --]

On Fri, Sep 03, 2004 at 11:08:01AM +0000, Gerrit Pape wrote:
> On Thu, Sep 02, 2004 at 11:58:21PM +0200, Enrico Scholz wrote:
> > pape@smarden.org (Gerrit Pape) writes:
> > >> Another issue: the '-/+...' filters seem to fail sometimes.
> > > ...
> > > This shouldn't happen.  Can you please give an exact example of a line
> > > that ends up in the wrong log, copied and pasted, and also re-check that
> > > you post the exact contents of the config files?  Do you use the -r or
> > > -R command line options to svlogd in socklog-ucspi-tcp/log/run?
> > Please see http://www.tu-chemnitz.de/~ensc/runit/badfilter.txt
> 
> Thanks for the detailed and complete report.  I've identified the
> problem in svlogd, and'll try to come up with a patch.

Here's a patch that should fix it, and additionally some other issues I
spotted while re-reading svlogd.c.  If you're going to test it, please
report problems or success here.

Thanks, Gerrit.

[-- Attachment #2: diff0 --]
[-- Type: text/plain, Size: 15878 bytes --]

Index: src/svlogd.c
===================================================================
RCS file: /var/lib/cvs/runit/src/svlogd.c,v
retrieving revision 1.12
diff -u -r1.12 svlogd.c
--- src/svlogd.c	28 Mar 2004 18:17:14 -0000	1.12
+++ src/svlogd.c	11 Sep 2004 09:42:46 -0000
@@ -7,7 +7,6 @@
 #include <sys/time.h>
 #include <dirent.h>
 #include <unistd.h>
-#include <dirent.h>
 #include <stdio.h>
 #include "pmatch.h"
 #include "fmt_ptime.h"
@@ -45,8 +44,10 @@
 
 unsigned int verbose =0;
 unsigned int timestamp =0;
-unsigned long linelen =1000;
+unsigned long linemax =1000;
 unsigned long buflen =1024;
+unsigned long linelen;
+
 const char *replace ="";
 char repl =0;
 
@@ -62,6 +63,9 @@
 char *line;
 char stamp[FMT_PTIME];
 unsigned int exitasap =0;
+unsigned int rotateasap =0;
+unsigned int reopenasap =0;
+unsigned int linecomplete =1;
 int fdudp =-1;
 
 struct logdir {
@@ -78,7 +82,7 @@
   char *name;
   int fdcur;
   int fdlock;
-  unsigned int match;
+  char match;
   struct sockaddr_in udpaddr;
   unsigned int udponly;
 } *dir;
@@ -125,6 +129,8 @@
     sig_unblock(sig_alarm);
     sig_unblock(sig_hangup);
     
+    if (verbose)
+      strerr_warn5(INFO, "processing: ", ld->name, "/", ld->fnsave, 0);
     if ((fd =open_read(ld->fnsave)) == -1)
       fatal2("unable to open input for processor", ld->name);
     if (fd_move(0, fd) == -1)
@@ -160,7 +166,7 @@
     fatal2("unable to run processor", ld->name);
   }
   ld->ppid =pid;
-  return(1);  
+  return(1);
 }
 unsigned int processorstop(struct logdir *ld) {
   char f[28];
@@ -172,14 +178,15 @@
     sig_block(sig_hangup);
     ld->ppid =0;
   }
+  if (ld->fddir == -1) return(1);
   while (fchdir(ld->fddir) == -1)
     pause2("unable to change directory, want processor", ld->name);
   if (wait_exitcode(wstat) != 0) {
-    warnx("processor crashed, restart", ld->name);
+    warnx("processor failed, restart", ld->name);
     ld->fnsave[26] ='t';
     unlink(ld->fnsave);
     ld->fnsave[26] ='u';
-    if (ld->processor.len) processorstart(ld);
+    processorstart(ld);
     while (fchdir(fdwdir) == -1)
       pause1("unable to change to initial working directory");
     return(ld->processor.len ? 0 : 1);
@@ -195,6 +202,7 @@
     pause2("unable to set mode of processed", ld->name);
   while (rename("newstate", "state") == -1)
     pause2("unable to rename state", ld->name);
+  if (verbose) strerr_warn5(INFO, "processed: ", ld->name, "/", f, 0);
   while (fchdir(fdwdir) == -1)
     pause1("unable to change to initial working directory");
   return(1);
@@ -207,6 +215,7 @@
   char tmp[FMT_ULONG +1];
   char oldest[FMT_PTIME];
 
+  if (ld->fddir == -1) return(0);
   if (ld->size <= 0) return(1);
   if (ld->ppid) while(! processorstop(ld));
 
@@ -245,11 +254,11 @@
   ld->size =0;
   while (fchmod(ld->fdcur, 0644) == -1)
     pause2("unable to set mode of current", ld->name);
-  buffer_init(&ld->b, buffer_unixwrite, ld->fdcur, ld->btmp, sizeof ld->btmp);
 
-  oldest[0] ='A'; oldest[1] =oldest[27] =0; errno =0;
+  oldest[0] ='A'; oldest[1] =oldest[27] =0;
   while (! (d =opendir(".")))
     pause2("unable to open directory, want rotate", ld->name);
+  errno =0;
   while ((f =readdir(d)))
     if ((f->d_name[0] == '@') && (str_len(f->d_name) == 27)) {
       ++n;
@@ -264,14 +273,29 @@
       warn2("unable to unlink oldest logfile", ld->name);
   }
 
-  if (ld->processor.len) {
-    processorstart(ld);
-  }
+  processorstart(ld);
   while (fchdir(fdwdir) == -1)
     pause1("unable to change to initial working directory");
   return(1);
 }
 
+int buffer_pwrite(int n, char *s, unsigned int len) {
+  int i;
+
+  if ((dir +n)->sizemax) {
+    if ((dir +n)->size >= (dir +n)->sizemax) rotate(dir +n);
+    if (len > ((dir +n)->sizemax -(dir +n)->size))
+      len =(dir +n)->sizemax -(dir +n)->size;
+  }
+  while ((i =write((dir +n)->fdcur, s, len)) == -1)
+    pause2("unable to write to current", (dir +n)->name);
+  (dir +n)->size +=i;
+  if ((dir +n)->sizemax)
+    if (s[i -1] == '\n')
+      if ((dir +n)->size >= ((dir +n)->sizemax -linemax)) rotate(dir +n);
+  return(i);
+}
+
 void logdir_close(struct logdir *ld) {
   if (ld->fddir == -1) return;
   if (verbose) strerr_warn3(INFO, "close: ", ld->name, 0);
@@ -288,6 +312,7 @@
   if (ld->fdlock == -1) return; /* impossible */
   close(ld->fdlock);
   ld->fdlock =-1;
+  while (! stralloc_copys(&ld->processor, "")) pause_nomem();
 }
 
 /* taken from libdjbdns */
@@ -335,7 +360,8 @@
   ld->sizemax =1000000;
   ld->nmax =10;
   ld->name =(char*)fn;
-  ld->match =0;
+  ld->ppid =0;
+  ld->match ='+';
   ld->udpaddr.sin_port =0;
   ld->udponly =0;
   while (! stralloc_copys(&ld->inst, "")) pause_nomem();
@@ -367,9 +393,6 @@
 	break;
       case 's':
 	scan_ulong(&sa.s[i +1], &ld->sizemax);
-	/*
-	  if (ld->sizemax && (ld->sizemax < linelen)) ld->sizemax =2 *linelen;
-	*/
 	break;
       case 'n':
 	scan_ulong(&sa.s[i +1], &ld->nmax);
@@ -421,6 +444,7 @@
       } while ((stat(ld->fnsave, &st) != -1) || (errno != error_noent));
       while (rename("current", ld->fnsave) == -1)
 	pause2("unable to rename current", ld->name);
+      i =-1;
     }
     else
       ld->size =st.st_size;
@@ -438,7 +462,7 @@
   coe(ld->fdcur);
   while (fchmod(ld->fdcur, 0644) == -1)
     pause2("unable to set mode of current", ld->name);
-  buffer_init(&ld->b, buffer_unixwrite, ld->fdcur, ld->btmp, sizeof ld->btmp);
+  buffer_init(&ld->b, buffer_pwrite, ld -dir, ld->btmp, buflen);
   
   if (verbose) {
     if (i == 0) strerr_warn4(INFO, "append: ", ld->name, "/current", 0);
@@ -461,121 +485,43 @@
   if (! ok) fatalx("no functional log directories.");
 }
 
-unsigned int linestart(struct logdir *ld, char *s, int len) {
-  int i;
-
-  /* check inst, set match */
-  ld->match ='+';
-  if (ld->inst.len) {
-    for (i =0; i < ld->inst.len; ++i) {
-      switch(ld->inst.s[i]) {
-      case '+':
-      case '-':
-	if (pmatch(&ld->inst.s[i +1], s, len))
-	  ld->match =ld->inst.s[i];
-	break;
-      case 'e':
-	if (pmatch(&ld->inst.s[i +1], s, len)) {
-	  if (timestamp) buffer_puts(buffer_2, stamp);
-	  buffer_put(buffer_2, s, len);
-	  if (len == linelen) buffer_puts(buffer_2, "...");
-	  buffer_putflush(buffer_2, "\n", 1);
-	}
-	break;
-      case 'E':
-	if (! pmatch(&ld->inst.s[i +1], s, len)) {
-	  if (timestamp) buffer_puts(buffer_2, stamp);
-	  buffer_put(buffer_2, s, len);
-	  if (len == linelen) buffer_puts(buffer_2, "...");
-	  buffer_putflush(buffer_2, "\n", 1);
-	}
-	break;
-      }
-      i +=byte_chr(&ld->inst.s[i], ld->inst.len -i, 0);
-    }
-  }
-  if (ld->match == '-') return(0);
-  if (! ld->udponly) {
-    if (timestamp) {
-      buffer_puts(&ld->b, stamp);
-      if (timestamp != 3) ld->size +=26;
-      else ld->size +=20;
-    }
-    buffer_put(&ld->b, s, len);
-    ld->size +=len;
-  }
-  if (ld->udpaddr.sin_port != 0) {
-    if (fdudp == -1) {
-      buffer_puts(&ld->b, "warning: no udp socket available: ");
-      buffer_put(&ld->b, s, len);
-      buffer_putflush(&ld->b, "\n", 1);
-    }
-    else {
-      if (len >= linelen -1) {
-	s[linelen -4] =s[linelen -3] =s[linelen -2] ='.';
-	len =linelen -1;
-      }
-      if (s[len -1] != '\n') s[len++] ='\n';
-      if (sendto(fdudp, s, len, 0, (struct sockaddr *)&ld->udpaddr,
-		 sizeof(ld->udpaddr)) != len) {
-	buffer_puts(&ld->b, "warning: failure sending through udp: ");
-	buffer_put(&ld->b, s, len);
-	buffer_putflush(&ld->b, "\n", 1);
-      }
-    }
-  }
-  return(1);
-}
-unsigned int lineadd(struct logdir *ld, char *s, int len) {
-  if ((ld->match != '+') || ld->udponly) return(0);
-  buffer_put(&ld->b, s, len);
-  ld->size +=len;
-  /*  if (ld->sizemax && (ld->size >= ld->sizemax)) rotate(ld); */
-  return(1);
-}
-unsigned int lineflush(struct logdir *ld, char *s, int len) {
-  switch(ld->match) {
-  case '-':
-    ld->match =0;
-    return(0);
-  case 0:
-    linestart(ld, s, len);
-    break;
-  case '+':
-    if (ld->udponly) {
-      ld->match =0;
-      return(0);
-    }
-    buffer_put(&ld->b, s, len);
-    ld->size +=len;
-    break;
-  }
-  if (ld->match == '+') {
-    buffer_putflush(&ld->b, "\n", 1);
-    ld->size +=1;
-    ld->match =0;
-    if (ld->sizemax)
-      if ((linelen > ld->sizemax) || (ld->size >= (ld->sizemax -linelen)))
-	rotate(ld);
-    return(1);
-  }
-  ld->match =0;
-  return(0);
-}
 int buffer_pread(int fd, char *s, unsigned int len) {
-  int rc;
+  int i;
 
-  rc =read(fd, s, len);
-  if ((rc == -1) && (errno == error_intr)) return(0);
-  return(rc);
+  for (i =0; i < dirn; ++i) buffer_flush(&dir[i].b);
+  if (rotateasap) {
+    for (i =0; i < dirn; ++i) rotate(dir +i);
+    rotateasap =0;
+  }
+  if (exitasap) {
+    if (linecomplete) return(0);
+    len =1;
+  }
+  if (reopenasap) {
+    logdirs_reopen();
+    reopenasap =0;
+  }
+  sig_unblock(sig_term);
+  sig_unblock(sig_child);
+  sig_unblock(sig_alarm);
+  sig_unblock(sig_hangup);
+  i =read(fd, s, len);
+  sig_block(sig_term);
+  sig_block(sig_child);
+  sig_block(sig_alarm);
+  sig_block(sig_hangup);
+  if (i == -1) if (errno != error_intr) warn("unable to read standard input");
+  if (i > 0) linecomplete =(s[i -1] == '\n');
+  return(i);
 }
 void sig_term_handler(void) {
-  exitasap =1;
   if (verbose) strerr_warn2(INFO, "sigterm received.", 0);
+  exitasap =1;
 }
 void sig_child_handler(void) {
   int pid, l;
 
+  if (verbose) strerr_warn2(INFO, "sigchild received.", 0);
   while ((pid =wait_nohang(&wstat)) > 0)
     for (l =0; l < dirn; ++l)
       if (dir[l].ppid == pid) {
@@ -583,26 +529,50 @@
 	processorstop(&dir[l]);
 	break;
       }
-  if (verbose) strerr_warn2(INFO, "sigchild received.", 0);
 }
 void sig_alarm_handler(void) {
-  int l;
-  
-  for (l =0; l < dirn; ++l)
-    if (dir[l].fddir != -1)
-      if (dir[l].size > 0)
-	rotate(&dir[l]);
   if (verbose) strerr_warn2(INFO, "sigalarm received.", 0);
+  rotateasap =1;
 }
 void sig_hangup_handler(void) {
-  logdirs_reopen();
   if (verbose) strerr_warn2(INFO, "sighangup received.", 0);
+  reopenasap =1;
 }
 
+void logmatch(struct logdir *ld) {
+  int i;
+
+  ld->match ='+';
+  for (i =0; i < ld->inst.len; ++i) {
+    switch(ld->inst.s[i]) {
+    case '+':
+    case '-':
+      if (pmatch(&ld->inst.s[i +1], line, linelen))
+	ld->match =ld->inst.s[i];
+      break;
+    case 'e':
+      if (pmatch(&ld->inst.s[i +1], line, linelen)) {
+	if (timestamp) buffer_puts(buffer_2, stamp);
+	buffer_put(buffer_2, line, linelen);
+	if (linelen == linemax) buffer_puts(buffer_2, "...");
+	buffer_put(buffer_2, "\n", 1); buffer_flush(buffer_2);
+      }
+      break;
+    case 'E':
+      if (! pmatch(&ld->inst.s[i +1], line, linelen)) {
+	if (timestamp) buffer_puts(buffer_2, stamp);
+	buffer_put(buffer_2, line, linelen);
+	if (linelen == linemax) buffer_puts(buffer_2, "...");
+	buffer_put(buffer_2, "\n", 1); buffer_flush(buffer_2);
+      }
+      break;
+    }
+    i +=byte_chr(&ld->inst.s[i], ld->inst.len -i, 0);
+  }
+}
 int main(int argc, const char **argv) {
   int i;
   int opt;
-  unsigned int eol;
 
   progname =*argv;
 
@@ -617,8 +587,8 @@
       if (! repl || *(optarg +1)) usage();
       break;
     case 'l':
-      scan_ulong(optarg, &linelen);
-      if (linelen == 0) linelen =1000;
+      scan_ulong(optarg, &linemax);
+      if (linemax == 0) linemax =1000;
       break;
     case 'b':
       scan_ulong(optarg, &buflen);
@@ -638,7 +608,7 @@
 
   dirn =argc -optind;
   if (dirn <= 0) usage();
-  if (buflen <= linelen) usage();
+  if (buflen <= linemax) usage();
   if ((fdwdir =open_read(".")) == -1)
     fatal("unable to open current working directory");
   coe(fdwdir);
@@ -653,7 +623,7 @@
   databuf =(char*)alloc(buflen *sizeof(char));
   if (! databuf) die_nomem();
   buffer_init(&data, buffer_pread, 0, databuf, buflen);
-  line =(char*)alloc(linelen *sizeof(char));
+  line =(char*)alloc(linemax *sizeof(char));
   if (! line) die_nomem();
   fndir =argv;
 
@@ -668,79 +638,101 @@
 
   logdirs_reopen();
 
-  for(eol =0;;) {
-    int r, len;
-    char *ch;
-
-    if (exitasap && ! data.p) break; /* buffer is empty */
-
-    sig_unblock(sig_term);
-    sig_unblock(sig_child);
-    sig_unblock(sig_alarm);
-    sig_unblock(sig_hangup);
-    errno =0;
-    r =buffer_feed(&data);
-    sig_block(sig_term);
-    sig_block(sig_child);
-    sig_block(sig_alarm);
-    sig_block(sig_hangup);
+  for(;;) {
+    char ch;
 
-    if (r == -1) {
-      warn("unable to read standard input");
-      continue;
-    }
-    if (r == 0) {
-      if (errno == error_intr) continue;
-      break; /* eof */
-    }
-    if (r > linelen) r =linelen;
-    if (timestamp && (eol == 0)) {
-      taia_now(&now);
-      switch (timestamp) {
-      case 1:
-	stamp[fmt_taia(stamp, &now)] =' ';
-	stamp[26] =0;
-	break;
-      case 2:
-	stamp[fmt_ptime(stamp, &now)] =' ';
-	stamp[26] =0;
-	break;
-      case 3:
-	stamp[fmt_ptime(stamp, &now)] =0;
-	stamp[19] =' '; stamp[20] =0;
+    linelen =0;
+    if (exitasap && ! data.p) break; /* data buffer is empty */
+    for (linelen =0; linelen < linemax; ++linelen) {
+      if (buffer_GETC(&data, &ch) <= 0) {
+	exitasap =1;
 	break;
       }
-    }
-    ch =buffer_peek(&data);
-    for (len =0; len < r; ++len, ++ch) {
-      if (*ch == '\n') {
-	eol =2;
-	break;
+      if (! linelen && timestamp) {
+	taia_now(&now);
+	switch (timestamp) {
+	case 1:
+	  stamp[fmt_taia(stamp, &now)] =' ';
+	  stamp[26] =0;
+	  break;
+	case 2:
+	  stamp[fmt_ptime(stamp, &now)] =' ';
+	  stamp[26] =0;
+	  break;
+	case 3:
+	  stamp[fmt_ptime(stamp, &now)] =0;
+	  stamp[19] =' '; stamp[20] =0;
+	  break;
+	}
       }
-      line[len] =*ch;
-      
+      if (ch == '\n') break;
       if (repl) {
-	if ((line[len] < 33) || (line[len] > 126))
-	  line[len] =repl;
+	if ((ch < 32) || (ch > 126))
+	  ch =repl;
 	else
 	  for (i =0; replace[i]; ++i)
-	    if (line[len] == replace[i]) {
-	      line[len] =repl;
+	    if (ch == replace[i]) {
+	      ch =repl;
 	      break;
 	    }
       }
+      line[linelen] =ch;
     }
-    buffer_seek(&data, len);
+    if (! linelen) continue;
+    for (i =0; i < dirn; ++i)
+      if (dir[i].fddir != -1) {
+	if (dir[i].inst.len) logmatch(&dir[i]);
+	if (dir[i].match != '+') continue;
+	if (! dir[i].udponly) {
+	  if (timestamp) buffer_puts(&dir[i].b, stamp);
+	  buffer_put(&dir[i].b, line, linelen);
+	}
+	if (dir[i].udpaddr.sin_port != 0) {
+	  if (fdudp == -1) {
+	    buffer_puts(&dir[i].b, "warning: no udp socket available: ");
+	    buffer_put(&dir[i].b, line, linelen);
+	    buffer_put(&dir[i].b, "\n", 1);
+	    buffer_flush(&dir[i].b);
+	  }
+	  else {
+	    if (linelen >= linemax -1) {
+	      line[linemax -4] =line[linemax -3] =line[linemax -2] ='.';
+	      linelen =linemax -1;
+	    }
+	    if (line[linelen -1] != '\n') line[linelen++] ='\n';
+	    if (sendto(fdudp, line, linelen, 0,
+		       (struct sockaddr *)&dir[i].udpaddr,
+		       sizeof(dir[i].udpaddr)) != linelen) {
+	      buffer_puts(&dir[i].b, "warning: failure sending through udp: ");
+	      buffer_put(&dir[i].b, line, linelen);
+	      buffer_put(&dir[i].b, "\n", 1);
+	      buffer_flush(&dir[i].b);
+	    }
+	  }
+	}
+      }
+    if (linelen == linemax)
+      for (;;) {
+	if (buffer_GETC(&data, &ch) <= 0) {
+	  exitasap =1;
+	  break;
+	}
+	if (ch == '\n') break;
+	for (i =0; i < dirn; ++i)
+	  if (dir[i].fddir != -1) {
+	    if (dir[i].match != '+') continue;
+	    if (! dir[i].udponly) buffer_PUTC(&dir[i].b, ch);
+	  }
+      }
     for (i =0; i < dirn; ++i)
       if (dir[i].fddir != -1) {
-	switch(eol) {
-	case 0: linestart(&dir[i], line, len); break;
-	case 1: lineadd(&dir[i], line, len); break;
-	case 2: lineflush(&dir[i], line, len); break;
+	if (dir[i].match != '+') continue;
+	if (! dir[i].udponly) {
+	  ch ='\n';
+	  buffer_PUTC(&dir[i].b, ch);
+	  buffer_flush(&dir[i].b);
 	}
       }
-    if (eol == 0) eol =1;
-    if (eol == 2) { eol =0; buffer_seek(&data, 1); }
   }
   
   for (i =0; i < dirn; ++i) {