Line data    Source code

       1             : /*
       2             :  * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
       3             :  */
       4             : 
       5             : #ifndef __QEOPSERVERPROXY_H__
       6             : #define __QEOPSERVERPROXY_H__
       7             : 
       8             : #include <boost/function.hpp>
       9             : #include <boost/scoped_ptr.hpp>
      10             : #include <string>
      11             : #include <map>
      12             : #include <vector>
      13             : #include <memory>
      14             : #include <boost/variant.hpp>
      15             : #include <boost/uuid/uuid.hpp>
      16             : #include <boost/shared_ptr.hpp>
      17             : 
      18             : extern "C" {
      19             : #include <base/tdigest.h>
      20             : };
      21             : 
      22             : class EventManager;
      23             : class QueryEngine;
      24             : class QueryResultMetaData;
      25             : class query_result_unit_t;
      26             : class AnalyticsQuery;
      27             : 
      28             : // This class represents the interface between the Query Engine and 
      29             : // the OpServer. It will internally talk to the OpServer using Redis
      30             : 
      31             : class QEOpServerProxy {
      32             : public:
      33             :     static const int nMaxChunks = 16;
      34             :     static const int nMaxRows = 2500000;
      35             : 
      36             :     // Increase max number of threads available by a factor of 4
      37             :     static const int nThreadCountMultFactor = 4;
      38             :     
      39             :     // This is the client-provided function that will be called when the OpServer
      40             :     // issues a query. Besides the QID, the function will get a map ,
      41             :     // indexed by field name.
      42             :     //
      43             :     // The field names are expected to be as follows:
      44             :     //   start_time, end_time, from_table, sorted, select_fields, where_and_over_or
      45             :     // The values are expected to be in the form of JSON-encoded strings.
      46             :     typedef boost::function<bool(void *, std::string qid, std::string startTime,
      47             :             std::map<std::string, std::string>)> QECallbackFn;
      48             : 
      49             :     // In addition to the Redis parameters and the EventManager,
      50             :     // the client should provide a callback function to be called
      51             :     // when the OpServer issues a Query.
      52             :     QEOpServerProxy(EventManager *evm, QueryEngine *qe,
      53             :             std::vector<std::string> redis_ip_ports,
      54             :             const std::string & redis_password,
      55             :             const bool redis_ssl_enable,
      56             :             const std::string & redis_keyfile,
      57             :             const std::string & redis_certfile,
      58             :             const std::string & redis_ca_cert,
      59             :             const std::string &host_ip,
      60             :             int max_chunks = nMaxChunks, int max_rows = nMaxRows);
      61             :     virtual ~QEOpServerProxy();
      62             : 
      63             :     // When the result of a Query is available, the client should
      64             :     // call this function with the QID, error code, and a vector of rows.
      65             :     // The rows are JSON-encoded query output rows.
      66             :     // Do not call QueryResult in the thread of execution of 
      67             :     // QECallbackFn.
      68             :     //
      69             :     // Ownership of the vector is transferred to QEOpServerProxy
      70             :     // after this call; the client should not free it.
      71             :     //
      72             :     // If the error field is non-zero, the vector should be empty
      73             :     // Some useful errors:
      74             :     //    EBADMSG         Bad message (JSON could not be parsed)
      75             :     //    EINVAL          Invalid argument (select/where parameters are invalid)
      76             :     //    ENOENT          No such file or directory (Invalid table name)
      77             :     //    EIO             Input/output error (Cassandra is down)
      78             :     typedef std::map<std::string /* Col Name */,
      79             :                      std::string /* Col Value */> OutRowT;
      80             :     typedef boost::shared_ptr<QueryResultMetaData> MetadataT;
      81             :     typedef std::pair<OutRowT, MetadataT> ResultRowT; 
      82             :     typedef std::vector<ResultRowT> BufferT;
      83             : 
      84             :     typedef boost::variant<boost::blank, std::string, uint64_t, double, boost::uuids::uuid, boost::shared_ptr<TDigest>, boost::shared_ptr<Centroid> > SubVal;
      85             :     enum VarType {
      86             :         BLANK=0,
      87             :         STRING=1,
      88             :         UINT64=2,
      89             :         DOUBLE=3,
      90             :         UUID=4,
      91             :         TDIGEST=5,
      92             :         CENTROID=6,
      93             :         MAP_ELEM=7,
      94             :         LIST=8,
      95             :     };
      96             :     enum AggOper {
      97             :         INVALID = 0,
      98             :         SUM = 1,
      99             :         COUNT = 2,
     100             :         CLASS = 3,
     101             :         MAX = 4,
     102             :         MIN = 5,
     103             :         PERCENTILES = 6,
     104             :         AVG = 7,
     105             :         COUNT_DISTINCT = 8,
     106             :     };
     107             : 
     108             :     // This is a map of aggregations for an output row
     109             :     // The key is the operation type and attribute name
     110             :     // The value is a vector of attribute values (or a single agg'ed value)
     111             :     typedef std::map<std::pair<AggOper,std::string>, SubVal> AggRowT;
     112             : 
     113             :     // The key of the multimap is a vector of sort-by values.
     114             :     // The last element of this vector is a hash of all unique cols
     115             :     typedef std::multimap<std::vector<SubVal>, 
     116             :                 // The first element of this pair is itself a map
     117             :                 // which includes all unique cols
     118             :                 std::pair<std::map<std::string, SubVal>,
     119             :                           // This second element of the pair is a map of aggregations
     120             :                           AggRowT> > OutRowMultimapT;
     121             : 
     122             :     struct QPerfInfo {
     123           0 :         QPerfInfo(uint32_t w, uint32_t s, uint32_t p) :
     124           0 :             chunk_where_time(w), chunk_select_time(s), chunk_postproc_time(p),
     125           0 :             error(0) {}
     126        3790 :         QPerfInfo() : 
     127        3790 :             chunk_where_time(0), chunk_select_time(0), chunk_postproc_time(0),
     128        3790 :             error(0) {}
     129             :         uint32_t chunk_where_time;
     130             :         uint32_t chunk_select_time; 
     131             :         uint32_t chunk_postproc_time;
     132             :         int error; 
     133             :     };
     134             : 
     135             :     void QueryResult(void *, QPerfInfo qperf, std::auto_ptr<BufferT> res,
     136             :             std::auto_ptr<OutRowMultimapT> mres);
     137             :     void QueryResult(void *, QPerfInfo qperf,
     138             :             std::auto_ptr<std::vector<query_result_unit_t> > res);
     139             :     void AddAnalyticsQuery(const std::string&,
     140             :             boost::shared_ptr<AnalyticsQuery>);
     141             : private:
     142             :     EventManager * const evm_;
     143             :     QueryEngine * const qe_;
     144             : 
     145             :     class QEOpServerImpl;
     146             :     boost::scoped_ptr<QEOpServerImpl> impl_;
     147             : 
     148             :     friend class QEOpServerImpl;
     149             : };
     150             : 
     151             : #endif