(ws) add subcommands: cobra subscribe, and cobra subscribe to statsd bridge

third_party/statsd-client-cpp/.gitignore

@@ -0,0 +1,13 @@
+# Compiled Object files
+*.slo
+*.lo
+*.o
+
+# Compiled Dynamic libraries
+*.so
+*.dylib
+
+# Compiled Static libraries
+*.lai
+*.la
+*.a

third_party/statsd-client-cpp/CMakeLists.txt

@@ -0,0 +1,18 @@
+cmake_minimum_required(VERSION 3.1)
+project(helloCLion)
+
+set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
+
+include_directories(
+    src
+)
+
+add_library(statsdcppclient STATIC src/statsd_client.cpp)
+add_definitions("-fPIC")
+target_link_libraries(statsdcppclient pthread)
+
+add_executable(system_monitor demo/system_monitor.cpp)
+target_link_libraries(system_monitor statsdcppclient)
+
+add_executable(test_client demo/test_client.cpp)
+target_link_libraries(test_client statsdcppclient)

third_party/statsd-client-cpp/LICENSE

@@ -0,0 +1,27 @@
+Copyright (c) 2014, Rex
+All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+* Redistributions of source code must retain the above copyright notice, this
+  list of conditions and the following disclaimer.
+
+* Redistributions in binary form must reproduce the above copyright notice,
+  this list of conditions and the following disclaimer in the documentation
+  and/or other materials provided with the distribution.
+
+* Neither the name of the {organization} nor the names of its
+  contributors may be used to endorse or promote products derived from
+  this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
+FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
+SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
+CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
+OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

third_party/statsd-client-cpp/README.md

@@ -0,0 +1,34 @@
+# a client sdk for StatsD, written in C++
+
+## API
+See [header file](src/statsd_client.h) for more api detail.
+
+** Notice: this client is not thread-safe **
+
+## Demo
+### test\_client
+This simple demo shows how the use this client.
+
+### system\_monitor
+This is a daemon for monitoring a Linux system.
+It'll wake up every minute and monitor the following:
+
+* load
+* cpu
+* free memory
+* free swap (disabled)
+* received bytes
+* transmitted bytes
+* procs
+* uptime
+
+The stats sent to statsd will be in "host.MACAddress" namespace.
+
+Usage:
+
+    system_monitor statsd-host interface-to-monitor
+
+e.g.
+
+    `system_monitor 172.16.42.1 eth0`
+

third_party/statsd-client-cpp/demo/system_monitor.cpp

@@ -0,0 +1,164 @@
+
+#include <sys/types.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <signal.h>
+#include <unistd.h>
+#include <string.h>
+#include <netdb.h>
+#include <sys/sysinfo.h>
+
+#include <sys/ioctl.h>
+#include <netinet/in.h>
+#include <net/if.h>
+#include <string>
+#include <vector>
+
+#include "statsd_client.h"
+
+using namespace std;
+
+static int running = 1;
+
+void sigterm(int sig)
+{
+    running = 0;
+}
+
+string localhost() {
+    struct addrinfo hints, *info, *p;
+    string hostname(1024, '\0');
+    gethostname((char*)hostname.data(), hostname.capacity());
+
+    memset(&hints, 0, sizeof hints);
+    hints.ai_family = AF_UNSPEC; /*either IPV4 or IPV6*/
+    hints.ai_socktype = SOCK_STREAM;
+    hints.ai_flags = AI_CANONNAME;
+
+    if ( getaddrinfo(hostname.c_str(), "http", &hints, &info) == 0) {
+        for(p = info; p != NULL; p = p->ai_next) {
+            hostname = p->ai_canonname;
+        }
+        freeaddrinfo(info);
+    }
+
+    string::size_type pos = hostname.find(".");
+    while ( pos != string::npos )
+    {
+        hostname[pos] = '_';
+        pos = hostname.find(".", pos);
+    }
+    return hostname;
+}
+
+vector<string>& StringSplitTrim(const string& sData,
+        const string& sDelim, vector<string>& vItems)
+{
+    vItems.clear();
+
+    string::size_type bpos = 0;
+    string::size_type epos = 0;
+    string::size_type nlen = sDelim.size();
+
+    while(sData.substr(epos,nlen) == sDelim)
+    {
+        epos += nlen;
+    }
+    bpos = epos;
+
+    while ((epos=sData.find(sDelim, epos)) != string::npos)
+    {
+        vItems.push_back(sData.substr(bpos, epos-bpos));
+        epos += nlen;
+        while(sData.substr(epos,nlen) == sDelim)
+        {
+            epos += nlen;
+        }
+        bpos = epos;
+    }
+
+    if(bpos != sData.size())
+    {
+        vItems.push_back(sData.substr(bpos, sData.size()-bpos));
+    }
+    return vItems;
+}
+
+int main(int argc, char *argv[])
+{
+    FILE *net, *stat;
+    struct sysinfo si;
+    char line[256];
+    unsigned int user, nice, sys, idle, total, busy, old_total=0, old_busy=0;
+
+    if (argc != 3) {
+        printf( "Usage: %s host port\n"
+                "Eg:    %s 127.0.0.1 8125\n",
+                argv[0], argv[0]);
+        exit(1);
+    }
+
+    signal(SIGHUP, SIG_IGN);
+    signal(SIGPIPE, SIG_IGN);
+    signal(SIGCHLD, SIG_IGN); /* will save one syscall per sleep */
+    signal(SIGTERM, sigterm);
+
+    if ( (net = fopen("/proc/net/dev", "r")) == NULL) {
+        perror("fopen");
+        exit(-1);
+    }
+
+    if ( (stat = fopen("/proc/stat", "r")) == NULL) {
+        perror("fopen");
+        exit(-1);
+    }
+
+    string ns = string("host.") + localhost().c_str() + ".";
+    statsd::StatsdClient client(argv[1], atoi(argv[2]), ns);
+
+    daemon(0,0);
+    printf("running in background.\n");
+
+    while(running) {
+        rewind(net);
+        vector<string> items;
+        while(!feof(net)) {
+            fgets(line, sizeof(line), net);
+            StringSplitTrim(line, " ", items);
+
+            if ( items.size() < 17 ) continue;
+            if ( items[0].find(":") == string::npos ) continue;
+            if ( items[1] == "0" and items[9] == "0" ) continue;
+
+            string netface = "network."+items[0].erase( items[0].find(":") );
+            client.count( netface+".receive.bytes", atoll(items[1].c_str()) );
+            client.count( netface+".receive.packets", atoll(items[2].c_str()) );
+            client.count( netface+".transmit.bytes", atoll(items[9].c_str()) );
+            client.count( netface+".transmit.packets", atoll(items[10].c_str()) );
+        }
+
+        sysinfo(&si);
+        client.gauge("system.load", 100*si.loads[0]/0x10000);
+        client.gauge("system.freemem", si.freeram/1024);
+        client.gauge("system.procs", si.procs);
+        client.count("system.uptime", si.uptime);
+
+        /* rewind doesn't do the trick for /proc/stat */
+        freopen("/proc/stat", "r", stat);
+        fgets(line, sizeof(line), stat);
+        sscanf(line, "cpu  %u %u %u %u", &user, &nice, &sys, &idle);
+        total = user + sys + idle;
+        busy = user + sys;
+
+        client.send("system.cpu", 100 * (busy - old_busy)/(total - old_total), "g", 1.0);
+
+        old_total = total;
+        old_busy = busy;
+        sleep(6);
+    }
+
+    fclose(net);
+    fclose(stat);
+
+    exit(0);
+}

third_party/statsd-client-cpp/demo/test_client.cpp

@@ -0,0 +1,28 @@
+
+#include <iostream>
+#include <unistd.h>
+#include "statsd_client.h"
+
+int main(void)
+{
+    std::cout << "running..." << std::endl;
+
+    statsd::StatsdClient client;
+    statsd::StatsdClient client2("127.0.0.1", 8125, "myproject.abx.", true);
+
+    client.count("count1", 123, 1.0);
+    client.count("count2", 125, 1.0);
+    client.gauge("speed", 10);
+    int i;
+    for (i=0; i<1000; i++)
+        client2.timing("request", i);
+    sleep(1);
+    client.inc("count1", 1.0);
+    client2.dec("count2", 1.0);
+//    for(i=0; i<1000; i++) {
+//        client2.count("count3", i, 0.8);
+//    }
+
+    std::cout << "done" << std::endl;
+    return 0;
+}

third_party/statsd-client-cpp/src/statsd_client.cpp

@@ -0,0 +1,246 @@
+#include <math.h>
+#include <netdb.h>
+#include <time.h>
+#include <unistd.h>
+#include <stdlib.h>
+#include <string.h>
+#include <stdio.h>
+#include <netinet/in.h>
+#include "statsd_client.h"
+
+using namespace std;
+namespace statsd {
+
+inline bool fequal(float a, float b)
+{
+    const float epsilon = 0.0001;
+    return ( fabs(a - b) < epsilon );
+}
+
+inline bool should_send(float sample_rate)
+{
+    if ( fequal(sample_rate, 1.0) )
+    {
+        return true;
+    }
+
+    float p = ((float)random() / RAND_MAX);
+    return sample_rate > p;
+}
+
+struct _StatsdClientData {
+    int     sock;
+    struct  sockaddr_in server;
+
+    string  ns;
+    string  host;
+    short   port;
+    bool    init;
+
+    char    errmsg[1024];
+};
+
+StatsdClient::StatsdClient(const string& host,
+                           int port,
+                           const string& ns,
+                           const bool batching)
+: batching_(batching), exit_(false)
+{
+    d = new _StatsdClientData;
+    d->sock = -1;
+    config(host, port, ns);
+    srandom(time(NULL));
+
+    if (batching_) {
+        pthread_mutex_init(&batching_mutex_lock_, nullptr);
+        batching_thread_ = std::thread([this] {
+          while (!exit_) {
+              std::deque<std::string> staged_message_queue;
+
+              pthread_mutex_lock(&batching_mutex_lock_);
+              batching_message_queue_.swap(staged_message_queue);
+              pthread_mutex_unlock(&batching_mutex_lock_);
+
+              while(!staged_message_queue.empty()) {
+                  send_to_daemon(staged_message_queue.front());
+                  staged_message_queue.pop_front();
+              }
+
+              std::this_thread::sleep_for(std::chrono::milliseconds(1000));
+          }
+        });
+    }
+}
+
+StatsdClient::~StatsdClient()
+{
+    if (batching_) {
+        exit_ = true;
+        batching_thread_.join();
+        pthread_mutex_destroy(&batching_mutex_lock_);
+    }
+
+
+    // close socket
+    if (d->sock >= 0) {
+        close(d->sock);
+        d->sock = -1;
+        delete d;
+        d = NULL;
+    }
+}
+
+void StatsdClient::config(const string& host, int port, const string& ns)
+{
+    d->ns = ns;
+    d->host = host;
+    d->port = port;
+    d->init = false;
+    if ( d->sock >= 0 ) {
+        close(d->sock);
+    }
+    d->sock = -1;
+}
+
+int StatsdClient::init()
+{
+    if ( d->init ) return 0;
+
+    d->sock = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP);
+    if ( d->sock == -1 ) {
+        snprintf(d->errmsg, sizeof(d->errmsg), "could not create socket, err=%m");
+        return -1;
+    }
+
+    memset(&d->server, 0, sizeof(d->server));
+    d->server.sin_family = AF_INET;
+    d->server.sin_port = htons(d->port);
+
+    int ret = inet_aton(d->host.c_str(), &d->server.sin_addr);
+    if ( ret == 0 )
+    {
+        // host must be a domain, get it from internet
+        struct addrinfo hints, *result = NULL;
+        memset(&hints, 0, sizeof(hints));
+        hints.ai_family = AF_INET;
+        hints.ai_socktype = SOCK_DGRAM;
+
+        ret = getaddrinfo(d->host.c_str(), NULL, &hints, &result);
+        if ( ret ) {
+            close(d->sock);
+            d->sock = -1;
+            snprintf(d->errmsg, sizeof(d->errmsg),
+                     "getaddrinfo fail, error=%d, msg=%s", ret, gai_strerror(ret) );
+            return -2;
+        }
+        struct sockaddr_in* host_addr = (struct sockaddr_in*)result->ai_addr;
+        memcpy(&d->server.sin_addr, &host_addr->sin_addr, sizeof(struct in_addr));
+        freeaddrinfo(result);
+    }
+
+    d->init = true;
+    return 0;
+}
+
+/* will change the original string */
+void StatsdClient::cleanup(string& key)
+{
+    size_t pos = key.find_first_of(":|@");
+    while ( pos != string::npos )
+    {
+        key[pos] = '_';
+        pos = key.find_first_of(":|@");
+    }
+}
+
+int StatsdClient::dec(const string& key, float sample_rate)
+{
+    return count(key, -1, sample_rate);
+}
+
+int StatsdClient::inc(const string& key, float sample_rate)
+{
+    return count(key, 1, sample_rate);
+}
+
+int StatsdClient::count(const string& key, size_t value, float sample_rate)
+{
+    return send(key, value, "c", sample_rate);
+}
+
+int StatsdClient::gauge(const string& key, size_t value, float sample_rate)
+{
+    return send(key, value, "g", sample_rate);
+}
+
+int StatsdClient::timing(const string& key, size_t ms, float sample_rate)
+{
+    return send(key, ms, "ms", sample_rate);
+}
+
+int StatsdClient::send(string key, size_t value, const string &type, float sample_rate)
+{
+    if (!should_send(sample_rate)) {
+        return 0;
+    }
+
+    cleanup(key);
+
+    char buf[256];
+    if ( fequal( sample_rate, 1.0 ) )
+    {
+        snprintf(buf, sizeof(buf), "%s%s:%zd|%s",
+                 d->ns.c_str(), key.c_str(), value, type.c_str());
+    }
+    else
+    {
+        snprintf(buf, sizeof(buf), "%s%s:%zd|%s|@%.2f",
+                 d->ns.c_str(), key.c_str(), value, type.c_str(), sample_rate);
+    }
+
+    return send(buf);
+}
+
+int StatsdClient::send(const string &message)
+{
+    if (batching_) {
+        pthread_mutex_lock(&batching_mutex_lock_);
+        if (batching_message_queue_.empty() ||
+            batching_message_queue_.back().length() > max_batching_size) {
+            batching_message_queue_.push_back(message);
+        } else {
+            (*batching_message_queue_.rbegin()).append("\n").append(message);
+        }
+        pthread_mutex_unlock(&batching_mutex_lock_);
+
+        return 0;
+    } else {
+        return send_to_daemon(message);
+    }
+}
+
+
+int StatsdClient::send_to_daemon(const string &message) {
+    std::cout << "send_to_daemon: " << message.length() << " B" << std::endl;
+    int ret = init();
+    if ( ret )
+    {
+        return ret;
+    }
+    ret = sendto(d->sock, message.data(), message.size(), 0, (struct sockaddr *) &d->server, sizeof(d->server));
+    if ( ret == -1) {
+        snprintf(d->errmsg, sizeof(d->errmsg),
+                 "sendto server fail, host=%s:%d, err=%m", d->host.c_str(), d->port);
+        return -1;
+    }
+
+    return 0;
+}
+
+const char* StatsdClient::errmsg()
+{
+    return d->errmsg;
+}
+
+}
+

third_party/statsd-client-cpp/src/statsd_client.h

@@ -0,0 +1,66 @@
+
+#ifndef STATSD_CLIENT_H
+#define STATSD_CLIENT_H
+
+#include <sys/types.h>
+#include <arpa/inet.h>
+#include <sys/socket.h>
+#include <pthread.h>
+#include <string>
+#include <thread>
+#include <deque>
+#include <iostream>
+
+namespace statsd {
+
+struct _StatsdClientData;
+
+class StatsdClient {
+public:
+    StatsdClient(const std::string& host="127.0.0.1", int port=8125, const std::string& ns = "", const bool batching = false);
+    ~StatsdClient();
+
+public:
+    // you can config at anytime; client will use new address (useful for Singleton)
+    void config(const std::string& host, int port, const std::string& ns = "");
+    const char* errmsg();
+    int send_to_daemon(const std::string &);
+
+public:
+    int inc(const std::string& key, float sample_rate = 1.0);
+    int dec(const std::string& key, float sample_rate = 1.0);
+    int count(const std::string& key, size_t value, float sample_rate = 1.0);
+    int gauge(const std::string& key, size_t value, float sample_rate = 1.0);
+    int timing(const std::string& key, size_t ms, float sample_rate = 1.0);
+
+public:
+    /**
+     * (Low Level Api) manually send a message
+     * which might be composed of several lines.
+     */
+    int send(const std::string& message);
+
+    /* (Low Level Api) manually send a message
+     * type = "c", "g" or "ms"
+     */
+    int send(std::string key, size_t value,
+             const std::string& type, float sample_rate);
+
+protected:
+    int init();
+    void cleanup(std::string& key);
+
+protected:
+    struct _StatsdClientData* d;
+
+    bool batching_;
+    bool exit_;
+    pthread_mutex_t batching_mutex_lock_;
+    std::thread batching_thread_;
+    std::deque<std::string> batching_message_queue_;
+    const uint64_t max_batching_size = 32768;
+};
+
+}; // end namespace
+
+#endif