GDP

/* vim: set ai sw=4 sts=4 ts=4 :*/

/*

**        ----- BEGIN LICENSE BLOCK -----

**        MQTT gateway to the Global Data Plane

**        From the Ubiquitous Swarm Lab, 490 Cory Hall, U.C. Berkeley.

**

**        Copyright (c) 2015-2017, Regents of the University of California.

**        All rights reserved.

**

**        Permission is hereby granted, without written agreement and without

**        license or royalty fees, to use, copy, modify, and distribute this

**        software and its documentation for any purpose, provided that the above

**        copyright notice and the following two paragraphs appear in all copies

**        of this software.

**

**        IN NO EVENT SHALL REGENTS BE LIABLE TO ANY PARTY FOR DIRECT, INDIRECT,

**        SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING LOST

**        PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS DOCUMENTATION,

**        EVEN IF REGENTS HAS BEEN ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

**

**        REGENTS SPECIFICALLY DISCLAIMS ANY WARRANTIES, INCLUDING, BUT NOT

**        LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS

**        FOR A PARTICULAR PURPOSE. THE SOFTWARE AND ACCOMPANYING DOCUMENTATION,

**        IF ANY, PROVIDED HEREUNDER IS PROVIDED "AS IS". REGENTS HAS NO

**        OBLIGATION TO PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS,

**        OR MODIFICATIONS.

**        ----- END LICENSE BLOCK -----

*/

#include <gdp/gdp.h>

#include <ep/ep_app.h>

#include <ep/ep_dbg.h>

#include <ep/ep_log.h>

#include <ep/ep_sd.h>

#include <ep/ep_string.h>

#include <mosquitto.h>

#include <errno.h>

#include <getopt.h>

#include <regex.h>

#include <sysexits.h>

#include <sys/stat.h>

static EP_DBG        Dbg = EP_DBG_INIT("mqtt-gdp-gateway", "MQTT to GDP gateway");

bool                SkipMetadata = false;                // exclude topic, qos, and len on output

bool                DropDups = false;                        // drop duplicate records

time_t                HeartbeatTime = -1;                        // include dups every so often

time_t                WarnMessageInterval = 600;        // issue traffic warnings every 10m

const char        *MqttBroker = NULL;                        // host name of broker

int                        MaxRestartBackoff = 30;                // max reconnect backoff in seconds

/*

**  Called by Mosquitto to do logging

*/

void

log_cb(struct mosquitto *mosq,

                                void *udata,

                                int level,

                                const char *str)

{

        const char *level_str = NULL;

        switch (level)

        {

        case MOSQ_LOG_DEBUG:

                //level_str = "debug";

                break;

        case MOSQ_LOG_INFO:

                level_str = "info";

                break;

        case MOSQ_LOG_WARNING:

                level_str = "warning";

                break;

        case MOSQ_LOG_ERR:

                level_str = "error";

                break;

        case MOSQ_LOG_NOTICE:

                level_str = "notice";

                break;

        default:

                fprintf(stderr, "unknown log level %d\n", level);

                level_str = "???";

                break;

        }

        if (level_str != NULL)

                fprintf(stderr, "MOSQ(%s): %s\n", level_str, str);

}

/*

**  GET_TOPIC_INFO --- get information about a given topic

**

**                This implementation is pretty bad.

*/

struct topic_info

{

        const char        *topic_pat;                        // pattern for topic

        const char        *log_xname;                        // printable external log name

        gdp_gin_t        *gin;                                // associated GDP Instance

        char                *oldrec;                        // text of old record

        char                *currec;                        // text of current record

        size_t                recbuflen;                        // size of previous two buffers

        time_t                oldrectime;                        // time of old record

        time_t                warntime;                        // warn if no record written in N seconds

        time_t                lastwarn;                        // time of last warning

};

struct topic_info        **Topics;                // information about topics

int                                        NTopics = 0;        // number of topics allocated

struct topic_info *

get_topic_info(const struct mosquitto_message *msg)

{

        int tno;

        ep_dbg_cprintf(Dbg, 22, "get_topic_info(%s): ", msg->topic);

        for (tno = 0; tno < NTopics; tno++)

        {

                struct topic_info *t = Topics[tno];

                if (t == NULL)

                        continue;

                bool res;

                int istat = mosquitto_topic_matches_sub(t->topic_pat, msg->topic, &res);

                if (istat != 0 || !res)

                        continue;

                // topic matches

                ep_dbg_cprintf(Dbg, 22, "%p (%s)\n", t, t->topic_pat);

                return t;

        }

        // no match

        if (ep_dbg_test(Dbg, 22))

                ep_dbg_printf("not found\n");

        else

                ep_dbg_cprintf(Dbg, 2, "get_topic_info(%s): not found\n", msg->topic);

        return NULL;

}

struct topic_info *

add_topic_info(const char *topic_pat)

{

        struct topic_info *t;

        t = ep_mem_zalloc(sizeof *t);

        t->topic_pat = topic_pat;

        Topics = ep_mem_realloc(Topics, (NTopics + 1) * sizeof t);

        Topics[NTopics++] = t;

        ep_dbg_cprintf(Dbg, 3, "add_topic_info(%s): %p\n", topic_pat, t);

        return t;

}

int

subscribe_to_all_topics(struct mosquitto *mosq, int qos)

{

        int tno;

        for (tno = 0; tno < NTopics; tno++)

        {

                int istat;

                int mid;                                // message id --- do we need this?

                struct topic_info *t = Topics[tno];

                if (t == NULL)

                        continue;

                ep_dbg_cprintf(Dbg, 5, "subscribe_to_all_topics(%s)\n", t->topic_pat);

                istat = mosquitto_subscribe(mosq, &mid, t->topic_pat, qos);

                if (istat != 0)

                        return istat;

        }

        return 0;

}

/*

**  Determine if a payload is a duplicate of the previous record,

**  i.e. if the record payload has changed in an interesting way.

**

**                Not just strcmp because we treat dates as being identical.

*/

bool

record_is_dup(const char *newrec, struct topic_info *t)

{

        int istat;

        size_t newreclen = strlen(newrec) + 1;

        struct timeval currectime;

        static bool regex_compiled = false;

        static regex_t datepat;

        // regular expression to match ISO 8601 dates (a subset thereof)

        const char *textdatepat =

                "([0-9]{4}-[0-9]{2}-[0-9]{2}T[0-9]{2}:[0-9]{2}(:[0-9]{2}(\\.[0-9]*)?)?)Z";

        regmatch_t pmatch[1];

        // compile the regular expression if not already done

        if (!regex_compiled)

        {

                istat = regcomp(&datepat, textdatepat, REG_EXTENDED);

                if (istat != 0)

                {

                        char ebuf[256];

                        regerror(istat, &datepat, ebuf, sizeof ebuf);

                        ep_app_fatal("Cannot compile regex \"%s\": %s\n",

                                                textdatepat, ebuf);

                }

                regex_compiled = true;

        }

        // copy the new record to the current record so we can step on it

        if (newreclen > t->recbuflen)

        {

                t->currec = ep_mem_realloc(t->currec, newreclen);

                t->oldrec = ep_mem_realloc(t->oldrec, newreclen);

                t->recbuflen = newreclen;

        }

        memcpy(t->currec, newrec, newreclen);

        while ((istat = regexec(&datepat, t->currec, 1, pmatch, 0)) == 0)

        {

                if (pmatch[0].rm_so >= 0)

                {

                        size_t i;

                        ep_dbg_cprintf(Dbg, 81,

                                        "record_is_dup found date from %zd to %zd\n",

                                        (size_t) pmatch[0].rm_so,

                                        (size_t) pmatch[0].rm_eo);

                        for (i = pmatch[0].rm_so; i < pmatch[0].rm_eo; i++)

                                t->currec[i] = '_';                // any fixed character will do

                }

        }

        ep_dbg_cprintf(Dbg, 80,

                        "record_is_dup: normalized record:\n   %s\n",

                        t->currec);

        gettimeofday(&currectime, NULL);

        // dates are normalized; let's see if there is a match

        if (strcmp(t->oldrec, t->currec) == 0)

        {

                // yes, they match, but check time

                if (HeartbeatTime > 0 &&

                        currectime.tv_sec >= t->oldrectime + HeartbeatTime)

                {

                        ep_dbg_cprintf(Dbg, 80,

                                        "dup record, but passed timeout (cur = %ld, old = %ld)\n",

                                        currectime.tv_sec, t->oldrectime + HeartbeatTime);

                }

                else

                {

                        ep_dbg_cprintf(Dbg, 81, "duplicate record\n");

                        return true;

                }

        }

        // not a dup --- save currec for next time

        memcpy(t->oldrec, t->currec, newreclen);

        t->oldrectime = currectime.tv_sec;

        ep_dbg_cprintf(Dbg, 80, "new record\n");

        return false;

}

/*

**  Check all topics to see that some data has been received recently.

**  If not, give a warning so we can poke things.

*/

void

show_idle(struct topic_info *t, time_t now)

{

        if (t->lastwarn + WarnMessageInterval > now)

        {

                // suppress output

                return;

        }

#if HUMAN_OUTPUT

        printf("Topic %s%s%s idle for %ld sec (log %s%s%s)\n",

                        EpChar->lquote, t->topic_pat, EpChar->rquote,

                        now - t->oldrectime,

                        EpChar->lquote, t->log_xname, EpChar->rquote);

#else

        printf("{ \"tag\": \"idle mqtt topic\", "

                        "\"broker\": \"%s\", "

                        "\"topic\": \"%s\", "

                        "\"log\": \"%s\", "

                        "\"idle\": %ld }\n",

                        MqttBroker, t->topic_pat, t->log_xname, now - t->oldrectime);

#endif

        t->lastwarn = now;

}

void

check_traffic(void)

{

        int tno;

        struct timeval now;

        gettimeofday(&now, NULL);

        for (tno = 0; tno < NTopics; tno++)

        {

                struct topic_info *t = Topics[tno];

                if (t->warntime > 0 && t->oldrectime + t->warntime <= now.tv_sec)

                {

                        // nothing has been sent recently

                        show_idle(t, now.tv_sec);

                }

        }

}

/*

**  Called by Mosquitto when a message comes in.

*/

void

message_cb(struct mosquitto *mosq,

                                void *udata,

                                const struct mosquitto_message *msg)

{

        struct topic_info *tinfo;

        const char *payload;

        size_t paylen;

        ep_dbg_cprintf(Dbg, 65, "message_cb: %s @%d => %s\n",

                        msg->topic, msg->qos, (const char *) msg->payload);

        paylen = msg->payloadlen;

        payload = msg->payload;

        if (paylen <= 0)

        {

                payload = "null";

                paylen = strlen(payload) + 1;

        }

        tinfo = get_topic_info(msg);

        if (tinfo == NULL)

        {

                // no place to log this message

                return;

        }

        // if this record is a duplicate, skip it

        if (DropDups && record_is_dup(payload, tinfo))

                        return;

        if (tinfo->gin != NULL)

        {

                EP_STAT estat;

                gdp_datum_t *datum = gdp_datum_new();

                gdp_recno_t oldnrecs;

                if (SkipMetadata)

                {

                        gdp_buf_printf(gdp_datum_getbuf(datum),

                                        "%s\n", payload);

                }

                else

                {

                        gdp_buf_printf(gdp_datum_getbuf(datum),

                                        "{\"topic\":\"%s\", \"qos\":%d, \"len\":%d, \"payload\":%s}\n",

                                        msg->topic, msg->qos, paylen, payload);

                }

                oldnrecs = gdp_gin_getnrecs(tinfo->gin);

                estat = gdp_gin_append(tinfo->gin, datum, NULL);

                if (!EP_STAT_ISOK(estat))

                {

                        ep_log(estat, "cannot log MQTT message for %s "

                                        "(nrecs %" PRIgdp_recno " => %" PRIgdp_recno ")",

                                        msg->topic, oldnrecs, gdp_gin_getnrecs(tinfo->gin));

                }

                gdp_datum_free(datum);

        }

        else

        {

                if (SkipMetadata)

                {

                        printf("%s\n", payload);

                }

                else

                {

                        printf("{\"topic\":\"%s\", \"qos\":%d, \"len\":%zd, \"payload\":%s}\n",

                                        msg->topic, msg->qos, paylen, payload);

                }

        }

}

/*

**  Run Mosquitto loop until failure.

*/

int

run_mosquitto_loop(struct mosquitto *mosq)

{

        int istat;

        do

        {

                // check for MQTT messages (1 second timeout)

                istat = mosquitto_loop(mosq, 1000, 1);

                if (istat == MOSQ_ERR_SUCCESS)

                {

                        // check for stuck topics

                        check_traffic();

                }

        } while (istat == MOSQ_ERR_SUCCESS);

        // some failure, can we recover?

        ep_dbg_cprintf(Dbg, 3, "mosquitto_loop stat %d (%s) errno %d (%s)\n",

                        istat, mosquitto_strerror(istat), errno, strerror(errno));

        return istat;

}

/*

**  Run Mosquitto against a designated broker.

*/

EP_STAT

run_new_mosquitto(

                        const char *broker,

                        int subscr_qos,

                        void *udata)

{

        struct mosquitto *mosq;

        const char *phase;

        int istat = MOSQ_ERR_ERRNO;

        EP_STAT estat = EP_STAT_ABORT;

        // get a mosquitto context

        phase = "setup";

        mosq = mosquitto_new(NULL, true, udata);

        if (mosq == NULL)

                goto fail0;

        // can't just pass in NULL for udata, since mosquitto_new automatically

        // maps NULL to mosq

        mosquitto_user_data_set(mosq, udata);

        // adjust reconnection parameters

        mosquitto_reconnect_delay_set(mosq, 2, 30, true);

        // set up callbacks

        mosquitto_log_callback_set(mosq, &log_cb);

        mosquitto_message_callback_set(mosq, &message_cb);

        // connect to the broker

        phase = "connect";

        {

                // parse the broker information

                int mqtt_port = 1883;

                char *bname = ep_mem_strdup(broker);

                char *p = strrchr(bname, ':');

                if (p != NULL && strchr(p, ']') == NULL)

                {

                        *p++ = '\0';

                        mqtt_port = atol(p);

                }

                ep_dbg_cprintf(Dbg, 1,

                                "mosquitto_connect(%s:%d)\n", bname, mqtt_port);

                istat = mosquitto_connect(mosq, bname, mqtt_port, 60);

                ep_mem_free(bname);

        }

        if (istat != 0)

                goto fail0;

        phase = "subscribe";

        istat = subscribe_to_all_topics(mosq, subscr_qos);

        if (istat != 0)

                goto fail0;

        // we can claim to be started now

        ep_sd_notifyf("READY=1\n");

        phase = "loop";

        istat = run_mosquitto_loop(mosq);

        // recoverable?

        switch (istat)

        {

                case MOSQ_ERR_NO_CONN:

                case MOSQ_ERR_CONN_REFUSED:

                case MOSQ_ERR_CONN_LOST:

                        estat = EP_STAT_ERROR;

                        break;

                case MOSQ_ERR_ERRNO:

                        if (errno != EPROTO)

                                estat = ep_stat_from_errno(errno);

                        break;

        }

fail0:

        fprintf(stderr, "mosquitto error in %s: %s\n",

                        phase, mosquitto_strerror(istat));

        if (EP_STAT_ISABORT(estat))

                ep_sd_notifyf("STOPPING=1\n");

        else

                ep_sd_notifyf("RELOADING=1\n");

        if (mosq != NULL)

        {

                (void) mosquitto_disconnect(mosq);        // ignore possible MOSQ_ERR_NO_CONN

                                                                                        // "not connected" error

                mosquitto_destroy(mosq);

        }

        return estat;

}

gdp_gin_t *

open_signed_log(const char *gdp_log_name, const char *signing_key_file)

{

        EP_STAT estat;

        gdp_name_t gname;

        gdp_open_info_t *info = NULL;

        gdp_gin_t *gin = NULL;

        // make sure we can parse the log name

        estat = gdp_parse_name(gdp_log_name, gname);

        if (!EP_STAT_ISOK(estat))

        {

                char ebuf[60];

                ep_app_error("cannot parse log name %s%s%s: %s",

                                EpChar->lquote, gdp_log_name, EpChar->rquote,

                                ep_stat_tostr(estat, ebuf, sizeof ebuf));

                goto fail0;

        }

        if (signing_key_file != NULL)

        {

                FILE *fp;

                EP_CRYPTO_KEY *skey;

                struct stat st;

                // set up any open information

                info = gdp_open_info_new();

                if (stat(signing_key_file, &st) == 0 &&

                                (st.st_mode & S_IFMT) == S_IFDIR)

                {

                        size_t sz;

                        char *p;

                        gdp_pname_t pname;

                        // we need the printable name for getting the key file name

                        gdp_printable_name(gname, pname);

                        // find the file in the directory

                        sz = strlen(signing_key_file) + sizeof pname + 6;

                        p = alloca(sz);

                        snprintf(p, sz, "%s/%s.pem", signing_key_file, pname);

                        signing_key_file = p;

                }

                fp = fopen(signing_key_file, "r");

                if (fp == NULL)

                {

                        estat = ep_stat_from_errno(errno);

                        ep_app_error("cannot open signing key file %s, log %s",

                                        signing_key_file, gdp_log_name);

                        goto fail1;

                }

                skey = ep_crypto_key_read_fp(fp, signing_key_file,

                                EP_CRYPTO_KEYFORM_PEM, EP_CRYPTO_F_SECRET);

                if (skey == NULL)

                {

                        estat = ep_stat_from_errno(errno);

                        if (EP_STAT_ISOK(estat))

                                estat = EP_STAT_CRYPTO_FAIL;

                        ep_app_error("cannot read signing key file %s, log %s",

                                        signing_key_file, gdp_log_name);

                        goto fail1;

                }

                estat = gdp_open_info_set_signing_key(info, skey);

fail1:

                if (fp != NULL)

                        fclose(fp);

        }

        EP_STAT_CHECK(estat, goto fail0);

        // open a log with the provided name

        estat = gdp_gin_open(gname, GDP_MODE_AO, info, &gin);

        if (!EP_STAT_ISOK(estat))

        {

                char ebuf[60];

                ep_app_error("cannot open log %s%s%s: %s",

                                EpChar->lquote, gdp_log_name, EpChar->rquote,

                                ep_stat_tostr(estat, ebuf, sizeof ebuf));

                goto fail0;

        }

fail0:

        if (info != NULL)

                gdp_open_info_free(info);

        return gin;

}

void

usage(void)

{

        fprintf(stderr,

                        "Usage: %s [-d] [-D dbgspec] [-G router_addr] [-h]\n"

                        "\t[-K signing_key_file] [-M broker_addr] [-q qos] [-s]\n"

                        "\t[-w idletime] [-W msg_interval] mqtt_topic gdp_log ...\n"

                        "    -d  drop duplicate records\n"

                        "    -D  set debugging flags\n"

                        "    -G  IP host to contact for gdp_router\n"

                        "    -h  record \"heartbeat\" interval, even if dups (seconds)\n"

                        "    -K  key file to use to sign GDP log writes\n"

                        "    -M  IP host to contact for MQTT broker\n"

                        "    -q  MQTT Quality of Service to request\n"

                        "    -s  skip metadata in output (topic, qos, len)\n"

                        "    -w  set warn interval for no topic traffic (seconds)\n"

                        "    -W  only issue warning messages every N seconds\n"

                        "mqtt_topic and gdp_log must be in pairs\n"

                        "",

                        ep_app_getprogname());

        exit(EX_USAGE);

}

int

main(int argc, char **argv)

{

        EP_STAT estat;

        bool show_usage = false;

        int mqtt_qos = 2;

        char *gdp_addr = NULL;

        char *signing_key_file = NULL;

        int opt;

        time_t warntime = 0;

        struct timeval now;

        char ebuf[60];

        while ((opt = getopt(argc, argv, "dD:G:h:K:M:q:sw:W:")) > 0)

        {

                switch (opt)

                {

                case 'd':

                        DropDups = true;

                        break;

                case 'D':

                        ep_dbg_set(optarg);

                        break;

                case 'G':

                        gdp_addr = optarg;

                        break;

                case 'h':

                        HeartbeatTime = atol(optarg);

                        break;

                case 'K':

                        signing_key_file = optarg;

                        break;

                case 'M':

                        MqttBroker = optarg;

                        break;

                case 'q':

                        mqtt_qos = atoi(optarg);

                        break;

                case 's':

                        SkipMetadata = true;

                        break;

                case 'w':

                        warntime = atoi(optarg);

                        break;

                case 'W':

                        WarnMessageInterval = atoi(optarg);

                        break;

                default:

                        show_usage = true;

                        break;

                }

        }

        argc -= optind;

        argv += optind;

        if (show_usage)

                usage();

        if (argc < 2)

        {

                fprintf(stderr, "missing required argument (argc = %d)\n", argc);

                usage();

        }

        // initialize GDP library

        estat = gdp_init(gdp_addr);

        if (!EP_STAT_ISOK(estat))

        {

                ep_app_error("GDP Initialization failed: %s",

                                ep_stat_tostr(estat, ebuf, sizeof ebuf));

                exit(EX_UNAVAILABLE);

        }

        // allow thread to settle to avoid interspersed debug output

        ep_time_nanosleep(INT64_C(100000000));

        // open any logs we may use

        gettimeofday(&now, NULL);

        for (; argc > 1; argv += 2, argc -= 2)

        {

                struct topic_info *t;

                char *mqtt_topic = argv[0];

                char *gdp_log_name = argv[1];

                gdp_gin_t *gin = NULL;

                if (strcmp(gdp_log_name, "-") != 0)

                {

                        gin = open_signed_log(gdp_log_name, signing_key_file);

                        if (gin == NULL)

                        {

                                // message already given

                                show_usage = true;

                                continue;

                        }

                }

                t = add_topic_info(mqtt_topic);

                t->gin = gin;

                t->log_xname = gdp_log_name;

                t->warntime = warntime;

                t->oldrectime = now.tv_sec;

                if (warntime > 0)

                {

                        printf("{ \"tag\": \"initial mqtt topic\", "

                                        "\"broker\": \"%s\", "

                                        "\"topic\": \"%s\", "

                                        "\"log\": \"%s\" } \n",

                                        MqttBroker, t->topic_pat, t->log_xname);

                }

        }

        if (show_usage)

                usage();

        if (MqttBroker == NULL)

        {

                MqttBroker = ep_adm_getstrparam("swarm.mqtt-gdp-gateway.broker",

                                                        "127.0.0.1");

        }

        if (HeartbeatTime < 0)

        {

                HeartbeatTime = ep_adm_getlongparam("swarm.mqtt-gdp-gateway.heartbeat.time",

                                60);

        }

        mosquitto_lib_init();

        int backoff = 0;

        for (;;)

        {

                estat = run_new_mosquitto(MqttBroker, mqtt_qos, NULL);

                ep_log(estat, "run_new_mosquitto %s returned", MqttBroker);

                // if this is an abort condition, die now

                if (EP_STAT_ISABORT(estat))

                        break;

                // try again (with backoff)

                if (backoff > 0)

                {

                        sleep(backoff);

                        backoff *= 2;

                }

                else

                {

                        backoff = 1;

                }

                if (backoff > MaxRestartBackoff)

                        backoff = MaxRestartBackoff;

        }

        exit(EX_SOFTWARE);

}