proto_load_step.c

Go to the documentation of this file.

/*
 *   This program is free software; you can redistribute it and/or modify
 *   it under the terms of the GNU General Public License as published by
 *   the Free Software Foundation; either version 2 of the License, or
 *   (at your option) any later version.
 *
 *   This program is distributed in the hope that it will be useful,
 *   but WITHOUT ANY WARRANTY; without even the implied warranty of
 *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *   GNU General Public License for more details.
 *
 *   You should have received a copy of the GNU General Public License
 *   along with this program; if not, write to the Free Software
 *   Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
 */
 
/**
 * $Id: 763d361486fa35dac02fdf0a1ac3f777bede1eee $
 * @file proto_load_step.c
 * @brief Generic protocol load generator
 *
 * @copyright 2019 The FreeRADIUS server project.
 * @copyright 2019 Network RADIUS SAS (legal@networkradius.com)
 */
#include <netdb.h>
#include <fcntl.h>
#include <freeradius-devel/server/protocol.h>
#include <freeradius-devel/server/main_loop.h>
#include <freeradius-devel/io/application.h>
#include <freeradius-devel/io/listen.h>
#include <freeradius-devel/io/schedule.h>
#include <freeradius-devel/io/load.h>
 
#include "proto_load.h"
 
extern fr_app_io_t proto_load_step;
 
typedef struct proto_load_step_s proto_load_step_t;
 
/** What to do when the load test is complete
 *
 *  The test is complete when the PPS ramp has passed max_pps, all
 *  max_requests packets (if set) have been sent, and every reply has
 *  been received.
 */
typedef enum {
        LOAD_STEP_ON_COMPLETE_EXIT = 0,                 //!< make the server exit
        LOAD_STEP_ON_COMPLETE_REPEAT,                   //!< run the test again from start_pps
        LOAD_STEP_ON_COMPLETE_STOP,                     //!< stop generating load, leave the server running
        LOAD_STEP_ON_COMPLETE_CONTINUE                  //!< never complete, keep sending at max_pps forever
} load_step_on_complete_t;
 
static fr_table_num_sorted_t const load_on_complete_table[] = {
        { L("continue"),        LOAD_STEP_ON_COMPLETE_CONTINUE },
        { L("exit"),            LOAD_STEP_ON_COMPLETE_EXIT },
        { L("repeat"),          LOAD_STEP_ON_COMPLETE_REPEAT },
        { L("stop"),            LOAD_STEP_ON_COMPLETE_STOP }
};
static size_t load_on_complete_table_len = NUM_ELEMENTS(load_on_complete_table);
 
typedef struct {
        fr_event_list_t                 *el;                    //!< event list
        fr_network_t                    *nr;                    //!< network handler
 
        char const                      *name;                  //!< socket name
        bool                            done;
        bool                            suspended;
 
        fr_time_t                       recv_time;              //!< recv time of the last packet
 
        proto_load_step_t const         *inst;
        fr_load_t                       *l;                     //!< load generation handler
        fr_load_config_t                load;                   //!< load configuration
        fr_stats_t                      stats;                  //!< statistics for this socket
 
        int                             fd;                     //!< for CSV files
        fr_timer_t                      *ev;                    //!< for writing statistics
 
        fr_listen_t                     *parent;                //!< master IO handler
} proto_load_step_thread_t;
 
struct proto_load_step_s {
        proto_load_t                    *parent;
 
        CONF_SECTION                    *cs;                    //!< our configuration
 
        char const                      *filename;              //!< where to read input packet from
        fr_pair_list_t                  pair_list;              //!< for input packet
 
        int                             code;
        uint32_t                        max_attributes;         //!< Limit maximum decodable attributes
 
        fr_client_t                     *client;                //!< static client
 
        fr_load_config_t                load;                   //!< load configuration
        load_step_on_complete_t         on_complete;            //!< what to do when the load test completes
        char const                      *csv;                   //!< where to write CSV stats
 
        fr_dict_t const                 *dict;                  //!< Our namespace.
};
 
 
static const conf_parser_t load_listen_config[] = {
        { FR_CONF_OFFSET_FLAGS("filename", CONF_FLAG_FILE_READABLE | CONF_FLAG_REQUIRED | CONF_FLAG_NOT_EMPTY, proto_load_step_t, filename) },
        { FR_CONF_OFFSET("csv", proto_load_step_t, csv) },
 
        { FR_CONF_OFFSET("max_attributes", proto_load_step_t, max_attributes), .dflt = STRINGIFY(RADIUS_MAX_ATTRIBUTES) } ,
 
        { FR_CONF_OFFSET("start_pps", proto_load_step_t, load.start_pps) },
        { FR_CONF_OFFSET("max_pps", proto_load_step_t, load.max_pps) },
        { FR_CONF_OFFSET("duration", proto_load_step_t, load.duration) },
        { FR_CONF_OFFSET("step", proto_load_step_t, load.step) },
        { FR_CONF_OFFSET("max_backlog", proto_load_step_t, load.milliseconds) },
        { FR_CONF_OFFSET("parallel", proto_load_step_t, load.parallel) },
        { FR_CONF_OFFSET("max_requests", proto_load_step_t, load.max_requests) },
        { FR_CONF_OFFSET("on_complete", proto_load_step_t, on_complete),
          .func = cf_table_parse_int,
          .uctx = &(cf_table_parse_ctx_t){ .table = load_on_complete_table, .len = &load_on_complete_table_len },
          .dflt = "exit" },
 
        CONF_PARSER_TERMINATOR
};
 
 
static ssize_t mod_read(fr_listen_t *li, void **packet_ctx, fr_time_t *recv_time_p, uint8_t *buffer, size_t buffer_len, size_t *leftover)
{
        proto_load_step_t const         *inst = talloc_get_type_abort_const(li->app_io_instance, proto_load_step_t);
        proto_load_step_thread_t        *thread = talloc_get_type_abort(li->thread_instance, proto_load_step_thread_t);
        fr_io_address_t                 *address, **address_p;
 
        if (thread->done) return -1;
 
        /*
         *      Suspend reading on the FD, because we let the timers
         *      take over the load generation.
         */
        if (!thread->suspended) {
                static fr_event_update_t pause[] = {
                        FR_EVENT_SUSPEND(fr_event_io_func_t, read),
                        FR_EVENT_SUSPEND(fr_event_io_func_t, write),
                        { 0 }
                };
 
                if (fr_event_filter_update(thread->el, li->fd, FR_EVENT_FILTER_IO, pause) < 0) {
                        fr_assert(0);
                }
 
                thread->suspended = true;
        }
 
        *leftover = 0;          /* always for load generation */
 
        /*
         *      Where the addresses should go.  This is a special case
         *      for proto_load.
         */
        address_p = (fr_io_address_t **) packet_ctx;
        address = *address_p;
 
        memset(address, 0, sizeof(*address));
        address->socket.inet.src_ipaddr.af = AF_INET;
        address->socket.inet.dst_ipaddr.af = AF_INET;
        address->radclient = inst->client;
 
        *recv_time_p = thread->recv_time;
 
        if (buffer_len < 1) {
                DEBUG2("proto_load_step read buffer is too small for input packet");
                return 0;
        }
 
        buffer[0] = 0;
 
        /*
         *      Print out what we received.
         */
        DEBUG2("proto_load_step - reading packet for %s",
               thread->name);
 
        return 1;
}
 
 
/** The load test is complete - apply on_complete
 *
 */
static void load_complete(proto_load_step_thread_t *thread)
{
        fr_load_stats_t const *stats = fr_load_generator_stats(thread->l);
 
        INFO("Load test for %s complete - sent %d packets, received %d replies",
             thread->name, stats->sent, stats->received);
 
        switch (thread->inst->on_complete) {
        case LOAD_STEP_ON_COMPLETE_EXIT:
                INFO("Signalling the server to exit");
                thread->done = true;
                main_loop_signal_raise(RADIUS_SIGNAL_SELF_TERM);
                break;
 
        case LOAD_STEP_ON_COMPLETE_REPEAT:
                (void) fr_load_generator_stop(thread->l); /* ensure l->ev is gone */
                (void) fr_load_generator_start(thread->l);
                break;
 
        case LOAD_STEP_ON_COMPLETE_STOP:
                thread->done = true;
                break;
 
        /*
         *      The generator never completes when it's told to
         *      continue forever.
         */
        case LOAD_STEP_ON_COMPLETE_CONTINUE:
                fr_assert(0);
                break;
        }
}
 
/** The generator noticed completion itself, without a final reply.
 *
 */
static void mod_load_done(void *uctx)
{
        fr_listen_t                     *li = uctx;
        proto_load_step_thread_t        *thread = talloc_get_type_abort(li->thread_instance, proto_load_step_thread_t);
 
        load_complete(thread);
}
 
static ssize_t mod_write(fr_listen_t *li, UNUSED void *packet_ctx, fr_time_t request_time,
                         UNUSED uint8_t *buffer, size_t buffer_len, UNUSED size_t written)
{
        proto_load_step_thread_t        *thread = talloc_get_type_abort(li->thread_instance, proto_load_step_thread_t);
        fr_load_reply_t state;
 
        /*
         *      @todo - share a stats interface with the parent?  or
         *      put the stats in the listener, so that proto_radius
         *      can update them, too.. <sigh>
         */
        thread->stats.total_responses++;
 
        /*
         *      Tell the load generator subsystem that we have a
         *      reply.  If the reply completes the load test, apply
         *      on_complete.
         */
        state = fr_load_generator_have_reply(thread->l, request_time);
        if (state == FR_LOAD_DONE) {
                load_complete(thread);
        }
 
        return buffer_len;
}
 
 
/** Open a load listener
 *
 */
static int mod_open(fr_listen_t *li)
{
        proto_load_step_t const         *inst = talloc_get_type_abort_const(li->app_io_instance, proto_load_step_t);
        proto_load_step_thread_t        *thread = talloc_get_type_abort(li->thread_instance, proto_load_step_thread_t);
 
        fr_ipaddr_t                     ipaddr;
 
        /*
         *      We never read or write to this file, but we need a
         *      readable FD in order to bootstrap the process.
         */
        li->fd = open(inst->filename, O_RDONLY);
        if (li->fd < 0) {
                cf_log_err(li->cs, "Failed opening %s - %s", inst->filename, fr_syserror(errno));
                return -1;
        }
 
        memset(&ipaddr, 0, sizeof(ipaddr));
        ipaddr.af = AF_INET;
        li->app_io_addr = fr_socket_addr_alloc_inet_src(li, IPPROTO_UDP, 0, &ipaddr, 0);
 
        fr_assert((cf_parent(inst->cs) != NULL) && (cf_parent(cf_parent(inst->cs)) != NULL));   /* listen { ... } */
 
        thread->name = talloc_typed_asprintf(thread, "load_step from filename %s", inst->filename);
        thread->parent = talloc_parent(li);
 
        return 0;
}
 
 
/** Generate traffic.
 *
 */
static int mod_generate(fr_time_t now, void *uctx)
{
        fr_listen_t                     *li = uctx;
        proto_load_step_thread_t        *thread = talloc_get_type_abort(li->thread_instance, proto_load_step_thread_t);
 
        thread->recv_time = now;
 
        /*
         *      Tell the network side to call our read routine.
         */
        fr_network_listen_read(thread->nr, thread->parent);
 
        return 0;
}
 
 
static void write_stats(fr_timer_list_t *tl, fr_time_t now, void *uctx)
{
        proto_load_step_thread_t        *thread = uctx;
        size_t len;
        char buffer[1024];
 
        (void) fr_timer_in(thread, tl, &thread->ev, fr_time_delta_from_sec(1), false, write_stats, thread);
 
        len = fr_load_generator_stats_sprint(thread->l, now, buffer, sizeof(buffer));
        if (write(thread->fd, buffer, len) < 0) {
                DEBUG("Failed writing to %s - %s", thread->inst->csv, fr_syserror(errno));
        }
}
 
 
/** Decode the packet
 *
 */
static int mod_decode(void const *instance, request_t *request, UNUSED uint8_t *const data, UNUSED size_t data_len)
{
        proto_load_step_t const *inst = talloc_get_type_abort_const(instance, proto_load_step_t);
        fr_io_track_t const     *track = talloc_get_type_abort_const(request->async->packet_ctx, fr_io_track_t);
        fr_io_address_t const   *address = track->address;
 
        /*
         *      Hacks for now until we have a lower-level decode routine.
         */
        if (inst->code) request->packet->code = inst->code;
        request->packet->id = fr_rand() & 0xff;
        request->reply->id = request->packet->id;
 
        request->packet->data = talloc_zero_array(request->packet, uint8_t, 1);
        request->packet->data_len = 1;
 
        (void) fr_pair_list_copy(request->request_ctx, &request->request_pairs, &inst->pair_list);
 
        /*
         *      Set the rest of the fields.
         */
        request->client = UNCONST(fr_client_t *, address->radclient);
 
        request->packet->socket = address->socket;
        fr_socket_addr_swap(&request->reply->socket, &address->socket);
 
        REQUEST_VERIFY(request);
 
        return 0;
}
 
/** Set the event list for a new socket
 *
 * @param[in] li the listener
 * @param[in] el the event list
 * @param[in] nr context from the network side
 */
static void mod_event_list_set(fr_listen_t *li, fr_event_list_t *el, void *nr)
{
        proto_load_step_t const       *inst = talloc_get_type_abort_const(li->app_io_instance, proto_load_step_t);
        proto_load_step_thread_t        *thread = talloc_get_type_abort(li->thread_instance, proto_load_step_thread_t);
        size_t len;
        char buffer[256];
 
        thread->el = el;
        thread->nr = nr;
        thread->inst = inst;
        thread->load = inst->load;
 
        thread->l = fr_load_generator_create(thread, el, &thread->load, mod_generate, mod_load_done, li);
        if (!thread->l) return;
 
        (void) fr_load_generator_start(thread->l);
 
        if (!inst->csv) return;
 
        thread->fd = open(inst->csv, O_WRONLY | O_CREAT | O_TRUNC | O_CLOEXEC, 0600);
        if (thread->fd < 0) {
                ERROR("Failed opening %s - %s", inst->csv, fr_syserror(errno));
                return;
        }
 
        (void) fr_timer_in(thread, thread->el->tl, &thread->ev, fr_time_delta_from_sec(1), false, write_stats, thread);
 
        len = fr_load_generator_stats_sprint(thread->l, fr_time(), buffer, sizeof(buffer));
        if (write(thread->fd, buffer, len) < 0) {
                DEBUG("Failed writing to %s - %s", thread->inst->csv, fr_syserror(errno));
        }
}
 
static char const *mod_name(fr_listen_t *li)
{
        proto_load_step_thread_t        *thread = talloc_get_type_abort(li->thread_instance, proto_load_step_thread_t);
 
        return thread->name;
}
 
static int mod_instantiate(module_inst_ctx_t const *mctx)
{
        proto_load_step_t       *inst = talloc_get_type_abort(mctx->mi->data, proto_load_step_t);
        CONF_SECTION            *conf = mctx->mi->conf;
        fr_client_t             *client;
        fr_pair_t               *vp;
        module_instance_t const *mi = mctx->mi;
 
        inst->dict = virtual_server_dict_by_child_ci(cf_section_to_item(conf));
        if (!inst->dict) {
                cf_log_err(conf, "Please define 'namespace' in this virtual server");
                return -1;
        }
        
        fr_pair_list_init(&inst->pair_list);
        MEM(inst->client = client = talloc_zero(inst, fr_client_t));
 
        client->ipaddr.af = AF_INET;
        client->src_ipaddr = client->ipaddr;
 
        client->longname = client->shortname = inst->filename;
        client->secret = talloc_strdup(client, "testing123");
        client->nas_type = talloc_strdup(client, "load");
        client->use_connected = false;
 
        if (inst->filename) {
                FILE *fp;
                bool done = false;
 
                fp = fopen(inst->filename, "r");
                if (!fp) {
                        cf_log_err(conf, "Failed opening %s - %s",
                                   inst->filename, fr_syserror(errno));
                        return -1;
                }
 
                if (fr_pair_list_afrom_file(inst, inst->dict, &inst->pair_list, fp, &done, true) < 0) {
                        cf_log_perr(conf, "Failed reading %s", inst->filename);
                        fclose(fp);
                        return -1;
                }
 
                fclose(fp);
        }
 
        inst->parent = talloc_get_type_abort(mi->parent->data, proto_load_t);
        inst->cs = conf;
 
        vp = fr_pair_find_by_da(&inst->pair_list, NULL, inst->parent->attr_packet_type);
        if (vp) inst->code = vp->vp_uint32;
 
        FR_INTEGER_BOUND_CHECK("start_pps", inst->load.start_pps, >=, 10);
        FR_INTEGER_BOUND_CHECK("start_pps", inst->load.start_pps, <, 400000);
 
        FR_INTEGER_BOUND_CHECK("step", inst->load.step, >=, 1);
        FR_INTEGER_BOUND_CHECK("step", inst->load.step, <, 100000);
 
        if (inst->load.max_pps > 0) FR_INTEGER_BOUND_CHECK("max_pps", inst->load.max_pps, >, inst->load.start_pps);
        FR_INTEGER_BOUND_CHECK("max_pps", inst->load.max_pps, <, 400000);
 
        FR_TIME_DELTA_BOUND_CHECK("duration", inst->load.duration, >=, fr_time_delta_from_sec(1));
        FR_TIME_DELTA_BOUND_CHECK("duration", inst->load.duration, <, fr_time_delta_from_sec(10000));
 
 
        FR_INTEGER_BOUND_CHECK("parallel", inst->load.parallel, >=, 1);
        FR_INTEGER_BOUND_CHECK("parallel", inst->load.parallel, <, 1000);
 
        FR_INTEGER_BOUND_CHECK("max_backlog", inst->load.milliseconds, >=, 1);
        FR_INTEGER_BOUND_CHECK("max_backlog", inst->load.milliseconds, <, 100000);
 
        /*
         *      "continue" means the test never completes, which
         *      contradicts a finite request count.
         */
        if (inst->load.max_requests && (inst->on_complete == LOAD_STEP_ON_COMPLETE_CONTINUE)) {
                cf_log_err(conf, "'max_requests' cannot be used with 'on_complete = continue'");
                return -1;
        }
 
        inst->load.unlimited = (inst->on_complete == LOAD_STEP_ON_COMPLETE_CONTINUE);
 
        return 0;
}
 
static fr_client_t *mod_client_find(fr_listen_t *li, UNUSED fr_ipaddr_t const *ipaddr, UNUSED int ipproto)
{
        proto_load_step_t const       *inst = talloc_get_type_abort_const(li->app_io_instance, proto_load_step_t);
 
        return inst->client;
}
 
fr_app_io_t proto_load_step = {
        .common = {
                .magic                  = MODULE_MAGIC_INIT,
                .name                   = "load_step",
                .config                 = load_listen_config,
                .inst_size              = sizeof(proto_load_step_t),
                .thread_inst_size       = sizeof(proto_load_step_thread_t),
                .instantiate            = mod_instantiate
        },
        .default_message_size   = 4096,
        .track_duplicates       = false,
 
        .open                   = mod_open,
        .read                   = mod_read,
        .write                  = mod_write,
        .event_list_set         = mod_event_list_set,
        .client_find            = mod_client_find,
        .get_name               = mod_name,
 
        .decode                 = mod_decode,
};