soa/service/carbon_connector.h

/* carbon_connector.h                                            -*- C++ -*-
   Jeremy Barnes, 3 August 2011
   Copyright (c) 2011 Datacratic.  All rights reserved.

   Class to accumulate operational stats and connect to carbon (directly).
*/

#pragma once

#include "soa/service/stat_aggregator.h"
#include "soa/service/stats_events.h"
#include "ace/INET_Addr.h"
#include "jml/stats/distribution.h"
#include "soa/types/date.h"
#include <unordered_map>
#include <map>
#include <memory>
#include <thread>
#include <condition_variable>
#include <boost/thread.hpp>


namespace Datacratic {


/*****************************************************************************/
/* MULTI AGGREGATOR                                                          */
/*****************************************************************************/

struct MultiAggregator {
    MultiAggregator();

    typedef std::function<void (const std::vector<StatReading> &)>
        OutputFn;

    MultiAggregator(const std::string & path,
                    const OutputFn & output = OutputFn(),
                    double dumpInterval = 1.0,
                    std::function<void ()> onStop
                        = std::function<void ()>());
    
    ~MultiAggregator();

    void open(const std::string & path,
              const OutputFn & output = OutputFn(),
              double dumpInterval = 1.0,
              std::function<void ()> onStop
                  = std::function<void ()>());

    OutputFn outputFn;

    virtual void doStat(const std::vector<StatReading> & value) const;

    void record(const std::string & stat,
                EventType type = ET_COUNT,
                float value = 1.0);

    void recordLevel(const std::string & stat,
                     float level);

    void recordOccurrence(const std::string & stat);

    void recordQuantity(const std::string & stat,
                        float quantity);
    
    void recordOutcome(const std::string & stat,
                       float measure);

    void dumpSync(std::ostream & stream) const;

    void dump();

    void stop();
    
protected:
    // Prefix to add to each stat to put it in its namespace
    std::string prefix;

    // Function to call when it's stopped/shutdown
    std::function<void ()> onStop;

    // Functions to implement the shutdown
    std::function<void ()> onPreShutdown, onPostShutdown;

private:
    // This map can only have things removed from it, never added to
    typedef std::map<std::string, std::shared_ptr<StatAggregator> > Stats;
    Stats stats;

    // R/W mutex for reading/writing stats.  Read to look up, write to
    // add a new stat.
    typedef std::mutex Lock;
    mutable Lock lock;

    typedef std::unordered_map<std::string, Stats::iterator> LookupCache;

    // Cache of lookups for each thread to avoid needing to acquire a lock
    // very much.
    boost::thread_specific_ptr<LookupCache> lookupCache;

    void runDumpingThread();

    void shutdown();

    StatAggregator & getAggregator(const std::string & stat,
                                   StatAggregator * (*createFn) ());
    
    std::unique_ptr<std::thread> dumpingThread;

    std::condition_variable cond;  // to wake up dumping thread
    std::mutex m;
    bool doShutdown;                 // thread woken up to shutdown
    bool doDump;                     // thread woken up to dump

    double dumpInterval;
};


/*****************************************************************************/
/* CARBON CONNECTOR                                                        */
/*****************************************************************************/

struct CarbonConnector : public MultiAggregator {
    CarbonConnector();

    CarbonConnector(const std::string & carbonAddr,
                    const std::string & path,
                    std::function<void ()> onStop
                        = std::function<void ()>());

    CarbonConnector(const std::vector<std::string> & carbonAddrs,
                    const std::string & path,
                    std::function<void ()> onStop
                        = std::function<void ()>());

    ~CarbonConnector();

    void open(const std::string & carbonAddr,
              const std::string & path,
              std::function<void ()> onStop
                  = std::function<void ()>());

    void open(const std::vector<std::string> & carbonAddrs,
              const std::string & path,
              std::function<void ()> onStop
                  = std::function<void ()>());

    virtual void doStat(const std::vector<StatReading> & value) const;

private:
    // Do our own internal shutdown
    void doShutdown();

    struct Connection {

        Connection(const std::string & addr)
            : fd(-1), addr(addr), shutdown(0)
        {
        }

        ~Connection();

        std::string connect();
        void reconnect();

        void send(const std::string & message);

        void close();

        void runReconnectionThread();

        // Networky things
        int fd;
        std::string addr;
        ACE_INET_Addr ip;

        // Reconnection thread if we are trying to reconnect
        std::shared_ptr<std::thread> reconnectionThread;
        bool reconnectionThreadActive;
        bool reconnectionThreadJoinable;
        int shutdown;
    };

    std::vector<std::shared_ptr<Connection> > connections;
};


} // namespace Datacratic