soa/service/service_base.h

/* service_base.h                                                  -*- C++ -*-
   Jeremy Barnes, 29 May 2012
   Copyright (c) 2012 Datacratic.  All rights reserved.

*/

#ifndef __service__service_base_h__
#define __service__service_base_h__

#include "port_range_service.h"
#include "soa/service/stats_events.h"
#include "stdarg.h"
#include "jml/compiler/compiler.h"
#include <string>
#include <boost/shared_ptr.hpp>
#include "jml/arch/exception.h"
#include "jml/arch/format.h"
#include "jml/arch/spinlock.h"
#include <map>
#include <mutex>
#include "soa/jsoncpp/json.h"
#include <unordered_map>
#include <mutex>
#include <thread>
#include "jml/utils/exc_assert.h"
#include "jml/utils/unnamed_bool.h"


namespace zmq {
struct context_t;
} // namespace zmq

namespace Datacratic {


class MultiAggregator;
class CarbonConnector;

/*****************************************************************************/
/* EVENT SERVICE                                                             */
/*****************************************************************************/

struct EventService {
    virtual ~EventService()
    {
    }
    
    virtual void onEvent(const std::string & name,
                         const char * event,
                         EventType type,
                         float value) = 0;

    virtual void dump(std::ostream & stream) const
    {
    }
};


/*****************************************************************************/
/* NULL EVENT SERVICE                                                        */
/*****************************************************************************/

struct NullEventService : public EventService {

    NullEventService();
    ~NullEventService();
    
    virtual void onEvent(const std::string & name,
                         const char * event,
                         EventType type,
                         float value);

    virtual void dump(std::ostream & stream) const;

    std::unique_ptr<MultiAggregator> stats;
};


/*****************************************************************************/
/* CARBON EVENT SERVICE                                                      */
/*****************************************************************************/

struct CarbonEventService : public EventService {

    CarbonEventService(std::shared_ptr<CarbonConnector> conn);
    CarbonEventService(const std::string & connection,
                       const std::string & prefix = "");
    CarbonEventService(const std::vector<std::string> & connections,
                       const std::string & prefix = "");

    virtual void onEvent(const std::string & name,
                         const char * event,
                         EventType type,
                         float value);

    std::shared_ptr<CarbonConnector> connector;
};


/*****************************************************************************/
/* CONFIGURATION SERVICE                                                     */
/*****************************************************************************/

struct ConfigurationService {

    virtual ~ConfigurationService()
    {
    }

    enum ChangeType {
        VALUE_CHANGED,   
        DELETED,         
        CREATED,         
        NEW_CHILD        

    };
    
    typedef std::function<void (std::string path,
                                ChangeType changeType)> OnChange;

    struct Watch {
        struct Data {
            Data(const OnChange & onChange)
                : onChange(onChange),
                  watchReferences(1)
            {
            }

            OnChange onChange;
            int watchReferences;  
        };
        
        Watch()
            : data(0)
        {
        }

        Watch(const OnChange & onChange)
            : data(new Data(onChange))
        {
        }

        Watch(const Watch & other)
            : data(other.data)
        {
            if (data)
                ++data->watchReferences;
        }

        Watch(Watch && other)
            : data(other.data)
        {
            other.data = 0;
        }

        void swap(Watch & other)
        {
            using std::swap;
            swap(data, other.data);
        }

        Watch & operator = (const Watch & other)
        {
            Watch newMe(other);
            swap(newMe);
            return *this;
        }

        Watch & operator = (Watch && other)
        {
            Watch newMe(other);
            swap(newMe);
            return *this;
        }

        void init(const OnChange & onChange)
        {
            if (data)
                throw ML::Exception("double initializing watch");
            data.reset(new Data(onChange));
        }

        void trigger(const std::string & path, ChangeType change)
        {
            if (!data)
                throw ML::Exception("triggered unused watch");
            if (!data->watchReferences <= 0)
                return;
            ExcAssert(data);
            data->onChange(path, change);
        }

        void disable()
        {
            if (!data)
                return;
            data->watchReferences = 0;
        }

        std::shared_ptr<Data> * get()
        {
            if (!data)
                return nullptr;
            return new std::shared_ptr<Data>(data);
        }

        ~Watch()
        {
            if (data) {
                __sync_add_and_fetch(&data->watchReferences, -1);
            }
            // TODO: we need to unregister ourselves here so that we don't get a
            // callback to something that doesn't exist
        }

        JML_IMPLEMENT_OPERATOR_BOOL(data.get());

    private:
        std::shared_ptr<Data> data;
    };

    virtual void set(const std::string & key,
                     const Json::Value & value) = 0;

    virtual std::string setUnique(const std::string & key,
                                  const Json::Value & value) = 0;

    virtual Json::Value
    getJson(const std::string & key, Watch watch = Watch()) = 0;
    
    typedef std::function<bool (std::string key, Json::Value value)>
    OnEntry;

    virtual std::vector<std::string>
    getChildren(const std::string & key,
                Watch watch = Watch()) = 0;

    virtual bool forEachEntry(const OnEntry & onEntry,
                              const std::string & startPrefix = "") const = 0;

    virtual void removePath(const std::string & path) = 0;

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

    Json::Value jsonDump() const;

    static std::pair<std::string, std::string>
    splitPath(const std::string & path);
};


/*****************************************************************************/
/* INTERNAL CONFIGURATION SERVICE                                            */
/*****************************************************************************/

struct InternalConfigurationService : public ConfigurationService {

    virtual ~InternalConfigurationService()
    {
    }

    virtual Json::Value getJson(const std::string & value,
                                Watch watch = Watch());
    
    virtual void set(const std::string & key,
                     const Json::Value & value);

    virtual std::string setUnique(const std::string & key,
                                  const Json::Value & value);

    virtual std::vector<std::string>
    getChildren(const std::string & key,
                Watch watch = Watch());

    virtual bool forEachEntry(const OnEntry & onEntry,
                              const std::string & startPrefix = "") const;

    virtual void removePath(const std::string & path);

private:
    struct Entry {
        Entry()
            : hasValue(false)
        {
        }

        bool hasValue;
        Json::Value value;
        Watch watch;
        std::unordered_map<std::string, std::shared_ptr<Entry> > children;
    };

    Entry & createNode(Entry & node, const std::string & key);
    const Entry * getNode(const Entry & node, const std::string & key) const;
    Entry * getNode(Entry & node, const std::string & key);

    Entry root;
    
    typedef std::recursive_mutex Lock;
    typedef std::unique_lock<Lock> Guard;
    mutable Lock lock;
};


/*****************************************************************************/
/* SERVICE PROXIES                                                           */
/*****************************************************************************/

struct ServiceProxies {
    ServiceProxies();

    std::shared_ptr<EventService> events;
    std::shared_ptr<ConfigurationService> config;
    std::shared_ptr<PortRangeService> ports;

    std::shared_ptr<zmq::context_t> zmqContext;

    void logToCarbon(std::shared_ptr<CarbonConnector> conn);
    void logToCarbon(const std::string & carbonConnection,
                     const std::string & prefix = "");
    void logToCarbon(const std::vector<std::string> & carbonConnections,
                     const std::string & prefix = "");

    void useZookeeper(std::string hostname = "localhost:2181",
                      std::string prefix = "CWD");

    void usePortRanges(const std::string& path);
    void usePortRanges(const Json::Value& config);

    std::vector<std::string>
    getServiceClassInstances(std::string const & name,
                             std::string const & protocol = "http");

    std::vector<std::string>
    getEndpointInstances(std::string const & name,
                         std::string const & protocol = "http");

    // Bootstrap the proxies services using a json configuration.
    void bootstrap(const std::string& path);
    void bootstrap(const Json::Value& config);
};


/*****************************************************************************/
/* EVENT RECORDER                                                            */
/*****************************************************************************/

struct EventRecorder {

    EventRecorder(const std::string & eventPrefix,
                  const std::shared_ptr<EventService> & events);

    EventRecorder(const std::string & eventPrefix,
                  const std::shared_ptr<ServiceProxies> & services);


    /*************************************************************************/
    /* EVENT RECORDING                                                       */
    /*************************************************************************/

    void recordEvent(const char * eventName,
                     EventType type = ET_COUNT,
                     float value = 1.0) const
    {
        EventService * es = 0;
        if (events_)
            es = events_.get();
        if (!es && services_)
            es = services_->events.get();
        if (!es)
        {
            std::cerr << "no services configured!!!!" << std::endl;
            return;
        }
        es->onEvent(eventPrefix_, eventName, type, value);
    }

    void recordEventFmt(EventType type,
                        float value,
                        const char * fmt, ...) const JML_FORMAT_STRING(4, 5);

    template<typename... Args>
    void recordHit(const std::string & event, Args... args) const
    {
        return recordEventFmt(ET_COUNT, 1.0, event.c_str(),
                              ML::forwardForPrintf(args)...);
    }

    template<typename... Args>
    JML_ALWAYS_INLINE
    void recordHit(const char * event, Args... args) const
    {
        return recordEventFmt(ET_COUNT, 1.0, event,
                              ML::forwardForPrintf(args)...);
    }

    void recordHit(const char * event) const
    {
        recordEvent(event, ET_COUNT, 1.0);
    }

    void recordHit(const std::string & event) const
    {
        recordEvent(event.c_str(), ET_COUNT, 1.0);
    }

    template<typename... Args>
    void recordCount(float count, const std::string & event, Args... args) const
    {
        return recordEventFmt(ET_COUNT, count, event.c_str(),
                              ML::forwardForPrintf(args)...);
    }
    
    template<typename... Args>
    JML_ALWAYS_INLINE
    void recordCount(float count, const char * event, Args... args) const
    {
        return recordEventFmt(ET_COUNT, count, event,
                              ML::forwardForPrintf(args)...);
    }

    void recordCount(float count, const char * event) const
    {
        recordEvent(event, ET_COUNT, count);
    }

    void recordCount(float count, const std::string & event) const
    {
        recordEvent(event.c_str(), ET_COUNT, count);
    }

    template<typename... Args>
    void recordOutcome(float outcome, const std::string & event, Args... args) const
    {
        return recordEventmt(ET_OUTCOME, outcome, event.c_str(),
                             ML::forwardForPrintf(args)...);
    }
    
    template<typename... Args>
    void recordOutcome(float outcome, const char * event, Args... args) const
    {
        return recordEventFmt(ET_OUTCOME, outcome, event,
                              ML::forwardForPrintf(args)...);
    }

    void recordOutcome(float outcome, const char * event) const
    {
        recordEvent(event, ET_OUTCOME, outcome);
    }

    void recordOutcome(float outcome, const std::string & event) const
    {
        recordEvent(event.c_str(), ET_OUTCOME, outcome);
    }
    
    template<typename... Args>
    void recordLevel(float level, const std::string & event, Args... args) const
    {
        return recordEventmt(ET_LEVEL, level, event.c_str(),
                             ML::forwardForPrintf(args)...);
    }
    
    template<typename... Args>
    void recordLevel(float level, const char * event, Args... args) const
    {
        return recordEventFmt(ET_LEVEL, level, event,
                              ML::forwardForPrintf(args)...);
    }

    void recordLevel(float level, const char * event) const
    {
        recordEvent(event, ET_LEVEL, level);
    }

    void recordLevel(float level, const std::string & event) const
    {
        recordEvent(event.c_str(), ET_LEVEL, level);
    }

protected:
    std::string eventPrefix_;
    std::shared_ptr<EventService> events_;
    std::shared_ptr<ServiceProxies> services_;
};


/*****************************************************************************/
/* SERVICE BASE                                                              */
/*****************************************************************************/

struct ServiceBase: public EventRecorder {
    ServiceBase(const std::string & serviceName,
                std::shared_ptr<ServiceProxies>
                    = std::shared_ptr<ServiceProxies>());

    ServiceBase(const std::string & subServiceName,
                ServiceBase & parent);

    virtual ~ServiceBase();

    void setServices(std::shared_ptr<ServiceProxies> services)
    {
        services_ = services;
    }

    std::shared_ptr<ServiceProxies> getServices() const
    {
        return services_;
    }

    std::string serviceName() const
    {
        return serviceName_;
    }

    /*************************************************************************/
    /* REGISTRATION                                                          */
    /*************************************************************************/

    void registerServiceProvider(const std::string & name,
                                 const std::vector<std::string> & serviceClasses);

    void unregisterServiceProvider(const std::string & name,
                                   const std::vector<std::string> & serviceClasses);


    /*************************************************************************/
    /* ZEROMQ CONTEXT                                                        */
    /*************************************************************************/

    std::shared_ptr<zmq::context_t> getZmqContext() const
    {
        return services_->zmqContext;
    }

    /*************************************************************************/
    /* EXCEPTION LOGGING                                                     */
    /*************************************************************************/

    void logException(std::exception_ptr exc,
                      const std::string & context)
    {
        std::cerr << "error: exception in context " << context << std::endl;
        std::rethrow_exception(exc);
    }


    /*************************************************************************/
    /* STATUS                                                                */
    /*************************************************************************/
    
    virtual Json::Value getServiceStatus() const;

    virtual void addChildServiceStatus(Json::Value & result) const;

protected:
    std::shared_ptr<ServiceProxies> services_;
    std::string serviceName_;
    ServiceBase * parent_;
    std::vector<ServiceBase *> children_;
};


/*****************************************************************************/
/* SUB SERVICE BASE                                                          */
/*****************************************************************************/

struct SubServiceBase : public ServiceBase {
};

} // namespace Datacratic


#endif /* __service__service_base_h__ */