Merge branch 'develop'

.github/workflows/deb-build.yaml

@@ -21,7 +21,6 @@ jobs:
            - "ubuntu:xenial"
            - "ubuntu:bionic"
            - "ubuntu:focal"
-           - "ubuntu:groovy"
 
      steps:
       - name: Checkout repo

README.md

@@ -1,6 +1,6 @@
 OpenLI -- open source ETSI-compliant Lawful Intercept software
 
-Version: 1.0.10
+Version: 1.0.11
 
 ---------------------------------------------------------------------------
 

configure.ac

@@ -1,6 +1,6 @@
 # Super primitive configure script
 
-AC_INIT(openli, 1.0.10, [email protected])
+AC_INIT(openli, 1.0.11, [email protected])
 
 AM_INIT_AUTOMAKE([subdir-objects])
 AC_CONFIG_SRCDIR(src/collector/collector.c)

debian/changelog

@@ -1,3 +1,26 @@
+openli (1.0.11-1) unstable; urgency=medium
+
+  * Improved collector encoding performance by saving and reusing
+    previously-encoded records that have the exact same layout.
+  * Improved encoding performance at high packet rates by sending encoded
+    records to the forwarding thread in batches.
+  * Mediators will now actively avoid splitting a record across multiple send
+    calls wherever possible.
+  * Fixed bug where a collector would simply stop forwarding records for an
+    LIID on to the mediator for no apparent reason, especially when more
+    encoder threads were being used.
+  * Fixed bug where encoding jobs would be lost without being seen by the
+    encoder thread.
+  * Fixed performance-related issue where an overwhelmed mediator would never
+    send data to its handovers.
+  * Fixed crash in mediator after a handover is disconnected for failing to
+    send a keep alive response.
+  * BER encoding optimization has been removed.
+  * Fix issue where collector memory usage would be extremely high when under
+    load.
+
+ -- Shane Alcock <[email protected]>  Wed, 21 Jul 2021 10:56:13 +1200
+
 openli (1.0.10-1) unstable; urgency=medium
 
   * Intercepts can now be configured with a start and/or end time (unix

debian/control

@@ -2,8 +2,8 @@ Source: openli
 Section: net
 Priority: optional
 Maintainer: Shane Alcock <[email protected]>
-Build-Depends: debhelper (>= 9), dh-autoreconf, dh-systemd (>=1.5),
- libtrace4-dev (>= 4.0.14), libyaml-dev, uthash-dev, libwandder2-dev,
+Build-Depends: debhelper-compat (= 12), dh-autoreconf, dh-systemd (>=1.5),
+ libtrace4-dev (>= 4.0.16), libyaml-dev, uthash-dev, libwandder2-dev,
  libjudy-dev, libzmq3-dev, libgoogle-perftools-dev, libosip2-dev,
  libssl1.0-dev (>=1.0.2r) | libssl-dev, librabbitmq-dev,
  libmicrohttpd-dev, libjson-c-dev, libsqlcipher-dev

debian/rules

@@ -1,7 +1,7 @@
 #!/usr/bin/make -f
 %:
 
-	dh $@ --with=systemd --with autoreconf
+	dh $@ --with autoreconf
 
 
 override_dh_auto_configure:

debpkg-setup.sh

@@ -11,7 +11,7 @@ export SOURCENAME=`echo ${GITHUB_REF##*/} | cut -d '-' -f 1`
 apt-get update
 apt-get install -y equivs devscripts dpkg-dev quilt curl apt-transport-https \
     apt-utils ssl-cert ca-certificates gnupg lsb-release debhelper git \
-    pkg-config
+    pkg-config sed
 
 DISTRO=$(lsb_release -sc)
 
@@ -21,8 +21,23 @@ curl -1sLf 'https://dl.cloudsmith.io/public/wand/libtrace/cfg/setup/bash.deb.sh'
 curl -1sLf 'https://dl.cloudsmith.io/public/wand/openli/cfg/setup/bash.deb.sh' | bash
 
 case ${DISTRO} in
-        jessie | xenial | stretch )
+        xenial )
                 curl -1sLf 'https://dl.cloudsmith.io/public/wand/dpdk-wand/cfg/setup/bash.deb.sh' | bash
+                apt-get install -y debhelper dh-systemd -t xenial-backports
+                sed -i 's/debhelper-compat (= 12)/debhelper (>= 10)/' debian/control
+                sed -i 's/--with auto/--with=systemd --with auto/' debian/rules
+                echo "10" > debian/compat
+        ;;
+
+        stretch )
+                curl -1sLf 'https://dl.cloudsmith.io/public/wand/dpdk-wand/cfg/setup/bash.deb.sh' | bash
+                sed -i 's/debhelper-compat (= 12)/debhelper (>= 10)/' debian/control
+                sed -i 's/--with auto/--with=systemd --with auto/' debian/rules
+                echo "10" > debian/compat
+        ;;
+
+        bionic )
+                apt-get install -y debhelper -t bionic-backports
         ;;
 esac
 

doc/CollectorDoc.md

@@ -155,28 +155,6 @@ A collector only requires a small amount of configuration: a username and
 password that can be used to authenticate against a local RabbitMQ instance,
 and a flag to inform the collector that RabbitMQ output is enabled.
 
-### Encoding Methods
-OpenLI supports two methods for encoding the intercepted records and meta-data
-so that they conform to the ETSI standard. The first method is DER encoding,
-whereby all fields are encoded in the most space-efficient way. The advantage
-of DER is that there is only one "correct" result for encoding a given input,
-so it is less likely to be incompatible with what the receiver is expecting.
-However, this encoding method is slower and therefore may limit your
-maximum interception capability.
-
-The second method is BER encoding, which allows us to encode numeric values
-using a fixed-width field, e.g. every integer consumes eight bytes, regardless
-of whether it is a small number or a large one. This is less space-efficient
-and relies on the receiver being prepared for fields that consume more space
-than they might need, but is much faster for OpenLI to encode because each
-record will have a relatively static structure.
-
-OpenLI versions prior to 1.0.9 supported DER encoding only, and this is the
-default encoding method as it has been well tested in production deployments.
-BER has the potential to be much more efficient, but we recommend doing some
-specific testing with your local LEAs before switching over to BER as not all
-LEA equipment may be able to decode it.
-
 ### Target Identification for VOIP Intercepts
 By default, OpenLI does NOT trust the "From:" field in SIP packets when it is
 determining whether a SIP packet has been sent by an intercept target. This
@@ -209,8 +187,6 @@ The basic option keys are:
 * interceptpointid  -- set the interception point ID
 * seqtrackerthreads -- set the number of threads to use for sequence number
                        tracking (defaults to 1).
-* encoding          -- choose the encoding method to use, either `ber` or
-                       `der` (defaults to `der`).
 * encoderthreads    -- set the number of threads to use for encoding ETSI
                        records (defaults to 2).
 * forwardingthreads -- set the number of threads to use for forwarding

doc/exampleconfigs/collector-example.yaml

@@ -28,11 +28,6 @@ encoderthreads: 2
 # mediators. You probably don't need to change this.
 forwardingthreads: 1
 
-# The encoding method to use for generating ETSI records. DER is the default
-# method. BER is an alternative that should be more efficient, but not all
-# LEAs may be able to decode it.
-encoding: der
-
 # Set this to yes if you want to override the policy of not trusting the
 # contents of the "From:" field in SIP packets (as this field is not
 # validated and can be easily spoofed).

rpm/openli.spec

@@ -1,5 +1,5 @@
 Name:           openli
-Version:        1.0.10
+Version:        1.0.11
 Release:        1%{?dist}
 Summary:        Software for performing ETSI-compliant lawful intercept
 
@@ -14,7 +14,7 @@ BuildRequires: bison
 BuildRequires: doxygen
 BuildRequires: flex
 BuildRequires: libyaml-devel
-BuildRequires: libtrace4-devel >= 4.0.14
+BuildRequires: libtrace4-devel >= 4.0.16
 BuildRequires: Judy-devel
 BuildRequires: uthash-devel
 BuildRequires: libwandder2-devel
@@ -232,6 +232,9 @@ fi
 
 
 %changelog
+* Mon Jul 5 2021 Shane Alcock <[email protected]> - 1.0.11-1
+- Updated for 1.0.11 release
+
 * Tue Apr 26 2021 Shane Alcock <[email protected]> - 1.0.10-1
 - Updated for 1.0.10 release
 

src/Makefile.am

@@ -40,7 +40,7 @@ openlicollector_SOURCES=collector/collector.c configparser.c configparser.h \
                 netcomms.c netcomms.h byteswap.c byteswap.h etsili_core.c \
                 collector/sipparsing.c collector/sipparsing.h \
                 collector/jenkinshash.c collector/ipmmcc.c collector/ipmmcc.h \
-		collector/ipmmiri.c collector/ipmmiri.h \
+		collector/ipmmiri.h \
                 collector/internetaccess.c collector/internetaccess.h \
 		collector/ipcc.c collector/ipcc.h \
                 coreserver.h coreserver.c collector/collector_push_messaging.c \
@@ -55,7 +55,6 @@ openlicollector_SOURCES=collector/collector.c configparser.c configparser.h \
                 collector/collector_seqtracker.c \
                 collector/collector_forwarder.c collector/jmirror_parser.c \
                 collector/jmirror_parser.h openli_tls.c openli_tls.h \
-                collector/umtscc.h collector/umtscc.c \
                 collector/umtsiri.h collector/umtsiri.c \
                 collector/radius_hasher.c collector/radius_hasher.h \
                 collector/timed_intercept.c collector/timed_intercept.h \

src/collector/collector.c

@@ -24,7 +24,7 @@
  *
  */
 
-
+#define _GNU_SOURCE
 #include <stdio.h>
 #include <stdlib.h>
 #include <assert.h>
@@ -196,7 +196,7 @@ static void process_tick(libtrace_t *trace, libtrace_thread_t *t,
 static void init_collocal(colthread_local_t *loc, collector_global_t *glob,
         int threadid) {
 
-    int zero = 0, i;
+    int zero = 0, i, hwm=1000;
     libtrace_message_queue_init(&(loc->fromsyncq_ip),
             sizeof(openli_pushed_t));
     libtrace_message_queue_init(&(loc->fromsyncq_voip),
@@ -227,18 +227,18 @@ static void init_collocal(colthread_local_t *loc, collector_global_t *glob,
 
         snprintf(pubsockname, 128, "inproc://openlipub-%d", i);
         loc->zmq_pubsocks[i] = zmq_socket(glob->zmq_ctxt, ZMQ_PUSH);
-        zmq_setsockopt(loc->zmq_pubsocks[i], ZMQ_SNDHWM, &zero, sizeof(zero));
+        zmq_setsockopt(loc->zmq_pubsocks[i], ZMQ_SNDHWM, &hwm, sizeof(hwm));
         zmq_connect(loc->zmq_pubsocks[i], pubsockname);
     }
 
     loc->fragreass = create_new_ipfrag_reassembler();
 
     loc->tosyncq_ip = zmq_socket(glob->zmq_ctxt, ZMQ_PUSH);
-    zmq_setsockopt(loc->tosyncq_ip, ZMQ_SNDHWM, &zero, sizeof(zero));
+    zmq_setsockopt(loc->tosyncq_ip, ZMQ_SNDHWM, &hwm, sizeof(hwm));
     zmq_connect(loc->tosyncq_ip, "inproc://openli-ipsync");
 
     loc->tosyncq_voip = zmq_socket(glob->zmq_ctxt, ZMQ_PUSH);
-    zmq_setsockopt(loc->tosyncq_voip, ZMQ_SNDHWM, &zero, sizeof(zero));
+    zmq_setsockopt(loc->tosyncq_voip, ZMQ_SNDHWM, &hwm, sizeof(hwm));
     zmq_connect(loc->tosyncq_voip, "inproc://openli-voipsync");
 
 }
@@ -1588,6 +1588,7 @@ int main(int argc, char *argv[]) {
     collector_global_t *glob = NULL;
     int i, ret, todaemon;
     colinput_t *inp, *tmp;
+    char name[1024];
 
     todaemon = 0;
     while (1) {
@@ -1678,6 +1679,9 @@ int main(int argc, char *argv[]) {
             sizeof(forwarding_thread_data_t));
 
     for (i = 0; i < glob->forwarding_threads; i++) {
+
+        snprintf(name, 1024, "forwarder-%d", i);
+
         glob->forwarders[i].zmq_ctxt = glob->zmq_ctxt;
         glob->forwarders[i].forwardid = i;
         glob->forwarders[i].encoders = glob->encoding_threads;
@@ -1692,29 +1696,30 @@ int main(int argc, char *argv[]) {
 
         pthread_create(&(glob->forwarders[i].threadid), NULL,
                 start_forwarding_thread, (void *)&(glob->forwarders[i]));
+        pthread_setname_np(glob->forwarders[i].threadid, name);
     }
 
     glob->seqtrackers = calloc(glob->seqtracker_threads,
             sizeof(seqtracker_thread_data_t));
 
     for (i = 0; i < glob->seqtracker_threads; i++) {
+        snprintf(name, 1024, "seqtracker-%d", i);
         glob->seqtrackers[i].zmq_ctxt = glob->zmq_ctxt;
         glob->seqtrackers[i].trackerid = i;
         glob->seqtrackers[i].zmq_pushjobsock = NULL;
         glob->seqtrackers[i].zmq_recvpublished = NULL;
         glob->seqtrackers[i].intercepts = NULL;
         glob->seqtrackers[i].colident = &(glob->sharedinfo);
         glob->seqtrackers[i].encoding_method = glob->encoding_method;
-#ifdef HAVE_BER_ENCODING
-        glob->seqtrackers[i].enc_ber = wandder_init_encoder_ber(1000, 512);
-#endif
         pthread_create(&(glob->seqtrackers[i].threadid), NULL,
                 start_seqtracker_thread, (void *)&(glob->seqtrackers[i]));
+        pthread_setname_np(glob->seqtrackers[i].threadid, name);
     }
 
     glob->encoders = calloc(glob->encoding_threads, sizeof(openli_encoder_t));
 
     for (i = 0; i < glob->encoding_threads; i++) {
+        snprintf(name, 1024, "encoder-%d", i);
         glob->encoders[i].zmq_ctxt = glob->zmq_ctxt;
         glob->encoders[i].zmq_recvjobs = NULL;
         glob->encoders[i].zmq_pushresults = NULL;
@@ -1724,12 +1729,15 @@ int main(int argc, char *argv[]) {
         glob->encoders[i].shared = &(glob->sharedinfo);
         glob->encoders[i].encoder = NULL;
         glob->encoders[i].freegenerics = NULL;
+        glob->encoders[i].saved_intercept_templates = NULL;
+        glob->encoders[i].saved_global_templates = NULL;
 
         glob->encoders[i].seqtrackers = glob->seqtracker_threads;
         glob->encoders[i].forwarders = glob->forwarding_threads;
 
         pthread_create(&(glob->encoders[i].threadid), NULL,
                 run_encoder_worker, (void *)&(glob->encoders[i]));
+        pthread_setname_np(glob->encoders[i].threadid, name);
     }
 
     /* Start IP intercept sync thread */
@@ -1739,6 +1747,8 @@ int main(int argc, char *argv[]) {
         logger(LOG_INFO, "OpenLI: error creating IP sync thread. Exiting.");
         return 1;
     }
+    snprintf(name, 1024, "sync-ip");
+    pthread_setname_np(glob->syncip.threadid, name);
 
     /* Start VOIP intercept sync thread */
     ret = pthread_create(&(glob->syncvoip.threadid), NULL,
@@ -1747,6 +1757,8 @@ int main(int argc, char *argv[]) {
         logger(LOG_INFO, "OpenLI: error creating VOIP sync thread. Exiting.");
         return 1;
     }
+    snprintf(name, 1024, "sync-voip");
+    pthread_setname_np(glob->syncvoip.threadid, name);
 
     if (pthread_sigmask(SIG_SETMASK, &sig_before, NULL)) {
         logger(LOG_INFO, "Unable to re-enable signals after starting threads.");

src/collector/collector_base.h

@@ -41,6 +41,8 @@
 #include "export_buffer.h"
 #include "openli_tls.h"
 
+#define MAX_ENCODED_RESULT_BATCH 50
+
 typedef struct export_dest {
     int failmsg;
     int fd;
@@ -167,23 +169,15 @@ typedef struct seqtracker_thread_data {
     exporter_intercept_state_t *intercepts;
     removed_intercept_t *removedints;
     uint8_t encoding_method;
-#if HAVE_BER_ENCODING
-    wandder_encoder_ber_t *enc_ber;
-#endif
 
 } seqtracker_thread_data_t;
 
-typedef struct stored_result {
-    openli_encoded_result_t res;
-    UT_hash_handle hh;
-} stored_result_t;
-
 typedef struct intercept_reorderer {
 
     char *liid;
     char *key;
     uint32_t expectedseqno;
-    stored_result_t *pending;
+    Pvoid_t pending;
 
 } int_reorderer_t;
 
@@ -235,18 +229,18 @@ typedef struct encoder_state {
     wandder_encoder_t *encoder;
     etsili_generic_freelist_t *freegenerics;
 
+    Pvoid_t saved_intercept_templates;
+    Pvoid_t saved_global_templates;
+
     int seqtrackers;
     int forwarders;
     uint8_t halted;
 } openli_encoder_t;
 
 typedef struct encoder_job {
     wandder_encode_job_t *preencoded;
-#ifdef HAVE_BER_ENCODING
-    wandder_etsili_top_t *top;
-    wandder_etsili_child_t *child;
-#endif
     uint32_t seqno;
+    int64_t cin;
     char *cinstr;
     openli_export_recv_t *origreq;
     char *liid;