Line data    Source code

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