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

#include <iostream>
#include <list>
#include <string>
#include <algorithm>

#include <cstdlib>
14
#include <cinttypes>
Axel Auweter's avatar
Axel Auweter committed
15

16
17
18
#include <boost/algorithm/string.hpp>

#include "dcdbendian.h"
Axel Auweter's avatar
Axel Auweter committed
19
20
21
#include "query.h"
#include "casshelper.h"

Axel Auweter's avatar
Axel Auweter committed
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
void DCDBQuery::setLocalTimeEnabled(bool enable) {
  useLocalTime = enable;
}

bool DCDBQuery::getLocalTimeEnabled() {
  return useLocalTime;
}

void DCDBQuery::setRawOutputEnabled(bool enable) {
  useRawOutput = enable;
}

bool DCDBQuery::getRawOutputEnabled() {
  return useRawOutput;
}

void DCDBQuery::doQuery(const char* hostname, std::list<std::string> sensors, DCDBTimeStamp start, DCDBTimeStamp end)
{
  /* Convert start and end to UTC if they're specified in local time. */
  if (useLocalTime) {
      start.convertFromLocal();
      end.convertFromLocal();
  }

  /* Create a new connection to the database */
  DCDBConnection* connection;
  connection = new DCDBConnection();
  connection->setHostname(hostname);
  if (!connection->connect()) {
      std::cout << "Cannot connect to database." << std::endl;
  }

  /* Initialize the SensorConfig interface */
  SensorConfig sensorConfig(connection);
  SensorDataStore sensorDataStore(connection);

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

  /* Iterate over list of sensors requested by the user */
  for (std::list<std::string>::iterator it = sensors.begin(); it != sensors.end(); it++) {
      /* Lookup the sensor in the published sensors table */
      std::string pattern;
      switch (sensorConfig.getSensorPattern(pattern, *it)) {
      case SC_OK:
        break;
      case SC_INVALIDSESSION:
        std::cout << "Invalid session." << std::endl;
        return;
      case SC_UNKNOWNSENSOR:
        std::cout << "Unknown sensor: " << *it << std::endl;
        return;
      default:
        std::cout << "Unknown error." << std::endl;
        return;
      }

      /* Expand the pattern into a list of existing sensors in the time range */
      std::list<SensorId> sensorIds;
      switch (sensorConfig.getSensorListForPattern(sensorIds, pattern, start, end)) {
      case SC_OK:
        break;
      case SC_INVALIDPATTERN:
        std::cout << "Invalid pattern." << std::endl;
        return;
      default:
        std::cout << "Unknown error." << std::endl;
        return;
      }

      /* Iterate over the expanded list of sensorIds and output the results in CSV format */
      for (std::list<SensorId>::iterator sit = sensorIds.begin(); sit != sensorIds.end(); sit++) {
          std::list <SensorDataStoreReading> readings;
          sensorDataStore.query(readings, *sit, start, end);

          /* Iterate over the readings */
          for (std::list<SensorDataStoreReading>::iterator rit = readings.begin(); rit != readings.end(); rit++) {
              SensorDataStoreReading reading = *rit;

              /* Print the sensor's public name */
              std::cout << *it << ",";

              /* Print the time stamp */
              if (useLocalTime) {
                  reading.timeStamp.convertToLocal();
              }
              if (useRawOutput) {
                  std::cout << reading.timeStamp.getRaw() << ",";
              }
              else {
                  std::cout << reading.timeStamp.getString() << ",";
              }

              /* Print the sensor value */
              std::cout << reading.value << std::endl;
          }
      }
  }

  /*
   * Clean up
   */
  connection->disconnect();
  delete connection;

}

#if 0
Axel Auweter's avatar
Axel Auweter committed
130
/* Lovely spaghetti code coming up next. Be aware... */
131
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
132
133
134
135
136
137
138
139
140
141
142
143
{
  /* 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);
  }

144
145
146
147
  /* 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
148
  for (std::list<std::string>::iterator it = sensors.begin(); it != sensors.end(); it++) {
149
150
151
      /* Lookup the sensor in the published sensors table */
      std::string sensorPattern;
      lookupPublishedSensorPattern(session, *it,sensorPattern);
Axel Auweter's avatar
Axel Auweter committed
152
153

      std::list<SensorId> sidList;
154
      expandSensorPattern(session, sensorPattern, start, end, sidList);
155
156
157
158
159
160
161

#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
162

163
      querySensorsCSV(session, *it, sidList, start, end, raw, localtime);
Axel Auweter's avatar
Axel Auweter committed
164
165
166
167
168
169
170
171
  }

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

}

172
void DCDBQuery::lookupPublishedSensorPattern(CassSession* session, std::string name, std::string& pattern)
Axel Auweter's avatar
Axel Auweter committed
173
174
175
176
177
178
{
  CassError rc = CASS_OK;
  CassStatement* statement = nullptr;
  CassFuture* future = nullptr;
  const CassPrepared* prepared = nullptr;

179
  CassString query = cass_string_init("SELECT pattern FROM dcdb_config.publishedsensors WHERE name = ? ;");
Axel Auweter's avatar
Axel Auweter committed
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
  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()));

195
196
  cass_future_free(future);

Axel Auweter's avatar
Axel Auweter committed
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
  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);
  }
220
221
222
  cass_future_free(future);
  cass_statement_free(statement);
  cass_prepared_free(prepared);
Axel Auweter's avatar
Axel Auweter committed
223
224
}

225
void DCDBQuery::expandSensorPattern(CassSession* session, std::string sensorPattern, DCDBTimeStamp start, DCDBTimeStamp end, std::list<SensorId>& sensorIds)
Axel Auweter's avatar
Axel Auweter committed
226
227
228
229
{
  /* Clear the list of sensorIds */
  sensorIds.clear();

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

233
234
235
236
237
238
  /* 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
239
240
  }

241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
  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
259
260
261
262
263
264
  /* 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;

265
  CassString query = cass_string_init("SELECT DISTINCT sid FROM dcdb.sensordata WHERE TOKEN(sid) >= ? and TOKEN(sid) <= ?;");
Axel Auweter's avatar
Axel Auweter committed
266
267
268
269
270
271
272
273
274
275
276
277

  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);
278
279
280
281
282
283

  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
284

285
286
  cass_future_free(future);

Axel Auweter's avatar
Axel Auweter committed
287
288
289
290
291
292
293
294
295
296
  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);

297
      while (cass_iterator_next(iterator)) {
Axel Auweter's avatar
Axel Auweter committed
298
          const CassRow* row = cass_iterator_get_row(iterator);
299
300
301
302
303
304
305
          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 */
306
          if (sidPatternMatch(sensor, sensorPattern)) {
307
308
            sensorIds.push_back(sensor);
          }
Axel Auweter's avatar
Axel Auweter committed
309
310
311
312
313
      }

      cass_result_free(result);
      cass_iterator_free(iterator);
  }
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
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394

  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;
}

395
void DCDBQuery::querySensorsCSV(CassSession* session, std::string sensorName, std::list<SensorId>& sidList, DCDBTimeStamp& start, DCDBTimeStamp& end, bool raw, bool localtime)
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
{
  /* 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());

416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
  // 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");
463
  }
Axel Auweter's avatar
Axel Auweter committed
464
}
Axel Auweter's avatar
Axel Auweter committed
465
#endif