Merge branch 'v5-stable' into v5-stable-newstats

Conflicts:
	configure.ac

8 files changed, 238 insertions, 10 deletions

diff --git a/ChangeLog b/ChangeLog
index b9e56ec2..245d7110 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -7,7 +7,17 @@
   $MainMsgQueueLightDelayMark, $ActionQueueLightDelayMark; both
   specify number of messages starting at which a delay happens.
 ---------------------------------------------------------------------------
-Version 5.8.9  [V5-stable] 2012-03-??
+Version 5.8.10  [V5-stable] 2012-??-??
+- bugfix: segfault on startup if $actionqueuefilename was missing for disk
+  queue config
+  Thanks to Tomas Heinrich for the patch.
+- bugfix: segfault if disk-queue was started up with old queue file
+  Thanks to Tomas Heinrich for the patch.
+- bugfix: memory leak in array passing output module mode
+---------------------------------------------------------------------------
+Version 5.8.9  [V5-stable] 2012-03-15
+- added tool to recover disk queue if .qi file is missing (recover_qi.pl)
+  Thanks to Kaiwang Chen for contributing this tool
 - bugfix: stopped DA queue was never processed after a restart due to a
   regression from statistics module
 - added better doc for statsobj interface
diff --git a/action.c b/action.c
index 7de10d49..1164fd66 100644
--- a/action.c
+++ b/action.c
@@ -826,8 +826,9 @@ static rsRetVal releaseBatch(action_t *pAction, batch_t *pBatch)
 					if(((uchar**)ppMsgs)[j] != NULL) {
 						jArr = 0;
 						while(ppMsgs[j][jArr] != NULL) {
-							d_free(ppMsgs[j][jArr++]);
-							ppMsgs[j][jArr++] = NULL;
+							d_free(ppMsgs[j][jArr]);
+							ppMsgs[j][jArr] = NULL;
+							++jArr;
 						}
 						d_free(((uchar**)ppMsgs)[j]);
 						((uchar**)ppMsgs)[j] = NULL;
diff --git a/configure.ac b/configure.ac
index 1200294b..b626da66 100644
--- a/configure.ac
+++ b/configure.ac
@@ -2,7 +2,7 @@
 # Process this file with autoconf to produce a configure script.
 
 AC_PREREQ(2.61)
-AC_INIT([rsyslog],[5.8.8-newstats11],[rsyslog@lists.adiscon.com])
+AC_INIT([rsyslog],[5.8.9-newstats12],[rsyslog@lists.adiscon.com])
 AM_INIT_AUTOMAKE
 
 m4_ifdef([AM_SILENT_RULES], [AM_SILENT_RULES([yes])])
diff --git a/doc/manual.html b/doc/manual.html
index 54a04f80..690143de 100644
--- a/doc/manual.html
+++ b/doc/manual.html
@@ -19,7 +19,7 @@ rsyslog support</a> available directly from the source!</p>
 <p><b>Please visit the <a href="http://www.rsyslog.com/sponsors">rsyslog sponsor's page</a>
 to honor the project sponsors or become one yourself!</b> We are very grateful for any help towards the
 project goals.</p>
-<p><b>This documentation is for version 5.8.8 (v5-stable branch) of rsyslog.</b>
+<p><b>This documentation is for version 5.8.9 (v5-stable branch) of rsyslog.</b>
 Visit the <i><a href="http://www.rsyslog.com/status">rsyslog status page</a></i></b>
 to obtain current version information and project status.
 </p><p><b>If you like rsyslog, you might
diff --git a/doc/v5compatibility.html b/doc/v5compatibility.html
index 6d60062f..cc875b7b 100644
--- a/doc/v5compatibility.html
+++ b/doc/v5compatibility.html
@@ -16,6 +16,11 @@ available. This processing was redundant and had a lot a drawbacks.
 For details, please see the
 <a href="v4compatibility.html">rsyslog v4 compatibility notes</a> which elaborate
 on the reasons and the (few) things you may need to change.
+<h2>Queue on-disk format</h2>
+<p>The queue information file format has been changed. When upgrading from v4 to
+v5, make sure that the queue is emptied and no on-disk structure present. We did
+not go great length in understanding the old format, as there was too little demand
+for that (and it being quite some effort if done right).
 <h2>Queue Worker Thread Shutdown</h2>
 <p>Previous rsyslog versions had the capability to &quot;run&quot; on zero queue worker
 if no work was required. This was done to save a very limited number of resources. However,
diff --git a/runtime/queue.c b/runtime/queue.c
index 0b717289..5503f2d9 100644
--- a/runtime/queue.c
+++ b/runtime/queue.c
@@ -2026,13 +2026,16 @@ static rsRetVal qqueuePersist(qqueue_t *pThis, int bIsCheckpoint)
 	CHKiRet(obj.EndSerialize(psQIF));
 
 	/* now persist the stream info */
-	CHKiRet(strm.Serialize(pThis->tVars.disk.pWrite, psQIF));
-	CHKiRet(strm.Serialize(pThis->tVars.disk.pReadDel, psQIF));
+	if(pThis->tVars.disk.pWrite != NULL)
+		CHKiRet(strm.Serialize(pThis->tVars.disk.pWrite, psQIF));
+	if(pThis->tVars.disk.pReadDel != NULL)
+		CHKiRet(strm.Serialize(pThis->tVars.disk.pReadDel, psQIF));
 	
 	/* tell the input file object that it must not delete the file on close if the queue
 	 * is non-empty - but only if we are not during a simple checkpoint
 	 */
-	if(bIsCheckpoint != QUEUE_CHECKPOINT) {
+	if(bIsCheckpoint != QUEUE_CHECKPOINT
+	   && pThis->tVars.disk.pReadDel != NULL) {
 		CHKiRet(strm.SetbDeleteOnClose(pThis->tVars.disk.pReadDel, 0));
 	}
 
@@ -2124,7 +2127,8 @@ CODESTARTobjDestruct(qqueue)
 	 * direct queue - because in both cases we have none... ;)
 	 * with a child! -- rgerhards, 2008-01-28
 	 */
-	if(pThis->qType != QUEUETYPE_DIRECT && !pThis->bEnqOnly && pThis->pqParent == NULL)
+	if(pThis->qType != QUEUETYPE_DIRECT && !pThis->bEnqOnly && pThis->pqParent == NULL
+	   && pThis->pWtpReg != NULL)
 		ShutdownWorkers(pThis);
 
 	if(pThis->bIsDA && getPhysicalQueueSize(pThis) > 0 && pThis->bSaveOnShutdown) {
diff --git a/tools/Makefile.am b/tools/Makefile.am
index 96657ad4..962ae507 100644
--- a/tools/Makefile.am
+++ b/tools/Makefile.am
@@ -47,4 +47,5 @@ zpipe_LDADD = -lz
 msggen_SOURCES = msggen.c
 endif
 
-EXTRA_DIST = $(man_MANS)
+EXTRA_DIST = $(man_MANS) \
+	recover_qi.pl
diff --git a/tools/recover_qi.pl b/tools/recover_qi.pl
new file mode 100644
index 00000000..4e2cf9d5
--- /dev/null
+++ b/tools/recover_qi.pl
@@ -0,0 +1,207 @@
+#!/usr/bin/perl -w

+# recover rsyslog disk queue index (.qi) from queue files (.nnnnnnnn).

+#

+# See:

+#   runtime/queue.c: qqueuePersist()

+#   runtime/queue.c: qqueueTryLoadPersistedInfo()

+#

+# kaiwang.chen@gmail.com 2012-03-14

+#

+use strict;

+use Getopt::Long;

+

+my %opt = ();

+GetOptions(\%opt,"spool|w=s","basename|f=s","digits|d=i","help!");

+if ($opt{help}) {

+  print "Usage:

+\t$0 -w WorkDirectory -f QueueFileName -d 8 > QueueFileName.qi

+";

+  exit;

+}

+

+# runtime/queue.c: qConstructDisk()

+my $iMaxFiles = 10000000; # 0+"1".( "0"x($opt{digits} - 1));

+

+# get the list of queue files, spool directory excluded

+my $re = qr/^\Q$opt{basename}\E\.\d{$opt{digits}}$/;

+opendir(DIR, $opt{spool}) or die "can’t open spool: $!";

+my @qf = grep { /$re/ && -f "$opt{spool}/$_" } readdir(DIR);

+closedir DIR;

+

+# ensure order and continuity

+@qf = sort @qf;

+my ($head) = ($qf[0] =~ /(\d+)$/);

+my ($tail) = ($qf[-1] =~ /(\d+)$/);

+$head += 0;

+$tail += 0;

+if ($tail-$head+1 != @qf || $tail > $iMaxFiles) {

+  die "broken queue: missing file(s) or wrong tail\n";

+}

+

+# collect some counters about the queue, assuming all are unprocessed entries.

+my $sizeOnDisk = 0;

+my $iQueueSize = 0;

+chdir($opt{spool}) or die "can't chdir to spool: $!";

+print STDERR "traversing ". @qf ." files, please wait...\n";

+for (@qf) {

+  open FH, "<", $_ or die "can't read queue file $_\n";

+  $sizeOnDisk += (stat FH)[7];

+  while (<FH>) {

+    $iQueueSize++ if /^<Obj/;  # runtime/msg.c: MsgSerialize()

+  }

+  close FH;

+}

+# happen to reuse last stat

+my $iCurrOffs_Write = (stat(_))[7];

+

+# runtime/queue.c: qqueuePersist()

+my $qqueue = Rsyslog::OPB->new("qqueue",1);

+$qqueue->property("iQueueSize", "INT", $iQueueSize);

+$qqueue->property("tVars.disk.sizeOnDisk", "INT64", $sizeOnDisk);

+$qqueue->property("tVars.disk.bytesRead", "INT64", 0);

+

+# runtime/stream.h: strmType_t

+my $STREAMTYPE_FILE_CIRCULAR = 1;

+# runtime/stream.h: strmMode_t

+my $STREAMMODE_READ = 1;

+my $STREAMMODE_WRITE_APPEND = 4;

+

+# runtime/stream.c: strmSerialize()

+# write to end

+my $strm_Write = Rsyslog::Obj->new("strm",1);

+$strm_Write->property(      "iCurrFNum",  "INT",                     $tail);

+$strm_Write->property(       "pszFName",  "PSZ",            $opt{basename});

+$strm_Write->property(      "iMaxFiles",  "INT",                $iMaxFiles);

+$strm_Write->property( "bDeleteOnClose",  "INT",                         0);

+$strm_Write->property(          "sType",  "INT", $STREAMTYPE_FILE_CIRCULAR);

+$strm_Write->property("tOperationsMode",  "INT",  $STREAMMODE_WRITE_APPEND);

+$strm_Write->property(      "tOpenMode",  "INT",                      0600);

+$strm_Write->property(      "iCurrOffs","INT64",          $iCurrOffs_Write);

+# read from head

+my $strm_ReadDel = Rsyslog::Obj->new("strm",1);

+$strm_ReadDel->property(      "iCurrFNum",  "INT",                     $head);

+$strm_ReadDel->property(       "pszFName",  "PSZ",            $opt{basename});

+$strm_ReadDel->property(      "iMaxFiles",  "INT",                $iMaxFiles);

+$strm_ReadDel->property( "bDeleteOnClose",  "INT",                         1);

+$strm_ReadDel->property(          "sType",  "INT", $STREAMTYPE_FILE_CIRCULAR);

+$strm_ReadDel->property("tOperationsMode",  "INT",          $STREAMMODE_READ);

+$strm_ReadDel->property(      "tOpenMode",  "INT",                      0600);

+$strm_ReadDel->property(      "iCurrOffs","INT64",                         0);

+

+# .qi

+print $qqueue->serialize();

+print $strm_Write->serialize();

+print $strm_ReadDel->serialize();

+

+exit;

+#-----------------------------------------------------------------------------

+

+package Rsyslog::Serializable;

+# runtime/obj.c 

+sub COOKIE_OBJLINE   { '<' }

+sub COOKIE_PROPLINE  { '+' }

+sub COOKIE_ENDLINE   { '>' }

+sub COOKIE_BLANKLINE { '.' }

+# VARTYPE(short_ptype)

+sub VARTYPE {

+  my ($t) = @_;

+  # runtime/obj-types.h: propType_t

+  my $ptype = "PROPTYPE_".$t;

+  # runtime/var.h: varType_t

+  my %vm = ( 

+    VARTYPE_NONE       => 0,

+    VARTYPE_STR        => 1,

+    VARTYPE_NUMBER     => 2,

+    VARTYPE_SYSLOGTIME => 3,

+  );

+  # runtime/obj.c: SerializeProp()

+  my %p2v = (

+    #PROPTYPE_NONE  => "",

+    PROPTYPE_PSZ   => "VARTYPE_STR",

+    PROPTYPE_SHORT => "VARTYPE_NUMBER",

+    PROPTYPE_INT   => "VARTYPE_NUMBER",

+    PROPTYPE_LONG  => "VARTYPE_NUMBER",

+    PROPTYPE_INT64 => "VARTYPE_NUMBER",

+    PROPTYPE_CSTR  => "VARTYPE_STR",

+    #PROPTYPE_SYSLOGTIME => "VARTYPE_SYSLOGTIME",

+  );

+  my $vtype = $p2v{$ptype};

+  unless ($vtype) {

+    die "property type $t is not supported!\n";

+  }

+  return $vm{$vtype};

+}

+sub serialize {

+  my $self = shift;

+  # runtime/obj.c: objSerializeHeader()

+  my $x = COOKIE_OBJLINE();

+  $x .= join(":", $self->type(), $self->cver(), $self->id(), $self->version());

+  $x .= ":\n";

+  for ( values %{$self->{props}} ) {

+    # runtime/obj.c: SerializeProp()

+    $x .= COOKIE_PROPLINE();

+    $x .= join(":",

+            $_->{name},

+            VARTYPE($_->{type}),

+            length($_->{value}),

+            $_->{value});

+    $x .= ":\n";

+  }

+  # runtime/obj.c: EndSerialize()

+  $x .= COOKIE_ENDLINE() . "End\n";

+  $x .= COOKIE_BLANKLINE() . "\n";

+}

+# constructor: new(id,version)

+sub new {

+  my ($class, $id, $version) = @_;

+  $class = ref $class if ref $class;

+  bless {

+    id => $id,

+    version => $version,

+    props => {},

+  }, $class;

+}

+sub id {

+  my $self = shift;

+  if (@_) {

+    my $x = $self->{id};

+    $self->{id} = shift;

+    return $x;

+  }

+  return $self->{id};

+}

+sub version {

+  my $self = shift;

+  if (@_) {

+    my $x = $self->{version};

+    $self->{version} = shift;

+    return $x;

+  }

+  return $self->{version};

+}

+# property(name, type, value)

+sub property {

+  my $self = shift;

+  my $name = shift;

+  if (@_) {

+    my $x = $self->{props}{$name};

+    $self->{props}{$name}{name} = $name;

+    $self->{props}{$name}{type} = shift;

+    $self->{props}{$name}{value} = shift;

+    return $x;

+  }

+  return $self->{props}{$name};

+}

+1;

+package Rsyslog::OPB;

+use base qw(Rsyslog::Serializable);

+sub type { 'OPB' }

+sub cver { 1 }

+sub new { shift->SUPER::new(@_) }

+1;

+package Rsyslog::Obj;

+use base qw(Rsyslog::Serializable);

+sub type { 'Obj' }

+sub cver { 1 }

+sub new { shift->SUPER::new(@_) }

+1;