um_self_mon - example of UM self-monitoring.

Table of contents

• um_self_mon - example of UM self-monitoring.
• Table of contents
    • Copyright and License
    • Repository
    • Introduction
        • Logging and Alerting Infrastructure
    • Code Notes
        • The "E()" macro
        • printf()
        • sample_stats()
        • check_stats()
        • Thresholds
        • Threading
        • Data Dump

Copyright and License

All of the documentation and software included in this and any other Informatica Ultra Messaging GitHub repository Copyright (C) Informatica. All rights reserved.

Permission is granted to licensees to use or alter this software for any purpose, including commercial applications, according to the terms laid out in the Software License Agreement.

This source code example is provided by Informatica for educational and evaluation purposes only.

THE SOFTWARE IS PROVIDED "AS IS" AND INFORMATICA DISCLAIMS ALL WARRANTIES EXPRESS OR IMPLIED, INCLUDING WITHOUT LIMITATION, ANY IMPLIED WARRANTIES OF NON-INFRINGEMENT, MERCHANTABILITY OR FITNESS FOR A PARTICULAR PURPOSE. INFORMATICA DOES NOT WARRANT THAT USE OF THE SOFTWARE WILL BE UNINTERRUPTED OR ERROR-FREE. INFORMATICA SHALL NOT, UNDER ANY CIRCUMSTANCES, BE LIABLE TO LICENSEE FOR LOST PROFITS, CONSEQUENTIAL, INCIDENTAL, SPECIAL OR INDIRECT DAMAGES ARISING OUT OF OR RELATED TO THIS AGREEMENT OR THE TRANSACTIONS CONTEMPLATED HEREUNDER, EVEN IF INFORMATICA HAS BEEN APPRISED OF THE LIKELIHOOD OF SUCH DAMAGES.

Repository

See https://github.com/UltraMessaging/um_self_mon for code and documentation.

Introduction

Informatica generally recommends deploying a centralized monitoring infrastructure to monitor UM operation. See https://ultramessaging.github.io/currdoc/doc/Operations/monitoring.html

However, we understand that adding a centralized monitoring infrastructure is sometimes not practical. This repository demonstrates "monitoring-lite" in which an application can monitor its own statistics and raise an alert if there is an issue.

Logging and Alerting Infrastructure

We assume that your system already has logging and alerting infrastructure. This example is intended to leverage your existing infrastructure to add UM problem reporting. As you look at the "um_self_mon.c" code, look for all instances of "printf" and design your solution around replacing the printf with calls to your own logging and alerting infrastructure.

WARNING Some of the instances of printf are in performance-sensitive code flows. It is important that your logging/alerting be very low overhead with no possibility of blocking (putting the calling thread to sleep). This might require a separate thread to do file or network I/O.

Also note that there can be multiple threads calling the logging/alerting functionality concurrently, so the interface must be thread-safe.

Code Notes

The "E()" macro

"E()" is an optional macro that I use to perform simple error checking.

Every call to UM should be followed by a check for successful status. This often leads to sequences of code like this:

  err = lbm_config(cfg_file_name);
  if (err == -1) {
    /* Issue an alert and (usually) exit with bad status because most of the time if UM
     * returns an error, it is pointless for the code to continue. */
  }

The "if" statements obscure the logic of the code and makes it hard to read.

The "E()" macro is a short-cut that lets you write code like this:

  E(lbm_config(cfg_file_name));

If it fails, the macro prints a line similar to this:

  ALERT: [um_self_mon.c:239]: lbm_config(cfg_file_name) failed: 'CoreApi-5688-62: could not open config file'

printf()

Every place you see a call to printf(), replace it with calls to your Logging and Alerting Infrastructure.

sample_stats()

When retrieving UM statistics, the data is returned in an array of stats structures with each element of the array corresponding to an active UM transport session. Note that even if you have, say, only one receiver object (one topic that you're listening to), you can still have multiple transport sessions delivering data to it. Two publishers might have sources for that topic; each publisher's data stream is a separate transport session, with its own independent statistics.

This means that the number of active transport sessions can change over time. A new publisher might start, which the receiver joins, increasing the transport session count. Unfortunately, the application usually cannot predict ahead of time how many transport sessions will be active, and therefore cannot pre-allocate the correct number of elements in the stats structure array.

The practical solution is to estimate the number of transport sessions needed by all sources and receivers and code the statistics retrieval function sample_stats() to dynamically increase the size if necessary.

Here's the receiver initialization code with the initial estimates:

  stats_state->rcv_entries_used = 0;
  stats_state->rcv_entries_max = 10;  /* Estimate number of receiver transport sessions. */
  stats_state->rcv_stats_array = (lbm_rcv_transport_stats_t *)malloc(stats_state->rcv_entries_max * sizeof(lbm_rcv_transport_stats_t));

The source initialization code is similarly coded.

Then the retrieval code contains a loop that handles the dynamic increasing:

  /* Need to loop in case there are more transports than we have allocated buffer space for. */
  while (1) {
    expected_entries = stats_state->rcv_entries_max;
    err = lbm_context_retrieve_rcv_transport_stats_ex(stats_state->ctx, &expected_entries, sizeof(*stats_state->rcv_stats_array), stats_state->rcv_stats_array);
    if (err == 0) {  /* Success. */
      break;  /* Out of while loop. */
    }
    else if (err == -1 && lbm_errnum() == LBM_EINVAL) {  /* Buffer too small. */
      stats_state->rcv_entries_max = 2 * expected_entries;  /* Double what UM told us it needs (in case it is growing). */
      stats_state->rcv_stats_array = realloc(stats_state->rcv_stats_array, stats_state->rcv_entries_max * sizeof(*stats_state->rcv_stats_array));
      if (stats_state->rcv_stats_array == NULL) { printf("ALERT: [%s:%d]: realloc returned NULL\n", __FILE__, __LINE__); exit(1); }
    }
    else { E(err); }  /* Fatal error. */
  }  /* while */
  stats_state->rcv_entries_used = expected_entries;

The lbm_context_retrieve_rcv_transport_stats_ex() function returns an error if the array is too small, but it also updates the expected_entries variable for the number that it needed. Note that in a real-time system, that number can change rapidly, potentially even in the time it takes the loop to cycle. So the code allocates double the expected number. In practice, the loop will only occasionally loop at all, and then will usually only loop once.

check_stats()

The check_stats() function steps through the statistics arrays accumulating the most critical fields to track. For subscribers they are:

• lost - packets detected as missing from the data stream. UM will automatically attempt to recover the missing packets. Note that successful recover does not reduce the lost count.
• unrecovered_tmo and unrecovered_txw; - two counts of unrecoverable loss at the transport level. They are different by the method used to declare the packet unrecoverable: tmo = timeout, txw = transmission window advance. For the purposes of this alerting, they do not need to be differentiated, and in this program, they are added together.

UM statistics are cumulative. They generally never reset to zero. But simply having a cumulative count over a long runtime doesn't easily indicate a problem. If the lost count is 1 million, that might be a big problem, unless it is evenly spread across a million minutes of operation. So the program maintains a previous value from the previous sample. This allows the calculation of the rate of change of the summed critical statistics.

Note that since check_stats() sums values across all active transport sessions, the sum can decrease. For example, if a lossy transport session closes between two samples, then the sample after the closure will no longer include that transport session's loss count. This will distort the rate of change since it represents a different sum, possibly generating a false negative. For example, the loss rate on a still-active transport might be enough to justify an alert, but the closing of an unrelated transport could mask it. But this should be a reasonably rare exception.

Here's the code that handles the case where the summed stat decreases (it skips the threshold comparison):

  if (cur_rcv_lost > stats_state->prev_rcv_lost) {  /* Summed stats can decrease if transport sessions close. */

Thresholds

The program uses some specific alerting thresholds:

/* Thresholds for different statistics change rate to trigger an operator alert. Measured in units per second. */
#define RCV_LOST_THRESHOLD 100  /* Receiver with more than 100 lost packets per second is alerted. */
#define RCV_UNREC_THRESHOLD 1   /* Receiver with more than 1 unrecoverable loss per second is alerted. */
#define SRC_NAKS_THRESHOLD 100  /* Source with more than 100 NAKs received per second is alerted. */

These were chosen to illustrate the operation of the threshold logic. They are not our recommendations for thresholds.

Some users live with more lost packets per second than this. A few of our users would consider even 5 lost packets per second to be unacceptably high. We recommend working to reduce your loss rates as low as possible. Remember that every loss event introduces a significant latency spike of many milliseconds.

You will have to examine your system behavior to decide what is a reasonable amount of loss to trigger an alert.

Threading

The example program creates a reactor-only context:

  /* Use a separate thread to run the stats timer. (A reactor-only context is just a thread.) */
  E(lbm_context_reactor_only_create(&stats_ctx, NULL));
  stats_state->running = 1;
  /* For this test, run the monitoring tasks every second. In production, should be once per minute or longer. */
  E(stats_state->timer_id = lbm_schedule_timer(stats_ctx, stats_timer_cb, stats_state, NULL, 1000));

This provides an independent thread to run the monitoring timer, which prevents it from interfering with the time-critical main context.

If creating an additional thread is not desired, that context can be eliminated and the monitoring timer can be set on an already-existing context in the application. However, be aware that given the looping behavior, there is no absolute bound on its overhead. We recommend against using a time-critical context for the timer.

For this test program, a wake up time of 1000 ms (1 sec) was chosen to illustrate the threshold logic.

  E(stats_state->timer_id = lbm_schedule_timer(stats_ctx, stats_timer_cb, stats_state, NULL, 1000));

In a production system, a longer period is usually more appropriate to minimize overhead. But do not choose a value that is too large. For example, if a one-hour period is chosen, you could have very intense spikes of loss that clearly justify an alert, but it gets averaged over the entire hour and is therefore not detected.

We don't have a clear recommendation, but 10 to 60 seconds seem reasonable.

Data Dump

In this minimalist approach, the only data being logged are the critical statistics that exceed their thresholds. However, Informatica UM Support organization prefers to see all of the fields of the receiver and source structures. When an alert is generated, we recommend including at least the full dump of the associated structures for all active transport sessions.

Even better is to maintain a "previous sample" collection that can also be dumped. This allows support to see a "before" and "after" picture of the problem.

Note that in systems with many active transports, this can result in thousands of individual statistics for one sample. It is not necessary to include this with the alert. For example, it might be written to an alternate data dump file. So long as its contents can be time-correlated with the alert, it is usually acceptable for the full data dump to require extra steps to retrieve.

The data dumping code was not included in this program for brevity's sake.