varnishevent: catch SIGPIPE and reopen output; check output errors

bin/varnishevent/signals.h

@@ -37,3 +37,4 @@ SIGDISP(SIGUSR2, ignore_action);
 SIGDISP(SIGABRT, stacktrace_action);
 SIGDISP(SIGSEGV, stacktrace_action);
 SIGDISP(SIGBUS, stacktrace_action);
+SIGDISP(SIGPIPE, reopen_action);

bin/varnishevent/varnishevent.c

@@ -170,10 +170,10 @@ static inline logline_t
         rdr_free = DATA_Take_Freelist(&reader_freelist);
 
         if (VSTAILQ_EMPTY(&reader_freelist)) {
-            waits++;
             LOG_Log0(LOG_DEBUG, "Reader: waiting for free space");
-            waiting = 1;
             AZ(pthread_mutex_lock(&data_ready_lock));
+            waiting = 1;
+            waits++;
             if (WRT_Waiting())
                 AZ(pthread_cond_signal(&spscq_ready_cond));
             AZ(pthread_cond_wait(&data_ready_cond, &data_ready_lock));
@@ -430,9 +430,10 @@ event(void *priv, enum VSL_tag_e tag, unsigned fd,
 /*--------------------------------------------------------------------*/
 
 static void
-sighup(int sig)
+sigreopen(int sig)
 {
-    (void) sig;
+    LOG_Log(LOG_WARNING, "Received signal %d (%s), reopening output",
+        sig, strsignal(sig));
     reopen = 1;
 }
 
@@ -599,7 +600,7 @@ main(int argc, char *argv[])
     AZ(sigemptyset(&dump_action.sa_mask));
     dump_action.sa_flags |= SA_RESTART;
 
-    reopen_action.sa_handler = sighup;
+    reopen_action.sa_handler = sigreopen;
     AZ(sigemptyset(&reopen_action.sa_mask));
     reopen_action.sa_flags |= SA_RESTART;
 
@@ -715,8 +716,10 @@ main(int argc, char *argv[])
     while (VSL_Dispatch(vd, event, NULL) >= 0)
         if (term || finite)
             break;
-        else if (reopen)
+        else if (reopen) {
             WRT_Reopen();
+            reopen = 0;
+        }
         else
             LOG_Log0(LOG_WARNING, "Log read interrupted, continuing");
 

bin/varnishevent/writer.c

@@ -72,13 +72,14 @@ static unsigned	wrt_nfree;
 static struct vsb *os;
 
 static FILE *fo;
-static char *obuf;
+static char *obuf = NULL;
 static pthread_mutex_t reopen_lock;
 
 /* stats */
 static unsigned deqs = 0;
 static unsigned waits = 0;
 static unsigned writes = 0;
+static unsigned errors = 0;
 
 typedef struct writer_data_s {
     unsigned magic;
@@ -99,7 +100,9 @@ open_log(void)
     else if ((fo = fopen(config.output_file, config.append ? "a" : "w"))
               == NULL)
         return errno;
-    
+
+    if (obuf != NULL)
+        free(obuf);
     obuf = (char *) malloc(config.output_bufsiz);
     if (obuf == NULL)
         return errno;
@@ -122,6 +125,8 @@ wrt_return_freelist(void)
 static inline void
 wrt_write(logline_t *ll)
 {
+    int errnum;
+    
     CHECK_OBJ_NOTNULL(ll, LOGLINE_MAGIC);
     assert(ll->state == DATA_DONE);
 
@@ -137,9 +142,9 @@ wrt_write(logline_t *ll)
                 config.output_file, strerror(errno));
             exit(EXIT_FAILURE);
         }
-        if (open_log() != 0) {
+        if ((errnum = open_log()) != 0) {
             LOG_Log(LOG_ALERT, "Cannot reopen %s, exiting: %s",
-                config.output_file, strerror(errno));
+                config.output_file, strerror(errnum));
             exit(EXIT_FAILURE);
         }
         reopen = 0;
@@ -149,7 +154,11 @@ wrt_write(logline_t *ll)
     VSB_clear(os);
     FMT_Format(ll, os);
     VSB_finish(os);
-    fprintf(fo, "%s", VSB_data(os));
+    if (fprintf(fo, "%s", VSB_data(os)) < 0) {
+        LOG_Log(LOG_ALERT, "Output error %d (%s), DATA DISCARDED [%s]",
+            errno, strerror(errno), VSB_data(os));
+        errors++;
+    }
 
     writes++;
     MON_StatsUpdate(STATS_WRITTEN);
@@ -195,12 +204,16 @@ static void
          * flush ouput and return space before sleeping
          */
 
-        wrt->state = WRT_WAITING;
 	if (wrt_nfree > 0)
             wrt_return_freelist();
-        fflush(fo);
+        if (fflush(fo) != 0) {
+            LOG_Log(LOG_ALERT, "Output flush failed, error %d (%s)",
+                errno, strerror(errno));
+            errors++;
+        }
+        
         AZ(pthread_mutex_lock(&spscq_ready_lock));
-
+        wrt->state = WRT_WAITING;
         /*
 	 * run is guaranteed to be fresh after the lock
 	 */
@@ -220,7 +233,11 @@ static void
         CHECK_OBJ(ll, LOGLINE_MAGIC);
         wrt_write(ll);
     }
-    fflush(fo);
+    if (fflush(fo) != 0) {
+        LOG_Log(LOG_ALERT, "Output flush failed, error %d (%s)",
+            errno, strerror(errno));
+        errors++;
+    }
 
     wrt->status = EXIT_SUCCESS;
     LOG_Log0(LOG_INFO, "Writer thread exiting");
@@ -266,8 +283,9 @@ WRT_Stats(void)
 {
     if (!run) return;
     
-    LOG_Log(LOG_INFO, "Writer (%s): seen=%d writes=%d waits=%d free=%d",
-        statename[wrt_data.state], deqs, writes, waits, wrt_nfree);
+    LOG_Log(LOG_INFO,
+        "Writer (%s): seen=%u writes=%u errors=%u waits=%u free=%u",
+        statename[wrt_data.state], deqs, writes, errors, waits, wrt_nfree);
 }
 
 int