src/serial_model.c

   1 /*======================================================================*
   2  * Copyright (c) 2010, OpenX Inc. All rights reserved.                  *
   3  *                                                                      *
   4  * Licensed under the New BSD License (the "License"); you may not use  *
   5  * this file except in compliance with the License.  Unless required    *
   6  * by applicable law or agreed to in writing, software distributed      *
   7  * under the License is distributed on an "AS IS" BASIS, WITHOUT        *
   8  * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.     *
   9  * See the License for the specific language governing permissions and  *
  10  * limitations under the License. See accompanying LICENSE file.        *
  11  *======================================================================*/
  12
  13 #include "config.h"
  14
  15 #include "journal.h"
  16 #include "log.h"
  17 #include "opt.h"
  18 #include "serial_model.h"
  19 #include "sig.h"
  20 #include "stats.h"
  21 #include "xport.h"
  22
  23 #include <fcntl.h>
  24 #include <lwes.h>
  25 #include <stdio.h>
  26 #include <stdlib.h>
  27 #include <string.h>
  28
  29 #define BUFLEN               (65535)
  30 #define SERIAL_DEPTH_COMMAND ("\026Internal::Queue::Depth")
  31
  32 struct xport             xpt;
  33 unsigned char            buf[BUFLEN];
  34 int                      buflen;
  35 struct journal           jrn;
  36 struct enqueuer_stats    est;
  37 struct dequeuer_stats    dst;
  38 unsigned long long       tm;
  39 struct lwes_emitter*     emitter   = NULL;
  40 unsigned long long       depth_tm  = 0;
  41 const unsigned long long depth_dtm = 10000;
  42 unsigned long long       count     = 0;
  43 unsigned long long       pending   = 0;
  44
  45 static void serial_open_journal(void);
  46
  47 static void serial_ctor(void)
  48 {
  49   install_signal_handlers();
  50   install_rotate_signal_handlers();
  51
  52   if ( enqueuer_stats_ctor (&est) < 0 )
  53     {
  54       LOG_ER("Failed to create initialize enqueuer stats.\n");
  55       exit(EXIT_FAILURE);
  56     }
  57
  58   if ( dequeuer_stats_ctor (&dst) < 0 )
  59     {
  60       LOG_ER("Failed to create initialize dequeuer stats.\n");
  61       exit(EXIT_FAILURE);
  62     }
  63
  64   memset(buf, 0, BUFLEN);           /* Clear the message. */
  65
  66   if ( (xport_factory(&xpt) < 0) || (xpt.vtbl->open(&xpt, O_RDONLY) < 0) )
  67     {
  68       LOG_ER("Failed to create xport object.\n");
  69       exit(EXIT_FAILURE);
  70     }
  71
  72   if (arg_njournalls != 1)
  73     {
  74       LOG_ER("Expected 1 journal name pattern, but found %d\n", arg_njournalls);
  75       exit(EXIT_FAILURE);
  76     }
  77
  78   /* Create journal object. */
  79   if (journal_factory(&jrn, arg_journalls[0]) < 0)
  80     {
  81       LOG_ER("Failed to create journal object for \"%s\".\n", arg_journalls[0]);
  82       exit(EXIT_FAILURE);
  83     }
  84
  85   serial_open_journal();
  86
  87   emitter = lwes_emitter_create( (LWES_CONST_SHORT_STRING) arg_ip,
  88                                  (LWES_CONST_SHORT_STRING) NULL, //arg_interface,
  89                                  (LWES_U_INT_32) arg_port, 0, 60 );
  90 }
  91
  92 static void serial_open_journal(void)
  93 {
  94   if (jrn.vtbl->open(&jrn, O_WRONLY) < 0)
  95     {
  96       LOG_ER("Failed to open the journal \"%s\".\n", arg_journalls[0]);
  97       exit(EXIT_FAILURE);
  98     }
  99 }
 100
 101 static void serial_close_journal(void)
 102 {
 103   enqueuer_stats_rotate(&est);
 104   dequeuer_stats_rotate(&dst);
 105
 106   if (jrn.vtbl->close(&jrn) < 0) {
 107     LOG_ER("Can't close journal  \"%s\".\n", arg_journalls[0]);
 108     exit(EXIT_FAILURE);
 109   }
 110 }
 111
 112 static void serial_rotate(void)
 113 {
 114   serial_close_journal();
 115   serial_open_journal();
 116   gbl_rotate = 0;
 117 }
 118
 119 static int serial_read(void)
 120 {
 121   unsigned long addr;
 122   short         port;
 123   int           xpt_read_ret =
 124       xpt.vtbl->read(&xpt, buf+HEADER_LENGTH, BUFLEN-HEADER_LENGTH, &addr, &port);
 125
 126   if (xpt_read_ret < 0)
 127     {
 128       enqueuer_stats_record_socket_error(&est);
 129       return -1;
 130     }
 131
 132   tm     = time_in_milliseconds();
 133   buflen = xpt_read_ret + HEADER_LENGTH;
 134   enqueuer_stats_record_datagram(&est, buflen);
 135   header_add(buf, xpt_read_ret, tm, addr, port);
 136   ++count;
 137
 138   return 0;
 139 }
 140
 141 static int serial_handle_depth_test()
 142 {
 143   if (toknam_eq((unsigned char *)buf + HEADER_LENGTH, (unsigned char *)SERIAL_DEPTH_COMMAND))
 144     {
 145       struct lwes_event_deserialize_tmp event_tmp;
 146       struct lwes_event_enumeration     keys;
 147       LWES_CONST_SHORT_STRING           key;
 148       LWES_TYPE                         type;
 149       int                               bytes_read;
 150       struct lwes_event*                event = lwes_event_create_no_name(NULL);
 151
 152       bytes_read = lwes_event_from_bytes(event, (LWES_BYTE_P)&buf[HEADER_LENGTH],
 153                                          buflen-HEADER_LENGTH, 0, &event_tmp);
 154       if (bytes_read != buflen-HEADER_LENGTH)
 155         {
 156           LOG_ER("Only able to read %d bytes; expected %d\n", bytes_read, buflen);
 157         }
 158       else if (!lwes_event_keys(event, &keys))
 159         {
 160           LOG_ER("Unable to iterate over keys\n");
 161         }
 162       else
 163         {
 164           while (lwes_event_enumeration_next_element(&keys, &key, &type))
 165             {
 166               if (strncmp("count",key,6)==0)
 167                 {
 168                   LWES_U_INT_64 value;
 169                   lwes_event_get_U_INT_64(event, key, &value);
 170                   pending  = count - value;
 171                   LOG_INF("Depth test reports a buffer length of %lld events.\n", pending);
 172                 }
 173             }
 174         }
 175
 176       lwes_event_destroy(event);
 177
 178       return 1;
 179     }
 180   else
 181     {
 182       return 0;
 183     }
 184 }
 185
 186 static void serial_send_buffer_depth_test(void)
 187 {
 188   struct lwes_event *event =
 189       lwes_event_create( (struct lwes_event_type_db *) NULL,
 190                          (LWES_SHORT_STRING) SERIAL_DEPTH_COMMAND+1);
 191   if (event == NULL) return;
 192   if (lwes_event_set_U_INT_64(event,"count",count) < 0)
 193     {
 194       LOG_ER("Unable to add count to depth event");
 195     }
 196   else
 197     {
 198       lwes_emitter_emitto((char*) "127.0.0.1", NULL, arg_port, emitter, event);
 199     }
 200   lwes_event_destroy(event);
 201
 202   depth_tm = time_in_milliseconds();
 203 }
 204
 205 static void serial_write(void)
 206 {
 207   /* Write the packet out to the journal. */
 208   int jrn_write_ret = jrn.vtbl->write(&jrn, buf, buflen);
 209
 210   if (jrn_write_ret != buflen)
 211     {
 212       LOG_ER("Journal write error -- attempted to write %d bytes, "
 213              "write returned %d.\n", buflen, jrn_write_ret);
 214       dequeuer_stats_record_loss(&dst);
 215     }
 216
 217   dequeuer_stats_record(&dst, buflen, pending);
 218 }
 219
 220 static void serial_dtor(void)
 221 {
 222   serial_close_journal();
 223 }
 224
 225 void serial_model(void)
 226 {
 227   serial_ctor();
 228
 229   do {
 230     if (serial_read() < 0)          continue;
 231     if (serial_handle_depth_test()) continue;
 232     serial_write();
 233     if (header_is_rotate(buf) || gbl_rotate) serial_rotate();
 234     if (tm >= depth_tm + depth_dtm) serial_send_buffer_depth_test();
 235   }
 236   while (!gbl_done);
 237
 238   serial_dtor();
 239 }