query.cpp 11.5 KB
Newer Older
Axel Auweter's avatar
Axel Auweter committed
1
2
3
4
5
6
7
8
/*
 * query.cpp
 *
 *  Created on: Feb 19, 2015
 *      Author: Axel Auweter
 */

#include <iostream>
9
10
#include <iomanip>
#include <sstream>
Axel Auweter's avatar
Axel Auweter committed
11
12
13
14
15
#include <list>
#include <string>
#include <algorithm>

#include <cstdlib>
16
#include <cinttypes>
Axel Auweter's avatar
Axel Auweter committed
17

18
19
20
#include <boost/algorithm/string.hpp>

#include "dcdbendian.h"
Axel Auweter's avatar
Axel Auweter committed
21
22
23
24
#include "query.h"
#include "casshelper.h"

/* Lovely spaghetti code coming up next. Be aware... */
25
void DCDBQuery::doQuery(const char* hostname, std::list<std::string> sensors, DCDBTimeStamp start, DCDBTimeStamp end, bool raw, bool localtime)
Axel Auweter's avatar
Axel Auweter committed
26
27
28
29
30
31
32
33
34
35
36
37
{
  /* Connect to db */
  CassCluster *cluster = CassHelper::create_cluster(hostname);
  CassSession* session = cass_session_new();

  if (CassHelper::connect_session(session, cluster) != CASS_OK) {
     cass_session_free(session);
     cass_cluster_free(cluster);
     std::cout << "Cannot connect to Cassandra database." << std::endl;
     exit(EXIT_FAILURE);
  }

38
39
40
41
  /* Print the CSV header */
  std::cout << "Sensor,Time,Value" << std::endl;

  /* Iterate over list of sensors requested by the user */
Axel Auweter's avatar
Axel Auweter committed
42
  for (std::list<std::string>::iterator it = sensors.begin(); it != sensors.end(); it++) {
43
44
45
      /* Lookup the sensor in the published sensors table */
      std::string sensorPattern;
      lookupPublishedSensorPattern(session, *it,sensorPattern);
Axel Auweter's avatar
Axel Auweter committed
46
47

      std::list<SensorId> sidList;
48
      expandSensorPattern(session, sensorPattern, start, end, sidList);
49
50
51
52
53
54
55

#if 0
      for (std::list<SensorId>::iterator sit = sidList.begin(); sit != sidList.end(); sit++) {
          std::cout << std::hex << std::setfill('0') << std::setw(16) << (*sit).raw[0]
                    << " " << std::hex << std::setfill('0') << std::setw(16) << (*sit).raw[1] << std::endl;
      }
#endif
Axel Auweter's avatar
Axel Auweter committed
56

57
      querySensorsCSV(session, *it, sidList, start, end, raw, localtime);
Axel Auweter's avatar
Axel Auweter committed
58
59
60
61
62
63
64
65
  }

  /* Clean up */
  cass_session_free(session);
  cass_cluster_free(cluster);

}

66
void DCDBQuery::lookupPublishedSensorPattern(CassSession* session, std::string name, std::string& pattern)
Axel Auweter's avatar
Axel Auweter committed
67
68
69
70
71
72
{
  CassError rc = CASS_OK;
  CassStatement* statement = nullptr;
  CassFuture* future = nullptr;
  const CassPrepared* prepared = nullptr;

73
  CassString query = cass_string_init("SELECT pattern FROM dcdb_config.publishedsensors WHERE name = ? ;");
Axel Auweter's avatar
Axel Auweter committed
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
  CassString pattern_cstr;

  future = cass_session_prepare(session, query);
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    CassHelper::print_error(future);
  } else {
    prepared = cass_future_get_prepared(future);
  }

  statement = cass_prepared_bind(prepared);
  cass_statement_bind_string_by_name(statement, "name", cass_string_init(name.c_str()));

89
90
  cass_future_free(future);

Axel Auweter's avatar
Axel Auweter committed
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
  future = cass_session_execute(session, statement);
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
      CassHelper::print_error(future);
  } else {
      const CassResult* result = cass_future_get_result(future);
      CassIterator* iterator = cass_iterator_from_result(result);

      if (cass_iterator_next(iterator)) {
          const CassRow* row = cass_iterator_get_row(iterator);
          cass_value_get_string(cass_row_get_column_by_name(row, "pattern"), &pattern_cstr);
          pattern = std::string(pattern_cstr.data, pattern_cstr.length);
      }
      else {
          std::cout << "Unknown sensor: " << name << std::endl;
          exit(EXIT_FAILURE);
      }

      cass_result_free(result);
      cass_iterator_free(iterator);
  }
114
115
116
  cass_future_free(future);
  cass_statement_free(statement);
  cass_prepared_free(prepared);
Axel Auweter's avatar
Axel Auweter committed
117
118
}

119
void DCDBQuery::expandSensorPattern(CassSession* session, std::string sensorPattern, DCDBTimeStamp start, DCDBTimeStamp end, std::list<SensorId>& sensorIds)
Axel Auweter's avatar
Axel Auweter committed
120
121
122
123
{
  /* Clear the list of sensorIds */
  sensorIds.clear();

124
125
  /* Strip all slashes from publishedSensorName */
  sensorPattern.erase(std::remove(sensorPattern.begin(), sensorPattern.end(), '/'), sensorPattern.end());
Axel Auweter's avatar
Axel Auweter committed
126

127
128
129
130
131
132
  /* Calculate lower and upper boundaries for the expansion of the pattern */
  std::string low  = sensorPattern;
  std::string high = sensorPattern;
  if (sensorPattern.find("*") != std::string::npos) {
      low.replace(sensorPattern.find("*"), 1, 33-sensorPattern.length(), '0');
      high.replace(sensorPattern.find("*"), 1, 33-sensorPattern.length(), 'F');
Axel Auweter's avatar
Axel Auweter committed
133
134
  }

135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
  SensorId lowId, highId;
  if (!topicToSid(lowId, low)) {
      std::cout << "Internal error" << std::endl;
      exit(EXIT_FAILURE);
  }
  if (!topicToSid(highId, high)) {
      std::cout << "Internal error" << std::endl;
      exit(EXIT_FAILURE);
  }
  lowId.dsid.rsvd = start.getWeekstamp();
  highId.dsid.rsvd = end.getWeekstamp();

  low = sidConvert(lowId);
  high = sidConvert(highId);

//  std::cout << "Lower boundary for sensor scan: " << std::hex << std::setfill('0') << std::setw(16) << lowId.raw[0] << " " << std::hex << std::setfill('0') << std::setw(16) << lowId.raw[1] << std::endl;
//  std::cout << "Upper boundary for sensor scan: " << std::hex << std::setfill('0') << std::setw(16) << highId.raw[0] << " " << std::hex << std::setfill('0') << std::setw(16) << highId.raw[1] << std::endl;

Axel Auweter's avatar
Axel Auweter committed
153
154
155
156
157
158
  /* Query the database to see which raw sensors actually exist in the interval between low and high */
  CassError rc = CASS_OK;
  CassStatement* statement = nullptr;
  CassFuture* future = nullptr;
  const CassPrepared* prepared = nullptr;

159
  CassString query = cass_string_init("SELECT DISTINCT sid FROM dcdb.sensordata WHERE TOKEN(sid) >= ? and TOKEN(sid) <= ?;");
Axel Auweter's avatar
Axel Auweter committed
160
161
162
163
164
165
166
167
168
169
170
171

  future = cass_session_prepare(session, query);
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    CassHelper::print_error(future);
  } else {
    prepared = cass_future_get_prepared(future);
  }

  statement = cass_prepared_bind(prepared);
172
173
174
175
176
177

  CassBytes lowCb = cass_bytes_init((cass_byte_t*)(low.c_str()), 16);
  CassBytes highCb = cass_bytes_init((cass_byte_t*)(high.c_str()), 16);

  cass_statement_bind_bytes(statement, 0, lowCb);
  cass_statement_bind_bytes(statement, 1, highCb);
Axel Auweter's avatar
Axel Auweter committed
178

179
180
  cass_future_free(future);

Axel Auweter's avatar
Axel Auweter committed
181
182
183
184
185
186
187
188
189
190
  future = cass_session_execute(session, statement);
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
      CassHelper::print_error(future);
  } else {
      const CassResult* result = cass_future_get_result(future);
      CassIterator* iterator = cass_iterator_from_result(result);

191
      while (cass_iterator_next(iterator)) {
Axel Auweter's avatar
Axel Auweter committed
192
          const CassRow* row = cass_iterator_get_row(iterator);
193
194
195
196
197
198
199
          CassBytes res;
          cass_value_get_bytes(cass_row_get_column_by_name(row, "sid"), &res);
          SensorId sensor;
          sensor.raw[0] = Endian::BEToHost(((uint64_t*)res.data)[0]);
          sensor.raw[1] = Endian::BEToHost(((uint64_t*)res.data)[1]);

          /* Check if the sensorId matches the pattern */
200
          if (sidPatternMatch(sensor, sensorPattern)) {
201
202
            sensorIds.push_back(sensor);
          }
Axel Auweter's avatar
Axel Auweter committed
203
204
205
206
207
      }

      cass_result_free(result);
      cass_iterator_free(iterator);
  }
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288

  cass_future_free(future);
  cass_statement_free(statement);
  cass_prepared_free(prepared);
}

/* Ugly copy of DCDBLib's topicToSid - It's really time to clean up the API */
bool DCDBQuery::topicToSid(SensorId& sid, std::string topic)
{
  uint64_t pos = 0;
  const char* buf = topic.c_str();
  sid.raw[0] = 0;
  sid.raw[1] = 0;
  while (*buf && pos < 128) {
      if (*buf >= '0' && *buf <= '9') {
          sid.raw[pos / 64] |= (((uint64_t)(*buf - '0')) << (60-(pos%64)));
          pos += 4;
      }
      else if (*buf >= 'A' && *buf <= 'F') {
          sid.raw[pos / 64] |= (((uint64_t)(*buf - 'A' + 0xa)) << (60-(pos%64)));
          pos += 4;
      }
      else if (*buf >= 'a' && *buf <= 'f') {
          sid.raw[pos / 64] |= (((uint64_t)(*buf - 'a' + 0xa)) << (60-(pos%64)));
          pos += 4;
      }
      buf++;
  }
  return pos == 128;
}

/* Ugly copy of DCDBLib's sidConvert - It's really time to clean up the API */
std::string DCDBQuery::sidConvert(SensorId& sid)
{
    uint64_t ll[2];
    ll[0] = Endian::hostToBE(sid.raw[0]);
    ll[1] = Endian::hostToBE(sid.raw[1]);
    return std::string((char*)ll, 16);
}

bool DCDBQuery::sidPatternMatch(SensorId& sid, std::string pattern)
{
  /* Strip all slashes from pattern */
  pattern.erase(std::remove(pattern.begin(), pattern.end(), '/'), pattern.end());

  /* Convert to lower case */
  boost::algorithm::to_lower(pattern);

  /* Calculate the wildcard length */
  int wl = (pattern.find("*") != std::string::npos) ? 33 - pattern.length() : 0;

  /* Do a character by character comparison */
  int posP = 0;
  int posS = 0;
  while (posS < 32) {
      char p, cs[2];
      uint64_t s;
      p = pattern.c_str()[posP];
      s = sid.raw[posS / 16];
      s >>= (60-((4*posS)%64));
      s &= 0xf;
      snprintf(cs, 2, "%" PRIx64, s);

      if (p == '*') {
          /* Jump over the wildcard */
          posS += wl;
          posP++;
          continue;
      }
      else if ((posS >= 24 && posS <= 27) || (p==cs[0])) {
          posS++;
          posP++;
          continue;
      }
      else {
          return false;
      }
  }
  return true;
}

289
void DCDBQuery::querySensorsCSV(CassSession* session, std::string sensorName, std::list<SensorId>& sidList, DCDBTimeStamp& start, DCDBTimeStamp& end, bool raw, bool localtime)
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
{
  /* Since everything has been mangled until here, we assume parameter safety and do a non-prepared statement... Oh oh... */
  std::stringstream query;
  query << "SELECT * FROM dcdb.sensordata WHERE sid IN (";
  for (std::list<SensorId>::iterator it = sidList.begin(); it != sidList.end(); ++it) {
      if (it != sidList.begin()) {
          query << ", ";
      }
      query << "0x";
      query << std::hex << std::setfill('0') << std::setw(16) << (*it).raw[0]
            << std::hex << std::setfill('0') << std::setw(16) << (*it).raw[1];
  }
  query << ") AND ts >= ";
  query << std::dec << start.getRaw();
  query << " AND ts <= ";
  query << std::dec << end.getRaw();
  query << ";";

  CassString csQuery = cass_string_init(query.str().c_str());

310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
  // HACK = sometimes the query fails for unknown reasons
  int retries_left = 10;
  while (retries_left) {
    CassStatement* statement = cass_statement_new(csQuery, 0);
    CassFuture* future = cass_session_execute(session, statement);

    cass_future_wait(future);

    if (cass_future_error_code(future) == CASS_OK) {
        const CassResult* result = cass_future_get_result(future);
        CassIterator* rows = cass_iterator_from_result(result);

        while (cass_iterator_next(rows)) {
            const CassRow* row = cass_iterator_get_row(rows);

            cass_int64_t ts, value;
            cass_value_get_int64(cass_row_get_column_by_name(row, "ts"), &ts);
            cass_value_get_int64(cass_row_get_column_by_name(row, "value"), &value);

            DCDBTimeStamp t((uint64_t)ts);
            if (localtime) {
                t.convertToLocal();
            }
            if (raw) {
                std::cout << sensorName << "," << std::dec << t.getRaw() << "," << std::dec << value << std::endl;
            }
            else {
                std::cout << sensorName << "," << t.getString() << "," << std::dec << value << std::endl;
            }
        }
        cass_iterator_free(rows);
        cass_statement_free(statement);
        cass_future_free(future);

        break;
    }
    else {
         retries_left--;
	 sleep(1);
         // CassString message = cass_future_error_message(future);
	 // fprintf(stderr, "Error: %.*s\n", (int)message.length, message.data);
    }
    cass_statement_free(statement);
    cass_future_free(future);
  }
  if (retries_left == 0) {
	  fprintf(stderr, "Error reading data. Please try again later.\n");
357
  }
Axel Auweter's avatar
Axel Auweter committed
358
}