<commit_before>/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
#include "config.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
#include <assert.h>
#include "memcached.h"
#include "cproxy.h"
#include "work.h"
#include "agent.h"
#include "log.h"
#include "cJSON.h"

// Integration with libconflate.
//
static void update_ptd_config(void *data0, void *data1);
static bool update_str_config(char **curr, char *next, char *descrip);

static bool update_behaviors_config(proxy_behavior **curr,
                                    int  *curr_num,
                                    proxy_behavior  *next,
                                    int   next_num,
                                    char *descrip);

void close_outdated_proxies(proxy_main *m, uint32_t new_config_ver);

char *parse_kvs_servers(char *prefix,
                        char *pool_name,
                        kvpair_t *kvs,
                        char **servers,
                        proxy_behavior_pool *behavior_pool);

char **parse_kvs_behavior(kvpair_t *kvs,
                          char *prefix,
                          char *name,
                          proxy_behavior *behavior);

static char *NULL_BUCKET = "[<NULL_BUCKET>]";

static void cproxy_init_null_bucket(proxy_main *m);

static void cproxy_on_config(void *data0, void *data1);

static void agent_logger(void *userdata,
                         enum conflate_log_level lvl,
                         const char *msg, ...)
{
    (void)userdata;
    (void)lvl;
    (void)msg;
// Issues compiling vfprintf(), so turn off this unused code path for now.
#undef AGENT_LOGGER
#ifdef AGENT_LOGGER
    char *n = NULL;
    bool v = false;

    switch(lvl) {
    case FATAL: n = "FATAL"; v = settings.verbose > 0; break;
    case ERROR: n = "ERROR"; v = settings.verbose > 0; break;
    case WARN:  n = "WARN";  v = settings.verbose > 1; break;
    case INFO:  n = "INFO";  v = settings.verbose > 1; break;
    case DEBUG: n = "DEBUG"; v = settings.verbose > 2; break;
    }
    if (!v) {
        return;
    }

    char fmt[strlen(msg) + 16];
    snprintf(fmt, sizeof(fmt), "%s: %s\n", n, msg);

    va_list ap;
    va_start(ap, msg);
    vfprintf(fmt, ap);
    va_end(ap);
#endif
}

static void init_extensions(void)
{
    conflate_register_mgmt_cb("client_stats", "Retrieve stats from moxi",
                              on_conflate_get_stats);
    conflate_register_mgmt_cb("reset_stats", "Reset moxi stats",
                              on_conflate_reset_stats);
    conflate_register_mgmt_cb("ping_test", "Perform a ping test",
                              on_conflate_ping_test);
}

/** The cfg_str looks like...
 *
 *    apikey=jidname@jhostname%jpassword,config=config,host=host
 *      or...
 *    jidname@jhostname%jpassword,config=config,host=host
 *
 *  Only the apikey is needed, so it can also look like...
 *
 *    jidname@jhostname%jpassword
 *
 *  Or...
 *
 *    http://host:port/default/pools/bucketsStreamingConfig/default
 *    url=http://host:port/default/pools/bucketsStreamingConfig/default
 *    auth=,url=http://host:port/default/pools/bucketsStreamingConfig/default
 *    auth=USER%PSWD,url=http://host:port/default/pools/bucketsStreamingConfig/default
 *    auth=Administrator%password,url=http://host:port/default/pools/bucketsStreamingConfig/default
 */
int cproxy_init_agent(char *cfg_str,
                      proxy_behavior behavior,
                      int nthreads) {
    init_extensions();

    if (cfg_str == NULL) {
        moxi_log_write("ERROR: missing cfg\n");
        if (ml->log_mode != ERRORLOG_STDERR) {
            fprintf(stderr, "ERROR: missing cfg\n");
        }
        exit(EXIT_FAILURE);
    }

    int cfg_len = strlen(cfg_str);
    if (cfg_len <= 0) {
        moxi_log_write("ERROR: empty cfg\n");
        if (ml->log_mode != ERRORLOG_STDERR) {
            fprintf(stderr, "ERROR: empty cfg\n");
        }
        exit(EXIT_FAILURE);
    }

    char *buff;

    if (strncmp(cfg_str, "apikey=", 7) == 0 ||
        strncmp(cfg_str, "auth=", 5) == 0 ||
        strncmp(cfg_str, "url=", 4) == 0) {
        buff = trimstrdup(cfg_str);
    } else {
        buff = calloc(cfg_len + 50, sizeof(char));
        if (buff != NULL) {
            if (strncmp(cfg_str, "http://", 7) == 0) {
                snprintf(buff, cfg_len + 50, "url=%s", cfg_str);
            } else {
                snprintf(buff, cfg_len + 50, "apikey=%s", cfg_str);
            }
        }
        buff = trimstr(buff);
    }

    char *next = buff;

    int rv = 0;

    while (next != NULL) {
        char *jid    = NULL;
        char *jpw    = NULL;
        char *jpwmem = NULL;
        char *dbpath = NULL;
        char *host   = NULL;

        char *cur = trimstr(strsep(&next, ";"));
        while (cur != NULL) {
            char *key_val = trimstr(strsep(&cur, ",\r\n"));
            if (key_val != NULL) {
                char *key = trimstr(strsep(&key_val, "="));
                char *val = trimstr(key_val);

                bool handled = true;

                if (key != NULL &&
                    val != NULL) {
                    if (wordeq(key, "apikey") ||
                        wordeq(key, "auth")) {
                        jid = strsep(&val, "%");
                        jpw = val;
                    } else if (wordeq(key, "config") ||
                               wordeq(key, "dbpath")) {
                        dbpath = val;
                    } else if (wordeq(key, "host") ||
                               wordeq(key, "url")) {
                        host = val;
                    } else {
                        handled = false;
                    }<commit_msg>default to auth from -Z or env vars<commit_after>/* -*- Mode: C; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
#include "config.h"
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <errno.h>
#include <pthread.h>
#include <assert.h>
#include "memcached.h"
#include "cproxy.h"
#include "work.h"
#include "agent.h"
#include "log.h"
#include "cJSON.h"

// Integration with libconflate.
//
static void update_ptd_config(void *data0, void *data1);
static bool update_str_config(char **curr, char *next, char *descrip);

static bool update_behaviors_config(proxy_behavior **curr,
                                    int  *curr_num,
                                    proxy_behavior  *next,
                                    int   next_num,
                                    char *descrip);

void close_outdated_proxies(proxy_main *m, uint32_t new_config_ver);

char *parse_kvs_servers(char *prefix,
                        char *pool_name,
                        kvpair_t *kvs,
                        char **servers,
                        proxy_behavior_pool *behavior_pool);

char **parse_kvs_behavior(kvpair_t *kvs,
                          char *prefix,
                          char *name,
                          proxy_behavior *behavior);

static char *NULL_BUCKET = "[<NULL_BUCKET>]";

static void cproxy_init_null_bucket(proxy_main *m);

static void cproxy_on_config(void *data0, void *data1);

static void agent_logger(void *userdata,
                         enum conflate_log_level lvl,
                         const char *msg, ...)
{
    (void)userdata;
    (void)lvl;
    (void)msg;
// Issues compiling vfprintf(), so turn off this unused code path for now.
#undef AGENT_LOGGER
#ifdef AGENT_LOGGER
    char *n = NULL;
    bool v = false;

    switch(lvl) {
    case FATAL: n = "FATAL"; v = settings.verbose > 0; break;
    case ERROR: n = "ERROR"; v = settings.verbose > 0; break;
    case WARN:  n = "WARN";  v = settings.verbose > 1; break;
    case INFO:  n = "INFO";  v = settings.verbose > 1; break;
    case DEBUG: n = "DEBUG"; v = settings.verbose > 2; break;
    }
    if (!v) {
        return;
    }

    char fmt[strlen(msg) + 16];
    snprintf(fmt, sizeof(fmt), "%s: %s\n", n, msg);

    va_list ap;
    va_start(ap, msg);
    vfprintf(fmt, ap);
    va_end(ap);
#endif
}

static void init_extensions(void)
{
    conflate_register_mgmt_cb("client_stats", "Retrieve stats from moxi",
                              on_conflate_get_stats);
    conflate_register_mgmt_cb("reset_stats", "Reset moxi stats",
                              on_conflate_reset_stats);
    conflate_register_mgmt_cb("ping_test", "Perform a ping test",
                              on_conflate_ping_test);
}

/** The cfg_str looks like...
 *
 *    apikey=jidname@jhostname%jpassword,config=config,host=host
 *      or...
 *    jidname@jhostname%jpassword,config=config,host=host
 *
 *  Only the apikey is needed, so it can also look like...
 *
 *    jidname@jhostname%jpassword
 *
 *  Or...
 *
 *    http://host:port/default/pools/bucketsStreamingConfig/default
 *    url=http://host:port/default/pools/bucketsStreamingConfig/default
 *    auth=,url=http://host:port/default/pools/bucketsStreamingConfig/default
 *    auth=USER%PSWD,url=http://host:port/default/pools/bucketsStreamingConfig/default
 *    auth=Administrator%password,url=http://host:port/default/pools/bucketsStreamingConfig/default
 */
int cproxy_init_agent(char *cfg_str,
                      proxy_behavior behavior,
                      int nthreads) {
    init_extensions();

    if (cfg_str == NULL) {
        moxi_log_write("ERROR: missing cfg\n");
        if (ml->log_mode != ERRORLOG_STDERR) {
            fprintf(stderr, "ERROR: missing cfg\n");
        }
        exit(EXIT_FAILURE);
    }

    int cfg_len = strlen(cfg_str);
    if (cfg_len <= 0) {
        moxi_log_write("ERROR: empty cfg\n");
        if (ml->log_mode != ERRORLOG_STDERR) {
            fprintf(stderr, "ERROR: empty cfg\n");
        }
        exit(EXIT_FAILURE);
    }

    char *buff;

    if (strncmp(cfg_str, "apikey=", 7) == 0 ||
        strncmp(cfg_str, "auth=", 5) == 0 ||
        strncmp(cfg_str, "url=", 4) == 0) {
        buff = trimstrdup(cfg_str);
    } else {
        buff = calloc(cfg_len + 50, sizeof(char));
        if (buff != NULL) {
            if (strncmp(cfg_str, "http://", 7) == 0) {
                snprintf(buff, cfg_len + 50, "url=%s", cfg_str);
            } else {
                snprintf(buff, cfg_len + 50, "apikey=%s", cfg_str);
            }
        }
        buff = trimstr(buff);
    }

    char *next = buff;

    int rv = 0;

    while (next != NULL) {
        char *jid    = behavior.usr;
        char *jpw    = behavior.pwd;
        char *jpwmem = NULL;
        char *dbpath = NULL;
        char *host   = NULL;

        char *cur = trimstr(strsep(&next, ";"));
        while (cur != NULL) {
            char *key_val = trimstr(strsep(&cur, ",\r\n"));
            if (key_val != NULL) {
                char *key = trimstr(strsep(&key_val, "="));
                char *val = trimstr(key_val);

                bool handled = true;

                if (key != NULL &&
                    val != NULL) {
                    if (wordeq(key, "apikey") ||
                        wordeq(key, "auth")) {
                        jid = strsep(&val, "%");
                        jpw = val;
                    } else if (wordeq(key, "config") ||
                               wordeq(key, "dbpath")) {
                        dbpath = val;
                    } else if (wordeq(key, "host") ||
                               wordeq(key, "url")) {
                        host = val;
                    } else {
                        handled = false;
                    }
{'fraction_non_alphanumeric': 0.13063453536754507, 'fraction_numerical': 0.005374479889042996, 'mean_word_length': 1.8022832159339324, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4574697', 'n_tokens_mistral': 3842, 'n_tokens_neox': 3680, 'n_words': 817}