Bash - Log timestamp of the moment data was received on stdin, coming from a pipe

Question

I have an application that outputs to stdout from time to time - random intervals, random durations of outputting before going silent again. Its output (binary data) is captured into a file.

I want to be able to log a timestamp of the moment when the output started, but only if the application kept outputting for at least 3 seconds, and also log the time when the output has stopped if no output was received for at least 5 seconds after the stop (if it is received, consider it part of the previous period).

Start: (date +'%d.%m - %H:%M:%S')
End: (date +'%d.%m - %H:%M:%S')

The output must be transparently redirected to the file after that.

Something like tee, but with timestamp logger, say:

app | monitor > out

monitor could output either to a log file or stderr, doesn't matter.

Is there a way to do this with common tools available on the shell?

I could write a Python script, but i don't wanna overcomplicate the task for no reason.

Answer 1

I've written something a while back when I wanted to log the data going through a pipe for a histogram. It's a pretty quick an dirty C program but It-work-for-me (tm). You can just put it somewhere in your pipe and the data going through it just passed along unaltered. JSON is outputted to stderror, or to a file with the -o switch.

Here's an example:

{ echo -n A; sleep 3; echo -n B; sleep 6; echo -n C; } | ./pipe_histo2 -z 2>&1 >/dev/null | jq .


{
  "measurements": [
     [ 0, 1 ],
     [ 2, 1 ],
     [ 8, 1 ]
  ],
  "min": 0,
  "max": 1,
  "avg": 0.333336,
  "total": 3,
  "start_time": "2024-06-28 23:31:34.338997847",
  "start_epoch": "1719610294",
  "end_time": "2024-06-28 23:31:43.338917382",
  "end_epoch": "1719610303",
  "elapsed": 8,
  "total_readable": "3B",
  "elapsed_readable": "08s"
}

Here's the C code. Copy this into a file called pipe_histo2.c

/*

To compile: gcc -Wall -o pipe_histo2 pipe_histo2.c

Examples:

{ echo -n A; sleep 3; echo -n B; sleep 3; echo -n C; } | ./pipe_histo 2>&1 >/dev/null | jq .
ls -als /dev | ./pipe_histo2

*/



/***************************************************************************/
/** Includes                                                              **/

#include <stdio.h>
#include <stdlib.h>         // atexit

#include <sys/time.h>       // select
#include <sys/types.h>      // select
#include <unistd.h>         // select

#include <time.h>           // clock_gettime

#include <string.h>         // strlen

char *output_file = NULL;
FILE *output_fd;

unsigned int omit_zero_measurements = 0;
unsigned int omit_seconds = 0;
unsigned int only_output_json = 0;
unsigned int print_graph = 0;


void die(char *S)
{
printf("%s\n", S);
exit(1);
}


void Usage(char *Argv0)
{
printf("\n"
       "Usage:     <some process> | %s [options] [| <some process>]\n"
       "\n"
       " -o <output file>    Output JSON to a file\n"
       " -z                  Omit zero measurements. If value == 0, do not print\n"
       " -s                  Omit seconds. Only print values in the JSON\n"
       " -j                  Only print JSON. Does not output what has been read, only prints JSON\n"
       " -g                  Output an ASCII art graph\n"
       " -h                  This help text\n"
       "\n"
       "Example:   { echo -n A; sleep 3; echo -n B; sleep 6; echo -n C; } | %s -z 2>&1 >/dev/null | jq .\n"
       "\n"
       "           stdbuf -i0 -o0 -e0 find / -xdev | %s -o result.json | wc -l\n"
       "\n"
       ,Argv0
       ,Argv0
       ,Argv0
       );
exit(0);
}


void parse_arg(int argc, char *argv[])
{
int c;

while((c = getopt(argc, argv, "ho:zsjg")) != -1)
  switch(c)
      {
      case 'o':    output_file = optarg;
                   break;
      case 'z':    omit_zero_measurements = 1;
                   break;
      case 's':    omit_seconds = 1;
                   break;
      case 'j':    only_output_json = 1;
                   break;
      case 'g':    print_graph = 1;
                   break;
      case 'h':    Usage(argv[0]);
                   break;
      case '?':    die("getopt: unknown option");
      default:     die("getopt: default reached. Internal error");
      }

output_fd = stderr;

if(only_output_json)
  output_fd = stdout;

if(output_file)
  if((output_fd = fopen(output_file, "w")) == NULL)
    die("fopen");
}

static void timespec_diff(struct timespec *end, struct timespec *start, struct timespec *result)
{
if(end->tv_nsec < start->tv_nsec)
  {
  result->tv_sec  = end->tv_sec - start->tv_sec - 1;
  result->tv_nsec = 1000000000 + end->tv_nsec - start->tv_nsec;
  }
  else
  {
  result->tv_sec  = end->tv_sec  - start->tv_sec;
  result->tv_nsec = end->tv_nsec - start->tv_nsec;
  }
}



// https://stackoverflow.com/questions/8304259/formatting-struct-timespec/14746954

static char *timespec2str(struct timespec *TS)
{
static char   ret[] = "2012-12-31 12:59:59.123456789";   // Maximum size
struct tm     t;

tzset();
if(localtime_r(&(TS->tv_sec), &t) == NULL)
  die("localtime_r");

if(strftime(ret, strlen(ret), "%F %T", &t) == 0)
  die("strftime");

sprintf(&ret[strlen(ret)], ".%09ld", TS->tv_nsec);

return ret;
}


double timespec2double(struct timespec *TS)
{
double ret = TS->tv_nsec;

while(ret > 1)
  ret = ret / 10;

return ret + TS->tv_sec;
}



char *printReadableSize(size_t Size)
{
unsigned long long int scale = 1;
char *postfix = "BKMGTPEZY";
static char result[16];

for(; Size / (scale * 1024); scale = scale * 1024, postfix++);

if(Size % scale)
  sprintf(result, "%.02f%.1s", (float) Size / scale, postfix);
  else
  sprintf(result, "%u%.1s", (unsigned int) (Size / scale), postfix);

return result;
}



char *printReadableTime(unsigned long int Seconds)
{
static char       ret[22];
unsigned int      ret_len = 0;

if(Seconds > 31536000) // years
  {
  ret_len = ret_len + sprintf(&ret[ret_len], "%luy", Seconds / 31536000);
  Seconds = Seconds % 31536000;
  }

if(Seconds > 604800) // weeks
  {
  ret_len = ret_len + sprintf(&ret[ret_len], "%luw", Seconds / 604800);
  Seconds = Seconds % 604800;
  }

if(Seconds > 86400) // days
  {
  ret_len = ret_len + sprintf(&ret[ret_len], "%lud", Seconds / 86400);
  Seconds = Seconds % 86400;
  }

if(Seconds > 3600) // hours
  {
  ret_len = ret_len + sprintf(&ret[ret_len], "%luh", Seconds / 3600);
  Seconds = Seconds % 3600;
  }

if(Seconds > 60) // minutes
  {
  ret_len = ret_len + sprintf(&ret[ret_len], "%lum", Seconds / 60);
  Seconds = Seconds % 60;
  }

ret_len = ret_len + sprintf(&ret[ret_len], "%02lus", Seconds);

ret[ret_len] = '\0';
return ret;
}




int main(int argc, char *argv[])
{
char             buf[4096];
unsigned int     buf_len=0;

fd_set               read_set;
fd_set               error_set;
unsigned long int    t;

struct timespec   start;
struct timespec   end;
struct timespec   tmp;


size_t histo[604800] = { 0 };   // A weeks worth of seconds
unsigned long int histo_len = 0;

size_t value_min;
size_t value_max;
size_t value_total;

unsigned int first = 1;


parse_arg(argc, argv);



FD_ZERO(&read_set);
FD_SET(STDIN_FILENO, &read_set);
FD_ZERO(&error_set);
FD_SET(STDIN_FILENO, &error_set);


if(clock_gettime(CLOCK_REALTIME, &start) == -1)
  die("CLOCK_REALTIME");

while(1)
  {
  t = select(STDIN_FILENO+1, &read_set, NULL, &error_set, NULL);

  if(FD_ISSET(STDIN_FILENO, &error_set))
    die("error_set\n");

  switch(t)
    {
    case -1:  die("Getch_timeout: select returned -1\n");

    case  0:  die("0 - on blocking call\n");

    default:  buf_len = read(STDIN_FILENO, buf, 4096);
              switch(buf_len)
                {
                case -1:   die("read");
                case  0:   goto exit_copy;          // EOF
                }

              if(clock_gettime(CLOCK_REALTIME, &end) == -1)
                die("CLOCK_REALTIME");
              timespec_diff(&end, &start, &tmp);
              histo[tmp.tv_sec] += buf_len;
              histo_len = tmp.tv_sec;

              if(!only_output_json)
                {
                t = write(STDOUT_FILENO, buf, buf_len);
                if(t != buf_len)
                  die("write != read\n");
                }
              break;

    } // switch
  } // while

exit_copy:


if(clock_gettime(CLOCK_REALTIME, &end) == -1)
  die("CLOCK_REALTIME");

// Display results

fprintf(output_fd, "{\"measurements\":[");

value_min = histo[0];
value_max = 0;
value_total = 0;

for(t=0; t <= histo_len; t++)
  {
  if(!omit_zero_measurements || histo[t])
    {
    if(first)
      first = 0;
      else
      fprintf(output_fd, ",");
    if(omit_seconds)
      fprintf(output_fd, "%lu", histo[t]);
      else
      fprintf(output_fd, "[%lu,%lu]", t, histo[t]);
    }
  if(histo[t] < value_min)
    value_min = histo[t];
  if(histo[t] > value_max)
    value_max = histo[t];
  value_total += histo[t];
  }


timespec_diff(&end, &start, &tmp);

fprintf(output_fd, "],\"min\":%lu,\"max\":%lu,\"avg\":%f,\"total\":%lu", value_min, value_max, (double) value_total / timespec2double(&tmp), value_total);
fprintf(output_fd, ",\"start_time\":\"%s\",\"start_epoch\":\"%lu\"", timespec2str(&start), start.tv_sec);
fprintf(output_fd, ",\"end_time\":\"%s\",\"end_epoch\":\"%lu\"", timespec2str(&end), end.tv_sec);
fprintf(output_fd, ",\"elapsed\":%lu", tmp.tv_sec);
fprintf(output_fd, ",\"total_readable\":\"%s\",\"elapsed_readable\":\"%s\"", printReadableSize(value_total), printReadableTime(tmp.tv_sec));

if(print_graph)
  {
  fprintf(output_fd, ",\"graph\":\"xx TODO xx\"");
  }

fprintf(output_fd, "}\n");

fclose(output_fd);
return 0;
}

// EOF